Pulsar 入门实战(5)--Java 操作 Pulsar

news/2025/1/12 19:03:05/文章来源:https://www.cnblogs.com/wuyongyin/p/18336575

本文主要介绍使用 Java 来操作 Pulsar，文中所使用到的软件版本：Java 17.0.7(Pulsar 服务使用)、Java 1.8.0_341(客户端使用)、Pulsar 3.3.0、pulsar-client 3.3.0。

1、引入依赖

<dependency><groupId>org.apache.pulsar</groupId><artifactId>pulsar-client</artifactId><version>3.3.0</version>
</dependency>
<dependency><groupId>org.apache.pulsar</groupId><artifactId>pulsar-client-admin</artifactId><version>3.3.0</version>
</dependency>

2、初始化 PulsarClient 和 PulsarAdmin

PulsarClient:

@Before
public void before() throws PulsarClientException {client = PulsarClient.builder().serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650").build();
}

PulsarAdmin:

@Before
public void before() throws PulsarClientException {admin = PulsarAdmin.builder().serviceHttpUrl("http://10.49.196.30:8080,10.49.196.31:8080,10.49.196.32:8080").build();
}

3、消费者

3.1、同步消费

@Test
public void sync() throws PulsarClientException {Consumer<String> consumer = client.newConsumer(Schema.STRING).topic("my-topic") //主题.subscriptionName("my-subscription") //订阅名称.subscriptionType(SubscriptionType.Shared) //订阅模式
            .subscribe();while (true) {Message<String> message = consumer.receive();try {log.info("topicName={},value={}", message.getTopicName(), message.getValue());consumer.acknowledge(message);} catch (Exception e) {consumer.negativeAcknowledge(message);}}
}

3.2、异步消费

public void async() throws InterruptedException {client.newConsumer(Schema.STRING).topic("my-topic2").subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).subscribeAsync().thenAccept(this::receiveAsync);Thread.sleep(1000 * 500);
}private void receiveAsync(Consumer<String> consumer) {consumer.receiveAsync().thenAccept(message -> {try {log.info("messageId={},value={}", message.getMessageId(), message.getValue());consumer.acknowledge(message);} catch (Exception e) {consumer.negativeAcknowledge(message);}receiveAsync(consumer);});
}

3.3、使用 MessageListener

@Test
public void messageListener() throws Exception {PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650").listenerThreads(3).build();MessageListener<String> messageListener = (consumer, msg) -> {try {log.info("messageId={},value={}", msg.getMessageId(), msg.getValue());consumer.acknowledge(msg);} catch (Exception e) {consumer.negativeAcknowledge(msg);}};client.newConsumer(Schema.STRING).topic("my-topic").subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).messageListener(messageListener).subscribe();Thread.sleep(1000 * 500);
}

3.4、重试信主题

/*** 重试信主题* 处理失败的消息会进入重试信主题，达到最大重试次数后进入死信主题*/
@Test
public void retryTopic() throws Exception {Consumer<String> consumer = client.newConsumer(Schema.STRING).topic("my-topic-r").subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).enableRetry(true).deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(5).build()).subscribe();while (true) {Message<String> message = consumer.receive();try {log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());//TODO:业务处理，可能发生异常//throw new RuntimeException();
            consumer.acknowledge(message);} catch (Exception e) {log.error("", e);consumer.reconsumeLater(message, 5L, TimeUnit.SECONDS);}}
}

3.5、死信主题

/*** 死信主题* 由确认超时、负面确认或重试信主题 三种情况消息处理失败后，重试最大次数后消息会进入死信主题*/
@Test
public void deadTopic() throws Exception {Consumer<String> consumer = client.newConsumer(Schema.STRING).topic("my-topic-d").subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(5).initialSubscriptionName("init-sub").build()).subscribe();while (true) {Message<String> message = consumer.receive();try {log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());//TODO:业务处理，可能发生异常//throw new RuntimeException();
            consumer.acknowledge(message);} catch (Exception e) {log.error("", e);consumer.negativeAcknowledge(message);}}
}/*** 订阅死信队列，处理其中的消息*/
@Test
public void consumerDeadTopic() throws Exception {Consumer<String> consumer = client.newConsumer(Schema.STRING).topic("my-topic-d-my-subscription-DLQ").subscriptionName("init-sub").subscriptionType(SubscriptionType.Shared).subscribe();while (true) {Message<String> message = consumer.receive();try {log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());} catch (Exception e) {log.error("", e);}consumer.acknowledge(message);}
}

