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

徐锦洪  金牌会员 | 2024-11-17 07:04:38 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 831|帖子 831|积分 2493

本文主要介绍使用 Java 来操作 Pulsar，文中所使用到的软件版本：Java 17.0.7(Pulsar 服务使用)、Java 1.8.0_341(客户端使用)、Pulsar 3.3.0、pulsar-client 3.3.0。
1、引入依赖
  1. <dependency>
  2.     <groupId>org.apache.pulsar</groupId>
  3.     <artifactId>pulsar-client</artifactId>
  4.     <version>3.3.0</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>org.apache.pulsar</groupId>
  8.     <artifactId>pulsar-client-admin</artifactId>
  9.     <version>3.3.0</version>
  10. </dependency>
复制代码
2、初始化 PulsarClient 和 PulsarAdmin

PulsarClient:
  1. @Before
  2. public void before() throws PulsarClientException {
  3.     client = PulsarClient.builder()
  4.             .serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650")
  5.             .build();
  6. }
复制代码
PulsarAdmin:
  1. @Before
  2. public void before() throws PulsarClientException {
  3.     admin = PulsarAdmin.builder()
  4.             .serviceHttpUrl("http://10.49.196.30:8080,10.49.196.31:8080,10.49.196.32:8080")
  5.             .build();
  6. }
复制代码
3、消费者

3.1、同步消费
  1. @Test
  2. public void sync() throws PulsarClientException {
  3.     Consumer<String> consumer = client.newConsumer(Schema.STRING)
  4.             .topic("my-topic") //主题
  5.             .subscriptionName("my-subscription") //订阅名称
  6.             .subscriptionType(SubscriptionType.Shared) //订阅模式
  7.             .subscribe();
  9.     while (true) {
  10.         Message<String> message = consumer.receive();
  11.         try {
  12.             log.info("topicName={},value={}", message.getTopicName(), message.getValue());
  13.             consumer.acknowledge(message);
  14.         } catch (Exception e) {
  15.             consumer.negativeAcknowledge(message);
  16.         }
  17.     }
  18. }
复制代码
3.2、异步消费
  1. public void async() throws InterruptedException {
  2.     client.newConsumer(Schema.STRING)
  3.             .topic("my-topic2")
  4.             .subscriptionName("my-subscription")
  5.             .subscriptionType(SubscriptionType.Shared)
  6.             .subscribeAsync()
  7.             .thenAccept(this::receiveAsync);
  9.     Thread.sleep(1000 * 500);
  10. }
  12. private void receiveAsync(Consumer<String> consumer) {
  13.     consumer.receiveAsync().thenAccept(message -> {
  14.         try {
  15.             log.info("messageId={},value={}", message.getMessageId(), message.getValue());
  16.             consumer.acknowledge(message);
  17.         } catch (Exception e) {
  18.             consumer.negativeAcknowledge(message);
  19.         }
  20.         receiveAsync(consumer);
  21.     });
  22. }
复制代码
3.3、使用 MessageListener
  1. @Test
  2. public void messageListener() throws Exception {
  3.     PulsarClient client = PulsarClient.builder()
  4.             .serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650")
  5.             .listenerThreads(3)
  6.             .build();
  7.     MessageListener<String> messageListener = (consumer, msg) -> {
  8.         try {
  9.             log.info("messageId={},value={}", msg.getMessageId(), msg.getValue());
  10.             consumer.acknowledge(msg);
  11.         } catch (Exception e) {
  12.             consumer.negativeAcknowledge(msg);
  13.         }
  14.     };
  16.     client.newConsumer(Schema.STRING)
  17.             .topic("my-topic")
  18.             .subscriptionName("my-subscription")
  19.             .subscriptionType(SubscriptionType.Shared)
  20.             .messageListener(messageListener)
  21.             .subscribe();
  22.     Thread.sleep(1000 * 500);
  23. }
