Java SDK

操作场景

本文以调用 Java SDK 为例介绍通过开源 SDK 实现消息收发的操作过程，帮助您更好地理解消息收发的完整过程。

前提条件

• 完成资源创建与准备

• 安装1.8或以上版本 JDK

• 安装2.5或以上版本 Maven

• 下载Demo

操作步骤

1. Java 项目中引入相关依赖，以 Maven 工程为例，在 pom.xml 添加以下依赖：

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

2. 创建 Pulsar 客户端。

   PulsarClient pulsarClient = PulsarClient.builder()
                  // 服务接入地址
                  .serviceUrl("pulsar://" + SERVICE_URL)
                  // 路由ID
                  .listenerName("custom:" + ROUTERID)
                  // 授权角色密钥
           .authentication(AuthenticationFactory.token(AUTHENTICATION)).build();

   参数	说明
   SERVICE_URL	集群接入地址，可以在控制台 集群管理 接入点页面查看并复制。
   AUTHENTICATION	角色密钥，在 角色管理 页面复制密钥列复制。
   ROUTERID

3. 创建生产者。

   // 构建生产者
   Producer<byte[]> producer = pulsarClient.newProducer()
           // 禁用掉batch功能
           .enableBatching(false)
           // topic完整路径，格式为persistent://集群（租户）ID/命名空间/Topic名称
           .topic(TOPICNAME).create();
   

   说明：

   参数 TOPICNAME 需要填入完整路径，即 persistent://clusterid/namespace/Topic，clusterid/namespace/topic 的部分可以从控制台上 Topic管理 页面直接复制。

4. 发送消息。

   //发送消息
   MessageId msgId = producer.newMessage()
       // 消息内容
       .value("this is a new message.".getBytes(StandardCharsets.UTF_8))
       .send();

5. 资源释放。

   // 关闭生产者
   producer.close();
   // 关闭客户端
   pulsarClient.close();

6. 创建消费者。

   // 构建byte[]类型（默认类型）的消费者
   Consumer<byte[]> consumer = pulsarClient.newConsumer()
       // topic完整路径，格式为persistent://集群（租户）ID/命名空间/Topic名称，从【Topic管理】处复制
       .topic(TOPICNAME)
       // 需要在控制台Topic详情页创建好一个订阅，此处填写订阅名
       .subscriptionName(SUBSCRIPTIONNAME)
       // 声明消费模式为share（共享）模式
       .subscriptionType(SubscriptionType.Shared)
       // 配置从最早开始消费，否则可能会消费不到历史消息
       .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
       // 订阅
       .subscribe();

   说明：

   • 参数 SUBSCRIPTIONNAME 需要写入订阅名，可在消费管理界面查看。
   • 参数 TOPICNAME 需要填入完整路径，即 persistent://clusterid/namespace/Topic，clusterid/namespace/topic 的部分可以从控制台上 Topic管理 页面直接复制。

7. 消费消息。

   // 接收当前offset对应的一条消息
   Message<byte[]> msg = consumer.receive();
   MessageId msgId = msg.getMessageId();
   String value = new String(msg.getValue());
   System.out.println("receive msg " + msgId + ",value:" + value);
   // 接收到之后必须要ack，否则offset会一直停留在当前消息，导致消息积压
   consumer.acknowledge(msg);

8. 使用监听器进行消费。

   // 消息监听器
   MessageListener<byte[]> myMessageListener = (consumer, msg) -> {
       try {
           System.out.println("Message received: " + new String(msg.getData()));
           // 回复ack
           consumer.acknowledge(msg);
       } catch (Exception e) {
           // 消费失败，回复nack
           consumer.negativeAcknowledge(msg);
       }
   };
   pulsarClient.newConsumer()
       // topic完整路径，格式为persistent://集群（租户）ID/命名空间/Topic名称，从【Topic管理】处复制
       .topic("persistent://pulsar-mmqwr5xx9n7g/sdk_java/topic1")
       // 需要在控制台Topic详情页创建好一个订阅，此处填写订阅名
       .subscriptionName("sub_topic1")
       // 声明消费模式为exclusive（独占）模式
       .subscriptionType(SubscriptionType.Exclusive)
       // 设置监听器
       .messageListener(myMessageListener)
       // 配置从最早开始消费，否则可能会消费不到历史消息
       .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
       .subscribe();

9. 登录 TDMQ Pulsar 版控制台，依次点击 Topic 管理 > Topic 名称进入消费管理页面，点开订阅名下方右三角号，可查看生产消费记录。

   

   说明：

   上述是对消息的发布和订阅方式的简单介绍。更多操作可参见 Demo 或 Pulsar 官方文档。