3.6、完整代码

package com.abc.demo.pulsar;import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.junit.Before;
import org.junit.Test;import java.util.concurrent.TimeUnit;@Slf4j
public class ConsumerCase {private PulsarClient client;@Beforepublic void before() throws PulsarClientException {client = PulsarClient.builder().serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650").build();}@Testpublic void sync() throws PulsarClientException {Consumer<String> consumer = client.newConsumer(Schema.STRING).topic("my-topic") //主题.subscriptionName("my-subscription") //订阅名称.subscriptionType(SubscriptionType.Shared) //订阅模式
                .subscribe();while (true) {Message<String> message = consumer.receive();try {log.info("topicName={},value={}", message.getTopicName(), message.getValue());consumer.acknowledge(message);} catch (Exception e) {consumer.negativeAcknowledge(message);}}}@Testpublic void async() throws InterruptedException {client.newConsumer(Schema.STRING).topic("my-topic2").subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).subscribeAsync().thenAccept(this::receiveAsync);Thread.sleep(1000 * 500);}private void receiveAsync(Consumer<String> consumer) {consumer.receiveAsync().thenAccept(message -> {try {log.info("messageId={},value={}", message.getMessageId(), message.getValue());consumer.acknowledge(message);} catch (Exception e) {consumer.negativeAcknowledge(message);}receiveAsync(consumer);});}@Testpublic void messageListener() throws Exception {PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650").listenerThreads(3).build();MessageListener<String> messageListener = (consumer, msg) -> {try {log.info("messageId={},value={}", msg.getMessageId(), msg.getValue());consumer.acknowledge(msg);} catch (Exception e) {consumer.negativeAcknowledge(msg);}};client.newConsumer(Schema.STRING).topic("my-topic").subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).messageListener(messageListener).subscribe();Thread.sleep(1000 * 500);}/*** 重试信主题* 处理失败的消息会进入重试信主题，达到最大重试次数后进入死信主题*/@Testpublic void retryTopic() throws Exception {Consumer<String> consumer = client.newConsumer(Schema.STRING).topic("my-topic-r").subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).enableRetry(true).deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(5).build()).subscribe();while (true) {Message<String> message = consumer.receive();try {log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());//TODO:业务处理，可能发生异常//throw new RuntimeException();
                consumer.acknowledge(message);} catch (Exception e) {log.error("", e);consumer.reconsumeLater(message, 5L, TimeUnit.SECONDS);}}}/*** 死信主题* 由确认超时、负面确认或重试信主题 三种情况消息处理失败后，重试最大次数后消息会进入死信主题*/@Testpublic void deadTopic() throws Exception {Consumer<String> consumer = client.newConsumer(Schema.STRING).topic("my-topic-d").subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared).deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(5).initialSubscriptionName("init-sub").build()).subscribe();while (true) {Message<String> message = consumer.receive();try {log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());//TODO:业务处理，可能发生异常//throw new RuntimeException();
                consumer.acknowledge(message);} catch (Exception e) {log.error("", e);consumer.negativeAcknowledge(message);}}}/*** 订阅死信队列，处理其中的消息*/@Testpublic void consumerDeadTopic() throws Exception {Consumer<String> consumer = client.newConsumer(Schema.STRING).topic("my-topic-d-my-subscription-DLQ").subscriptionName("init-sub").subscriptionType(SubscriptionType.Shared).subscribe();while (true) {Message<String> message = consumer.receive();try {log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());} catch (Exception e) {log.error("", e);}consumer.acknowledge(message);}}
}
ConsumerCase.java

4、Reader

@Test
public void reader() throws Exception {//MessageId messageId = MessageId.fromByteArray(Base64.getDecoder().decode("CBcQBTAA"));MessageId messageId = MessageId.earliest;Reader reader = client.newReader(Schema.STRING).topic("my-topic").startMessageId(messageId).create();while (true) {Message msg = reader.readNext();log.info("messageId={},messageIdBase64={},value={}", msg.getMessageId(), Base64.getEncoder().encodeToString(msg.getMessageId().toByteArray()), msg.getValue());}
}

完整代码：