复制代码
3.4、重试信主题
  1. /**
  2. * 重试信主题
  3. * 处理失败的消息会进入重试信主题，达到最大重试次数后进入死信主题
  4. */
  5. @Test
  6. public void retryTopic() throws Exception {
  7.     Consumer<String> consumer = client.newConsumer(Schema.STRING)
  8.             .topic("my-topic-r")
  9.             .subscriptionName("my-subscription")
  10.             .subscriptionType(SubscriptionType.Shared)
  11.             .enableRetry(true)
  12.             .deadLetterPolicy(DeadLetterPolicy.builder()
  13.                     .maxRedeliverCount(5)
  14.                     .build())
  15.             .subscribe();
  16.     while (true) {
  17.         Message<String> message = consumer.receive();
  18.         try {
  19.             log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());
  20.             //TODO:业务处理，可能发生异常
  21.             //throw new RuntimeException();
  22.             consumer.acknowledge(message);
  23.         } catch (Exception e) {
  24.             log.error("", e);
  25.             consumer.reconsumeLater(message, 5L, TimeUnit.SECONDS);
  26.         }
  27.     }
  28. }
复制代码
3.5、死信主题
  1. /**
  2. * 死信主题
  3. * 由确认超时、负面确认或重试信主题 三种情况消息处理失败后，重试最大次数后消息会进入死信主题
  4. */
  5. @Test
  6. public void deadTopic() throws Exception {
  7.     Consumer<String> consumer = client.newConsumer(Schema.STRING)
  8.             .topic("my-topic-d")
  9.             .subscriptionName("my-subscription")
  10.             .subscriptionType(SubscriptionType.Shared)
  11.             .deadLetterPolicy(DeadLetterPolicy.builder()
  12.                     .maxRedeliverCount(5)
  13.                     .initialSubscriptionName("init-sub")
  14.                     .build())
  15.             .subscribe();
  16.     while (true) {
  17.         Message<String> message = consumer.receive();
  18.         try {
  19.             log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());
  20.             //TODO:业务处理，可能发生异常
  21.             //throw new RuntimeException();
  22.             consumer.acknowledge(message);
  23.         } catch (Exception e) {
  24.             log.error("", e);
  25.             consumer.negativeAcknowledge(message);
  26.         }
  27.     }
  28. }
  30. /**
  31. * 订阅死信队列，处理其中的消息
  32. */
  33. @Test
  34. public void consumerDeadTopic() throws Exception {
  35.     Consumer<String> consumer = client.newConsumer(Schema.STRING)
  36.             .topic("my-topic-d-my-subscription-DLQ")
  37.             .subscriptionName("init-sub")
  38.             .subscriptionType(SubscriptionType.Shared)
  39.             .subscribe();
  40.     while (true) {
  41.         Message<String> message = consumer.receive();
  42.         try {
  43.             log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());
  44.         } catch (Exception e) {
  45.             log.error("", e);
  46.         }
  47.         consumer.acknowledge(message);
  48.     }
  49. }
复制代码
3.6、完备代码

  1. package com.abc.demo.pulsar;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.pulsar.client.api.*;
  5. import org.junit.Before;
  6. import org.junit.Test;
  8. import java.util.concurrent.TimeUnit;
  10. @Slf4j
  11. public class ConsumerCase {
  12.     private PulsarClient client;
  14.     @Before
  15.     public void before() throws PulsarClientException {
  16.         client = PulsarClient.builder()
  17.                 .serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650")
  18.                 .build();
  19.     }
  21.     @Test
  22.     public void sync() throws PulsarClientException {
  23.         Consumer<String> consumer = client.newConsumer(Schema.STRING)
  24.                 .topic("my-topic") //主题
  25.                 .subscriptionName("my-subscription") //订阅名称
  26.                 .subscriptionType(SubscriptionType.Shared) //订阅模式
  27.                 .subscribe();
  29.         while (true) {
  30.             Message<String> message = consumer.receive();
  31.             try {
  32.                 log.info("topicName={},value={}", message.getTopicName(), message.getValue());
  33.                 consumer.acknowledge(message);
  34.             } catch (Exception e) {
  35.                 consumer.negativeAcknowledge(message);
  36.             }
  37.         }
  38.     }
  40.     @Test
  41.     public void async() throws InterruptedException {
  42.         client.newConsumer(Schema.STRING)
  43.                 .topic("my-topic2")
  44.                 .subscriptionName("my-subscription")
  45.                 .subscriptionType(SubscriptionType.Shared)
  46.                 .subscribeAsync()
  47.                 .thenAccept(this::receiveAsync);
  49.         Thread.sleep(1000 * 500);
  50.     }
  52.     private void receiveAsync(Consumer<String> consumer) {
  53.         consumer.receiveAsync().thenAccept(message -> {
  54.             try {
  55.                 log.info("messageId={},value={}", message.getMessageId(), message.getValue());
  56.                 consumer.acknowledge(message);
  57.             } catch (Exception e) {
  58.                 consumer.negativeAcknowledge(message);
  59.             }
  60.             receiveAsync(consumer);
  61.         });
  62.     }
  64.     @Test
  65.     public void messageListener() throws Exception {
  66.         PulsarClient client = PulsarClient.builder()
  67.                 .serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650")
  68.                 .listenerThreads(3)
  69.                 .build();
  70.         MessageListener<String> messageListener = (consumer, msg) -> {
  71.             try {
  72.                 log.info("messageId={},value={}", msg.getMessageId(), msg.getValue());
  73.                 consumer.acknowledge(msg);
  74.             } catch (Exception e) {
  75.                 consumer.negativeAcknowledge(msg);
  76.             }
  77.         };
  79.         client.newConsumer(Schema.STRING)
  80.                 .topic("my-topic")
  81.                 .subscriptionName("my-subscription")
  82.                 .subscriptionType(SubscriptionType.Shared)
  83.                 .messageListener(messageListener)
  84.                 .subscribe();
  85.         Thread.sleep(1000 * 500);
  86.     }
  88.     /**
  89.      * 重试信主题
  90.      * 处理失败的消息会进入重试信主题，达到最大重试次数后进入死信主题
  91.      */
  92.     @Test
  93.     public void retryTopic() throws Exception {
  94.         Consumer<String> consumer = client.newConsumer(Schema.STRING)
  95.                 .topic("my-topic-r")
  96.                 .subscriptionName("my-subscription")
  97.                 .subscriptionType(SubscriptionType.Shared)
  98.                 .enableRetry(true)
  99.                 .deadLetterPolicy(DeadLetterPolicy.builder()
  100.                         .maxRedeliverCount(5)
  101.                         .build())
  102.                 .subscribe();
  103.         while (true) {
  104.             Message<String> message = consumer.receive();
  105.             try {
  106.                 log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());
  107.                 //TODO:业务处理，可能发生异常
  108.                 //throw new RuntimeException();
  109.                 consumer.acknowledge(message);
  110.             } catch (Exception e) {
  111.                 log.error("", e);
  112.                 consumer.reconsumeLater(message, 5L, TimeUnit.SECONDS);
  113.             }
  114.         }
  115.     }
  117.     /**
  118.      * 死信主题
  119.      * 由确认超时、负面确认或重试信主题 三种情况消息处理失败后，重试最大次数后消息会进入死信主题
  120.      */
  121.     @Test
  122.     public void deadTopic() throws Exception {
  123.         Consumer<String> consumer = client.newConsumer(Schema.STRING)
  124.                 .topic("my-topic-d")
  125.                 .subscriptionName("my-subscription")
  126.                 .subscriptionType(SubscriptionType.Shared)
  127.                 .deadLetterPolicy(DeadLetterPolicy.builder()
  128.                         .maxRedeliverCount(5)
  129.                         .initialSubscriptionName("init-sub")
  130.                         .build())
  131.                 .subscribe();
  132.         while (true) {
  133.             Message<String> message = consumer.receive();
  134.             try {
  135.                 log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());
  136.                 //TODO:业务处理，可能发生异常
  137.                 //throw new RuntimeException();
  138.                 consumer.acknowledge(message);
  139.             } catch (Exception e) {
  140.                 log.error("", e);
  141.                 consumer.negativeAcknowledge(message);
  142.             }
  143.         }
  144.     }
  146.     /**
  147.      * 订阅死信队列，处理其中的消息
  148.      */
  149.     @Test
  150.     public void consumerDeadTopic() throws Exception {
  151.         Consumer<String> consumer = client.newConsumer(Schema.STRING)
  152.                 .topic("my-topic-d-my-subscription-DLQ")
  153.                 .subscriptionName("init-sub")
  154.                 .subscriptionType(SubscriptionType.Shared)
  155.                 .subscribe();
  156.         while (true) {
  157.             Message<String> message = consumer.receive();
  158.             try {
  159.                 log.info("messageId={},topicName={},value={}", message.getMessageId(), message.getTopicName(), message.getValue());
  160.             } catch (Exception e) {
  161.                 log.error("", e);
  162.             }
  163.             consumer.acknowledge(message);
  164.         }
  165.     }
  166. }