package com.abc.demo.pulsar;import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.junit.Before;
import org.junit.Test;import java.util.Base64;@Slf4j
public class ReaderCase {private PulsarClient client;@Beforepublic void before() throws PulsarClientException {client = PulsarClient.builder().serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650").build();}@Testpublic void reader() throws Exception {//MessageId messageId = MessageId.fromByteArray(Base64.getDecoder().decode("CBcQBTAA"));MessageId messageId = MessageId.earliest;Reader reader = client.newReader(Schema.STRING).topic("my-topic").startMessageId(messageId).create();while (true) {Message msg = reader.readNext();log.info("messageId={},messageIdBase64={},value={}", msg.getMessageId(), Base64.getEncoder().encodeToString(msg.getMessageId().toByteArray()), msg.getValue());}}
}
ReaderCase.java

5、生产者

5.1、同步发送

@Test
public void sync() throws PulsarClientException {Producer<String> producer = client.newProducer(Schema.STRING).topic("my-topic-d").create();for (int i = 0; i < 3; i++) {MessageId messageId = producer.send(("hello" + i));log.info("messageId={}", messageId);}producer.close();
}

5.2、异步发送

@Test
public void async() throws InterruptedException {client.newProducer(Schema.STRING).topic("my-topic2").createAsync().thenAccept(producer -> {for (int i = 0; i < 10; i++) {producer.sendAsync("hello" + i).thenAccept(messageId -> {log.info("messageId={}", messageId);});}});Thread.sleep(1000 * 5);
}

5.3、详细设置消息

@Test
public void configMessage() throws PulsarClientException {Producer<byte[]> producer = client.newProducer().topic("my-topic").create();MessageId messageId = producer.newMessage(Schema.STRING).key("my-key") //设置消息key.eventTime(System.currentTimeMillis()) //设置事件事件.sequenceId(123) //设置 sequenceId.deliverAfter(1, TimeUnit.MINUTES) //延迟投递消息.property("my-key", "my-value") //自定义属性.value("content").send();log.info("messageId={}", messageId);producer.close();
}

5.4、完整代码

package com.abc.demo.pulsar;import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;@Slf4j
public class ProducerCase {private PulsarClient client;@Beforepublic void before() throws PulsarClientException {client = PulsarClient.builder()//.serviceUrl("pulsar://10.49.196.33:6650").serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650").build();}@Afterpublic void after() throws PulsarClientException {client.close();}@Testpublic void sync() throws PulsarClientException {Producer<String> producer = client.newProducer(Schema.STRING).topic("my-topic-d").create();for (int i = 0; i < 3; i++) {MessageId messageId = producer.send(("hello" + i));log.info("messageId={}", messageId);}producer.close();}@Testpublic void async() throws InterruptedException {client.newProducer(Schema.STRING).topic("my-topic2").createAsync().thenAccept(producer -> {for (int i = 0; i < 10; i++) {producer.sendAsync("hello" + i).thenAccept(messageId -> {log.info("messageId={}", messageId);});}});Thread.sleep(1000 * 5);}@Testpublic void configMessage() throws PulsarClientException {Producer<byte[]> producer = client.newProducer().topic("my-topic").create();MessageId messageId = producer.newMessage(Schema.STRING).key("my-key") //设置消息key.eventTime(System.currentTimeMillis()) //设置事件事件.sequenceId(123) //设置 sequenceId.deliverAfter(1, TimeUnit.MINUTES) //延迟投递消息.property("my-key", "my-value") //自定义属性.value("content").send();log.info("messageId={}", messageId);producer.close();}
}
ProducerCase.java

6、Admin

6.1、Brokers

6.1.1、列出活动 broker

admin.brokers().getActiveBrokers(clusterName)

6.1.2、列出 broker 的命名空间

admin.brokers().getOwnedNamespaces(cluster,brokerUrl);

6.1.3、获取动态配置名称

admin.brokers().getDynamicConfigurationNames();

6.1.4、更新动态配置

admin.brokers().updateDynamicConfiguration(configName, configValue);

6.1.5、获取已经更新过的动态配置

admin.brokers().getAllDynamicConfigurations();

6.1.6、获取 leader broker

admin.brokers().getLeaderBroker()

6.2、Clusters

6.2.1、获取集群配置信息

admin.clusters().getCluster(clusterName);

6.2.2、获取集群列表

admin.clusters().getClusters();

6.3、Tenant

6.3.1、列出租户

admin.tenants().getTenants();

6.3.2、创建租户

admin.tenants().createTenant(tenantName, tenantInfo);

6.3.3、获取租户配置信息

admin.tenants().getTenantInfo(tenantName);

6.3.4、删除租户

admin.Tenants().deleteTenant(tenantName);

6.3.5、更新租户

admin.tenants().updateTenant(tenantName, tenantInfo);

6.4、Namespaces

6.4.1、创建命名空间

admin.namespaces().createNamespace(namespace);

6.4.2、获取命名空间策略

admin.namespaces().getPolicies(namespace);

6.4.3、列出命名空间

admin.namespaces().getNamespaces(tenant);

6.4.4、删除命名空间

admin.namespaces().deleteNamespace(namespace);

6.5、Topics

6.5.1、查看订阅的待消费消息(不会消费消息)

admin.topics().peekMessages(topic, subName, numMessages);

6.5.2、通过消息 ID 获取消息

admin.topics().getMessageById(topic, ledgerId, entryId);

6.5.3、通过相对于最早或最新消息的位置来查找消息

admin.topics().examineMessage(topic, "latest", 1);

6.5.4、通过时间获取消息的 ID

admin.topics().getMessageIdByTimestamp(topic, timestamp);

6.5.5、跳过特定主题的某个订阅中的若干条未消费消息

admin.topics().skipMessages(topic, subName, numMessages);

6.5.6、跳过特定主题的某个订阅中的所有未消费消息

admin.topics().skipAllMessages(topic, subName);

6.5.7、根据时间重置游标

admin.topics().resetCursor(topic, subName, timestamp);

6.5.8、获取主题所属的 broker

admin.lookups().lookupTopic(topic);

6.5.9、获取分区主题所属的 broker

admin.lookups().lookupPartitionedTopic(topic);

6.5.10、获取主题的订阅信息

admin.topics().getSubscriptions(topic);

6.5.11、获取最新消息的 ID

admin.topics().getLastMessage(topic);

6.5.12、创建非分区主题

admin.topics().createNonPartitionedTopic(topicName);

6.5.13、删除非分区主题

admin.topics().delete(topic);

6.5.14、列出非分区主题

admin.topics().getList(namespace);

6.5.15、获取非分区主题状态

admin.topics().getStats(topic, false /* is precise backlog */);

6.5.16、获取非分区主题内部状态

admin.topics().getInternalStats(topic);

6.5.17、创建分区主题

admin.topics().createPartitionedTopic(topicName, numPartitions);

6.5.18、获取分区主题元数据信息

admin.topics().getPartitionedTopicMetadata(topicName);

6.5.19、更新分区主题的分区数(只能比原来大)

admin.topics().updatePartitionedTopic(topic, numPartitions);

6.5.20、删除分区主题

admin.topics().deletePartitionedTopic(topic);

6.5.21、列出分区主题

admin.topics().getPartitionedTopicList(namespace);

6.5.22、获取分区主题状态

admin.topics().getPartitionedStats(topic, true /* per partition */, false /* is precise backlog */);

6.5.23、获取分区主题内部状态

admin.topics().getPartitionedInternalStats(topic);

6.5.24、创建订阅

admin.topics().createSubscription(topic, subscriptionName, MessageId.latest);

6.5.25、获取订阅

admin.topics().getSubscriptions(topic);

6.5.26、删除订阅

admin.topics().deleteSubscription(topic, subscriptionName);

6.6、完整代码

package com.abc.demo.pulsar;import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;import java.util.Collections;
import java.util.HashSet;@Slf4j
public class AdminCase {private PulsarAdmin admin;@Beforepublic void before() throws PulsarClientException {admin = PulsarAdmin.builder().serviceHttpUrl("http://10.49.196.30:8080,10.49.196.31:8080,10.49.196.32:8080").build();}@Afterpublic void after() {admin.close();}@Testpublic void broker() throws PulsarAdminException {log.info("getActiveBrokers={}", admin.brokers().getActiveBrokers());log.info("getOwnedNamespaces={}", admin.brokers().getOwnedNamespaces("pulsar-cluster-1", "app1:8080"));log.info("getDynamicConfigurationNames={}", admin.brokers().getDynamicConfigurationNames());admin.brokers().updateDynamicConfiguration("allowAutoTopicCreation", "true");log.info("getAllDynamicConfigurations={}", admin.brokers().getAllDynamicConfigurations());log.info("getLeaderBroker={}", admin.brokers().getLeaderBroker());}@Testpublic void clusters() throws PulsarAdminException {log.info("getCluster={}", admin.clusters().getCluster("pulsar-cluster-1"));log.info("getClusters={}", admin.clusters().getClusters());}@Testpublic void tenants() throws PulsarAdminException {log.info("getTenants={}", admin.tenants().getTenants());TenantInfoImpl tenantInfo = new TenantInfoImpl();tenantInfo.setAdminRoles(new HashSet<>());tenantInfo.setAllowedClusters(Collections.singleton("pulsar-cluster-1"));admin.tenants().createTenant("test-tenant", tenantInfo);log.info("getTenantInfo={}", admin.tenants().getTenantInfo("test-tenant"));admin.tenants().updateTenant("test-tenant", tenantInfo);admin.tenants().deleteTenant("test-tenant");}@Testpublic void namespaces() throws PulsarAdminException {admin.namespaces().createNamespace("public/test-ns");log.info("getPolicies={}", admin.namespaces().getPolicies("public/default"));log.info("getNamespaces={}", admin.namespaces().getNamespaces("public"));admin.namespaces().deleteNamespace("public/test-ns");}@Testpublic void topics() throws PulsarAdminException {log.info("peekMessages={}", admin.topics().peekMessages("persistent://public/default/my-topic", "my-subscription", 3));Message<byte[]> message = admin.topics().getMessageById("persistent://public/default/my-topic", 171, 16);log.info("getMessageById={}", new String(message.getData()));message = admin.topics().examineMessage("persistent://public/default/my-topic", "latest", 1);log.info("examineMessage={}", new String(message.getData()));log.info("getMessageIdByTimestamp={}", admin.topics().getMessageIdByTimestamp("persistent://public/default/my-topic", System.currentTimeMillis()));admin.topics().skipMessages("persistent://public/default/my-topic", "my-subscription", 1);admin.topics().skipAllMessages("persistent://public/default/my-topic", "my-subscription");admin.topics().resetCursor("persistent://public/default/my-topic", "my-subscription", System.currentTimeMillis() - 1000 * 60 * 15);log.info("lookupTopic={}", admin.lookups().lookupTopic("persistent://public/default/my-topic"));log.info("lookupPartitionedTopic={}", admin.lookups().lookupPartitionedTopic("persistent://public/default/my-topic2"));log.info("getSubscriptions={}", admin.topics().getSubscriptions("persistent://public/default/my-topic"));log.info("getLastMessageId={}", admin.topics().getLastMessageId("persistent://public/default/my-topic"));admin.topics().createNonPartitionedTopic("persistent://public/default/test-topic");admin.topics().delete("persistent://public/default/test-topic");log.info("getList={}", admin.topics().getList("public/default"));log.info("getStats={}", admin.topics().getStats("persistent://public/default/my-topic", false));log.info("getInternalStats={}", admin.topics().getInternalStats("persistent://public/default/my-topic"));admin.topics().createPartitionedTopic("persistent://public/default/test-topic-p", 2);log.info("getPartitionedTopicMetadata={}", admin.topics().getPartitionedTopicMetadata("persistent://public/default/test-topic-p"));admin.topics().updatePartitionedTopic("persistent://public/default/test-topic-p", 3);admin.topics().deletePartitionedTopic("persistent://public/default/test-topic-p");log.info("getStats={}", admin.topics().getPartitionedStats("persistent://public/default/my-topic2", false));log.info("getInternalStats={}", admin.topics().getPartitionedInternalStats("persistent://public/default/my-topic2"));admin.topics().createSubscription("persistent://public/default/my-topic", "test-subscription", MessageId.latest);log.info("getStats={}", admin.topics().getSubscriptions("persistent://public/default/my-topic"));admin.topics().deleteSubscription("persistent://public/default/my-topic", "test-subscription");}
}
AdminCase.java

 

 

参考：
https://pulsar.apache.org/docs/3.3.x/client-libraries-java/
https://pulsar.apache.org/docs/3.3.x/admin-api-overview/

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/835189.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

使用Microsoft.Extensions.AI简化.NET中的AI集成

使用Microsoft.Extensions.AI简化.NET中的AI集成

项目介绍 Microsoft.Extensions.AI是一个创新的 .NET 库，它为平台开发人员提供了一个内聚的 C# 抽象层，简化了与大型语言模型 （LLMs） 和嵌入等 AI 服务的交互。它支持通过一组一致且标准化的 API 和约定将 AI 功能无缝集成到 .NET 应用程序中。注意：目前Microsoft.Extensi…
阅读更多...
读数据质量管理：数据可靠性与数据质量问题解决之道06数据测试

读数据质量管理：数据可靠性与数据质量问题解决之道06数据测试

数据测试1. 运行分析型数据转换 1.1. 确保ETL期间的数据质量1.1.1. ETL即“提取-转换-加载”​1.1.2. 步骤1.1.2.1. 在提取步骤中，原始数据从一些上游数据源中导出，并将其移动到暂存区> 1.1.2.1.1. MySQL> 1.1.2.1.2. NoSQL服务器> 1.1.2.1.3. CRM系统> 1.1.…
阅读更多...
鸿蒙NEXT开发案例：计数器

鸿蒙NEXT开发案例：计数器

【引言】（完整代码在最后面） 本文将通过一个简单的计数器应用案例，介绍如何利用鸿蒙NEXT的特性开发高效、美观的应用程序。我们将涵盖计数器的基本功能实现、用户界面设计、数据持久化及动画效果的添加。 【环境准备】 电脑系统：windows 10 开发工具：DevEco Studio 5.0.1 …
阅读更多...
CuVLER：通过穷尽式自监督Transformer增强无监督对象发现

CuVLER：通过穷尽式自监督Transformer增强无监督对象发现

CuVLER：通过穷尽式自监督Transformer增强无监督对象发现介绍了VoteCut，这是一种创新的无监督对象发现方法，它利用了来自多个自监督模型的特征表示。VoteCut采用基于归一化切割的图分割、聚类和像素投票方法。此外，还介绍了CuVLER（Cut-Vote-and-LEaRn），一种零样本模型，使…
阅读更多...
考研打卡（20）

考研打卡（20）

开局（20） 开始时间 2024-11-17 01:43:23 结束时间 2024-11-17 02:20:40再弄一篇数据结构一棵二叉树的前序遍历序列为ABCDEFG，它的中序遍历序列可能是______（湖南大学 2015年） A CABDEFG B ABCDEFG C DACEFBG D DBCEAFGB 答案只要按照前序序列的顺序入栈，无论怎么出栈肯…
阅读更多...
考研打卡（19）

考研打卡（19）

开局（19） 开始时间 2024-11-17 00:53:40 结束时间 2024-11-17 01:36:36在网吧数据结构 假设在有序线性表A[1..30]上进行二分查找，则比较五次查找成功的结点数为______(厦门大学 2018年) A 8 B 12 C 15 D 16C 答案查找一次成功的节点数为1，值为15 查找二次成功的节点数…
阅读更多...
51单片机定时器数码管显示

51单片机定时器数码管显示

51单片机定时器数码管显示 本次的实现效果方式采用模拟进行，芯片为AT89C51,开发软件为keil5,proteus 通过定时器实现数码管0-99秒表计数 @目录上代码效果展示介绍 上代码 代码如下: #include "reg51.h" //包含头文件reg51.h，定义了51单片机的专用寄存器unsigned …
阅读更多...
idea安装激活教程(2024.x.x,亲测有效)

idea安装激活教程(2024.x.x,亲测有效)

前言 很多朋友简信告诉我，看不懂纯文字。好吧，如今重要部分带图片，自行阅读！ 第一步 前往idea的官网，下载idea 2024.1.5(博主亲测版本)下载完成后，进行安装，next，安装完成 首次打开，会要求输入激活码才能使用 第二步 点击获取补丁文件 保存下载之后 进入文件夹*** /Je…
阅读更多...
基于字符数组s[]的s，s

基于字符数组s[]的s，s

看这样一段代码： #include <iostream> using namespace std;int main() {const char s[] = "hello";cout << "Array content (s): " << s << endl; // 输出字符串内容cout << "Address of s (&s): " &…
阅读更多...
UE5 打包安卓后出现permission required you must approve this premission in app settings: storage弹窗

UE5 打包安卓后出现permission required you must approve this premission in app settings: storage弹窗

论坛里面有人给出了利用UPL解决的方法但不会UPL，没有使用这个方法，而是参考了这篇知乎文章 其实都提到了在项目文件\Intermediate\Android\arm64_AndroidManifest.xml这个文件中的修改， 而在一开始的这个弹窗是其中的这条语句<meta-data android:name="com.epicgame…
阅读更多...
20222405 2024-2025-1 《网络与系统攻防技术》实验六实验报告

20222405 2024-2025-1 《网络与系统攻防技术》实验六实验报告

1.实验内容 经过高强度的攻击渗透操作，我现在已经熟练的掌握了Metasploit工具的用法，也深刻体会到漏洞的可怕之处，这些漏洞会严重威胁到我们的隐私安全。 2.实验过程 （1）前期渗透 ①主机发现 输入命令msfconsole进入控制台 再search arp_sweep搜索一下成功找到 输入use 0使…
阅读更多...
阿珂挂个人给别人人格化

阿珂挂个人给别人人格化

阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023