复制代码
ConsumerCase.java4、Reader
  1. @Test
  2. public void reader() throws Exception {
  3.     //MessageId messageId = MessageId.fromByteArray(Base64.getDecoder().decode("CBcQBTAA"));
  4.     MessageId messageId = MessageId.earliest;
  5.     Reader reader = client.newReader(Schema.STRING)
  6.             .topic("my-topic")
  7.             .startMessageId(messageId)
  8.             .create();
  10.     while (true) {
  11.         Message msg = reader.readNext();
  12.         log.info("messageId={},messageIdBase64={},value={}", msg.getMessageId(), Base64.getEncoder().encodeToString(msg.getMessageId().toByteArray()), msg.getValue());
  13.     }
  14. }
复制代码
完备代码：
  1. package com.abc.demo.pulsar;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.pulsar.client.api.*;
  5. import org.junit.Before;
  6. import org.junit.Test;
  8. import java.util.Base64;
  10. @Slf4j
  11. public class ReaderCase {
  12.     private PulsarClient client;
  14.     @Before
  15.     public void before() throws PulsarClientException {
  16.         client = PulsarClient.builder()
  17.                 .serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650")
  18.                 .build();
  19.     }
  21.     @Test
  22.     public void reader() throws Exception {
  23.         //MessageId messageId = MessageId.fromByteArray(Base64.getDecoder().decode("CBcQBTAA"));
  24.         MessageId messageId = MessageId.earliest;
  25.         Reader reader = client.newReader(Schema.STRING)
  26.                 .topic("my-topic")
  27.                 .startMessageId(messageId)
  28.                 .create();
  30.         while (true) {
  31.             Message msg = reader.readNext();
  32.             log.info("messageId={},messageIdBase64={},value={}", msg.getMessageId(), Base64.getEncoder().encodeToString(msg.getMessageId().toByteArray()), msg.getValue());
  33.         }
  34.     }
  35. }
复制代码
ReaderCase.java5、生产者

5.1、同步发送
  1. @Test
  2. public void sync() throws PulsarClientException {
  3.     Producer<String> producer = client.newProducer(Schema.STRING)
  4.             .topic("my-topic-d")
  5.             .create();
  6.     for (int i = 0; i < 3; i++) {
  7.         MessageId messageId = producer.send(("hello" + i));
  8.         log.info("messageId={}", messageId);
  9.     }
  10.     producer.close();
  11. }
复制代码
5.2、异步发送
  1. @Test
  2. public void async() throws InterruptedException {
  3.     client.newProducer(Schema.STRING)
  4.             .topic("my-topic2")
  5.             .createAsync()
  6.             .thenAccept(producer -> {
  7.                 for (int i = 0; i < 10; i++) {
  8.                     producer.sendAsync("hello" + i).thenAccept(messageId -> {
  9.                                 log.info("messageId={}", messageId);
  10.                     });
  11.                 }
  12.             });
  13.     Thread.sleep(1000 * 5);
  14. }
复制代码
5.3、详细设置消息
  1. @Test
  2. public void configMessage() throws PulsarClientException {
  3.     Producer<byte[]> producer = client.newProducer()
  4.             .topic("my-topic")
  5.             .create();
  6.     MessageId messageId = producer.newMessage(Schema.STRING)
  7.             .key("my-key") //设置消息key
  8.             .eventTime(System.currentTimeMillis()) //设置事件事件
  9.             .sequenceId(123) //设置 sequenceId
  10.             .deliverAfter(1, TimeUnit.MINUTES) //延迟投递消息
  11.             .property("my-key", "my-value") //自定义属性
  12.             .value("content")
  13.             .send();
  14.     log.info("messageId={}", messageId);
  15.     producer.close();
  16. }
复制代码
5.4、完备代码

  1. package com.abc.demo.pulsar;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.pulsar.client.api.*;
  5. import org.junit.After;
  6. import org.junit.Before;
  7. import org.junit.Test;
  9. import java.util.concurrent.CompletableFuture;
  10. import java.util.concurrent.TimeUnit;
  13. @Slf4j
  14. public class ProducerCase {
  15.     private PulsarClient client;
  17.     @Before
  18.     public void before() throws PulsarClientException {
  19.         client = PulsarClient.builder()
  20.                 //.serviceUrl("pulsar://10.49.196.33:6650")
  21.                 .serviceUrl("pulsar://10.49.196.30:6650,10.49.196.31:6650,10.49.196.32:6650")
  22.                 .build();
  23.     }
  25.     @After
  26.     public void after() throws PulsarClientException {
  27.         client.close();
  28.     }
  30.     @Test
  31.     public void sync() throws PulsarClientException {
  32.         Producer<String> producer = client.newProducer(Schema.STRING)
  33.                 .topic("my-topic-d")
  34.                 .create();
  35.         for (int i = 0; i < 3; i++) {
  36.             MessageId messageId = producer.send(("hello" + i));
  37.             log.info("messageId={}", messageId);
  38.         }
  39.         producer.close();
  40.     }
  42.     @Test
  43.     public void async() throws InterruptedException {
  44.         client.newProducer(Schema.STRING)
  45.                 .topic("my-topic2")
  46.                 .createAsync()
  47.                 .thenAccept(producer -> {
  48.                     for (int i = 0; i < 10; i++) {
  49.                         producer.sendAsync("hello" + i).thenAccept(messageId -> {
  50.                                     log.info("messageId={}", messageId);
  51.                         });
  52.                     }
  53.                 });
  54.         Thread.sleep(1000 * 5);
  55.     }
  57.     @Test
  58.     public void configMessage() throws PulsarClientException {
  59.         Producer<byte[]> producer = client.newProducer()
  60.                 .topic("my-topic")
  61.                 .create();
  62.         MessageId messageId = producer.newMessage(Schema.STRING)
  63.                 .key("my-key") //设置消息key
  64.                 .eventTime(System.currentTimeMillis()) //设置事件事件
  65.                 .sequenceId(123) //设置 sequenceId
  66.                 .deliverAfter(1, TimeUnit.MINUTES) //延迟投递消息
  67.                 .property("my-key", "my-value") //自定义属性
  68.                 .value("content")
  69.                 .send();
  70.         log.info("messageId={}", messageId);
  71.         producer.close();
  72.     }
  73. }
复制代码
ProducerCase.java6、Admin

6.1、Brokers

6.1.1、列出活动 broker
  1. admin.brokers().getActiveBrokers(clusterName)
复制代码
6.1.2、列出 broker 的定名空间
  1. admin.brokers().getOwnedNamespaces(cluster,brokerUrl);
复制代码
6.1.3、获取动态配置名称
  1. admin.brokers().getDynamicConfigurationNames();
复制代码
6.1.4、更新动态配置
  1. admin.brokers().updateDynamicConfiguration(configName, configValue);
复制代码
6.1.5、获取已经更新过的动态配置
  1. admin.brokers().getAllDynamicConfigurations();
复制代码
6.1.6、获取 leader broker
  1. admin.brokers().getLeaderBroker()
复制代码
6.2、Clusters

6.2.1、获取集群配置信息
  1. admin.clusters().getCluster(clusterName);
复制代码
6.2.2、获取集群列表
  1. admin.clusters().getClusters();
复制代码
6.3、Tenant

6.3.1、列出租户
  1. admin.tenants().getTenants();
复制代码
6.3.2、创建租户
  1. admin.tenants().createTenant(tenantName, tenantInfo);
复制代码
6.3.3、获取租户配置信息
  1. admin.tenants().getTenantInfo(tenantName);
复制代码
6.3.4、删除租户
  1. admin.Tenants().deleteTenant(tenantName);
复制代码
6.3.5、更新租户
  1. admin.tenants().updateTenant(tenantName, tenantInfo);
复制代码
6.4、Namespaces

6.4.1、创建定名空间
  1. admin.namespaces().createNamespace(namespace);
复制代码
6.4.2、获取定名空间策略
  1. admin.namespaces().getPolicies(namespace);
复制代码
6.4.3、列出定名空间
  1. admin.namespaces().getNamespaces(tenant);
复制代码
6.4.4、删除定名空间
  1. admin.namespaces().deleteNamespace(namespace);
复制代码
6.5、Topics

6.5.1、查察订阅的待消费消息(不会消费消息)
  1. admin.topics().peekMessages(topic, subName, numMessages);
复制代码
6.5.2、通过消息 ID 获取消息
  1. admin.topics().getMessageById(topic, ledgerId, entryId);
复制代码
6.5.3、通过相对于最早或最新消息的位置来查找消息
  1. admin.topics().examineMessage(topic, "latest", 1);
复制代码
6.5.4、通过期间获取消息的 ID
  1. admin.topics().getMessageIdByTimestamp(topic, timestamp);
复制代码
6.5.5、跳过特定主题的某个订阅中的多少条未消费消息
  1. admin.topics().skipMessages(topic, subName, numMessages);
复制代码
6.5.6、跳过特定主题的某个订阅中的所有未消费消息
  1. admin.topics().skipAllMessages(topic, subName);
复制代码
6.5.7、根据时间重置游标
  1. admin.topics().resetCursor(topic, subName, timestamp);
复制代码
6.5.8、获取主题所属的 broker
  1. admin.lookups().lookupTopic(topic);
复制代码
6.5.9、获取分区主题所属的 broker
  1. admin.lookups().lookupPartitionedTopic(topic);
复制代码
6.5.10、获取主题的订阅信息
  1. admin.topics().getSubscriptions(topic);
复制代码
6.5.11、获取最新消息的 ID
  1. admin.topics().getLastMessage(topic);
复制代码
6.5.12、创建非分区主题
  1. admin.topics().createNonPartitionedTopic(topicName);
复制代码
6.5.13、删除非分区主题
  1. admin.topics().delete(topic);
复制代码
6.5.14、列出非分区主题
  1. admin.topics().getList(namespace);
复制代码
6.5.15、获取非分区主题状态
  1. admin.topics().getStats(topic, false /* is precise backlog */);
复制代码
6.5.16、获取非分区主题内部状态
  1. admin.topics().getInternalStats(topic);
复制代码
6.5.17、创建分区主题
  1. admin.topics().createPartitionedTopic(topicName, numPartitions);
复制代码
6.5.18、获取分区主题元数据信息
  1. admin.topics().getPartitionedTopicMetadata(topicName);
复制代码
6.5.19、更新分区主题的分区数(只能比原来大)
  1. admin.topics().updatePartitionedTopic(topic, numPartitions);
复制代码
6.5.20、删除分区主题
  1. admin.topics().deletePartitionedTopic(topic);
复制代码
6.5.21、列出分区主题
  1. admin.topics().getPartitionedTopicList(namespace);
复制代码
6.5.22、获取分区主题状态
  1. admin.topics().getPartitionedStats(topic, true /* per partition */, false /* is precise backlog */);
复制代码
6.5.23、获取分区主题内部状态
  1. admin.topics().getPartitionedInternalStats(topic);
复制代码
6.5.24、创建订阅
  1. admin.topics().createSubscription(topic, subscriptionName, MessageId.latest);
复制代码
6.5.25、获取订阅
  1. admin.topics().getSubscriptions(topic);
复制代码
6.5.26、删除订阅
  1. admin.topics().deleteSubscription(topic, subscriptionName);
复制代码
6.6、完备代码

  1. package com.abc.demo.pulsar;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.pulsar.client.admin.PulsarAdmin;
  5. import org.apache.pulsar.client.admin.PulsarAdminException;
  6. import org.apache.pulsar.client.api.Message;
  7. import org.apache.pulsar.client.api.MessageId;
  8. import org.apache.pulsar.client.api.PulsarClientException;
  9. import org.apache.pulsar.common.policies.data.TenantInfoImpl;
  10. import org.junit.After;
  11. import org.junit.Before;
  12. import org.junit.Test;
  14. import java.util.Collections;
  15. import java.util.HashSet;
  17. @Slf4j
  18. public class AdminCase {
  19.     private PulsarAdmin admin;
  21.     @Before
  22.     public void before() throws PulsarClientException {
  23.         admin = PulsarAdmin.builder()
  24.                 .serviceHttpUrl("http://10.49.196.30:8080,10.49.196.31:8080,10.49.196.32:8080")
  25.                 .build();
  26.     }
  28.     @After
  29.     public void after() {
  30.         admin.close();
  31.     }
  33.     @Test
  34.     public void broker() throws PulsarAdminException {
  35.         log.info("getActiveBrokers={}", admin.brokers().getActiveBrokers());
  36.         log.info("getOwnedNamespaces={}", admin.brokers().getOwnedNamespaces("pulsar-cluster-1", "app1:8080"));
  37.         log.info("getDynamicConfigurationNames={}", admin.brokers().getDynamicConfigurationNames());
  38.         admin.brokers().updateDynamicConfiguration("allowAutoTopicCreation", "true");
  39.         log.info("getAllDynamicConfigurations={}", admin.brokers().getAllDynamicConfigurations());
  40.         log.info("getLeaderBroker={}", admin.brokers().getLeaderBroker());
  41.     }
  43.     @Test
  44.     public void clusters() throws PulsarAdminException {
  45.         log.info("getCluster={}", admin.clusters().getCluster("pulsar-cluster-1"));
  46.         log.info("getClusters={}", admin.clusters().getClusters());
  47.     }
  48.    
  49.     @Test
  50.     public void tenants() throws PulsarAdminException {
  51.         log.info("getTenants={}", admin.tenants().getTenants());
  52.         TenantInfoImpl tenantInfo = new TenantInfoImpl();
  53.         tenantInfo.setAdminRoles(new HashSet<>());
  54.         tenantInfo.setAllowedClusters(Collections.singleton("pulsar-cluster-1"));
  55.         admin.tenants().createTenant("test-tenant", tenantInfo);
  56.         log.info("getTenantInfo={}", admin.tenants().getTenantInfo("test-tenant"));
  57.         admin.tenants().updateTenant("test-tenant", tenantInfo);
  58.         admin.tenants().deleteTenant("test-tenant");
  59.     }
  61.     @Test
  62.     public void namespaces() throws PulsarAdminException {
  63.         admin.namespaces().createNamespace("public/test-ns");
  64.         log.info("getPolicies={}", admin.namespaces().getPolicies("public/default"));
  65.         log.info("getNamespaces={}", admin.namespaces().getNamespaces("public"));
  66.         admin.namespaces().deleteNamespace("public/test-ns");
  67.     }
  68.    
  69.     @Test
  70.     public void topics() throws PulsarAdminException {
  71.         log.info("peekMessages={}", admin.topics().peekMessages("persistent://public/default/my-topic", "my-subscription", 3));
  72.         Message<byte[]> message = admin.topics().getMessageById("persistent://public/default/my-topic", 171, 16);
  73.         log.info("getMessageById={}", new String(message.getData()));
  74.         message = admin.topics().examineMessage("persistent://public/default/my-topic", "latest", 1);
  75.         log.info("examineMessage={}", new String(message.getData()));
  76.         log.info("getMessageIdByTimestamp={}", admin.topics().getMessageIdByTimestamp("persistent://public/default/my-topic", System.currentTimeMillis()));
  77.         admin.topics().skipMessages("persistent://public/default/my-topic", "my-subscription", 1);
  78.         admin.topics().skipAllMessages("persistent://public/default/my-topic", "my-subscription");
  79.         admin.topics().resetCursor("persistent://public/default/my-topic", "my-subscription", System.currentTimeMillis() - 1000 * 60 * 15);
  80.         log.info("lookupTopic={}", admin.lookups().lookupTopic("persistent://public/default/my-topic"));
  81.         log.info("lookupPartitionedTopic={}", admin.lookups().lookupPartitionedTopic("persistent://public/default/my-topic2"));
  82.         log.info("getSubscriptions={}", admin.topics().getSubscriptions("persistent://public/default/my-topic"));
  83.         log.info("getLastMessageId={}", admin.topics().getLastMessageId("persistent://public/default/my-topic"));
  85.         admin.topics().createNonPartitionedTopic("persistent://public/default/test-topic");
  86.         admin.topics().delete("persistent://public/default/test-topic");
  87.         log.info("getList={}", admin.topics().getList("public/default"));
  88.         log.info("getStats={}", admin.topics().getStats("persistent://public/default/my-topic", false));
  89.         log.info("getInternalStats={}", admin.topics().getInternalStats("persistent://public/default/my-topic"));
  91.         admin.topics().createPartitionedTopic("persistent://public/default/test-topic-p", 2);
  92.         log.info("getPartitionedTopicMetadata={}", admin.topics().getPartitionedTopicMetadata("persistent://public/default/test-topic-p"));
  93.         admin.topics().updatePartitionedTopic("persistent://public/default/test-topic-p", 3);
  94.         admin.topics().deletePartitionedTopic("persistent://public/default/test-topic-p");
  95.         log.info("getStats={}", admin.topics().getPartitionedStats("persistent://public/default/my-topic2", false));
  96.         log.info("getInternalStats={}", admin.topics().getPartitionedInternalStats("persistent://public/default/my-topic2"));
  98.         admin.topics().createSubscription("persistent://public/default/my-topic", "test-subscription", MessageId.latest);
  99.         log.info("getStats={}", admin.topics().getSubscriptions("persistent://public/default/my-topic"));
  100.         admin.topics().deleteSubscription("persistent://public/default/my-topic", "test-subscription");
  101.     }
  102. }
复制代码
AdminCase.java 
 
参考：
https://pulsar.apache.org/docs/3.3.x/client-libraries-java/
https://pulsar.apache.org/docs/3.3.x/admin-api-overview/

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

徐锦洪

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表