新功能发布记录
公告
4.x 集群使用 4.x SDK 收发普通消息
操作场景
消息队列 RocketMQ 版支持多种语言的 SDK 收发不同类型的消息，本文以调用 Java SDK 为例介绍通过 4.x SDK 连接消息队列 RocketMQ 版服务端实现普通消息收发的操作过程，
前提条件
已完成前期的 RocketMQ 集群资源创建。
已参考准备工作完成 Linux 服务器准备和环境配置。
操作步骤
步骤1：安装 Java 依赖库
在 Java 项目中引入相关依赖，以 Maven 工程为例，在 pom.xml 添加以下依赖：
说明：
依赖版本要求 ≥ 4.9.3。
<!-- in your <dependencies> block --><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.9.3</version></dependency><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-acl</artifactId><version>4.9.3</version></dependency>
步骤2：生产消息
1. 创建消息生产者
// 实例化消息生产者ProducerDefaultMQProducer producer = new DefaultMQProducer(namespace,groupName,new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))// ACL权限);// 设置NameServer的地址producer.setNamesrvAddr(nameserver);// 启动Producer实例producer.start();
参数 | 说明 |
namespace | 命名空间的名称，在控制台命名空间页面复制。如果您使用的是 4.x 通用集群，此处填写集群 ID 即可。  |
groupName | 生产者组名称，在控制台集群管理中 Group 页签中复制。 |
nameserver | 集群接入地址，在控制台集群基本信息页面的接入信息模块获取。  |
secretKey | 角色名称，在控制台的角色管理页面 SecretKey 列复制。  |
accessKey | 角色密钥，在控制台的角色管理页面 AccessKey 列复制。 |
2. 发送消息
发送消息有多种方式：同步发送、异步发送、单向发送等。
for (int i = 0; i < 10; i++) {// 创建消息实例，设置topic和消息内容Message msg = new Message(topic_name, "TAG", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));// 发送消息SendResult sendResult = producer.send(msg);System.out.printf("%s%n", sendResult);}
参数 | 说明 |
topic_name | Topic 的名称，在控制台 Topic 管理页面复制。  |
TAG | 用来设置消息的 TAG。 |
// 设置发送失败后不重试producer.setRetryTimesWhenSendAsyncFailed(0);// 设置发送消息的数量int messageCount = 10;final CountDownLatch countDownLatch = new CountDownLatch(messageCount);for (int i = 0; i < messageCount; i++) {try {final int index = i;// 创建消息实体，设置topic和消息内容Message msg = new Message(topic_name, "TAG", ("Hello rocketMq " + index).getBytes(RemotingHelper.DEFAULT_CHARSET));producer.send(msg, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {// 消息发送成功逻辑countDownLatch.countDown();System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());}@Overridepublic void onException(Throwable e) {// 消息发送失败逻辑countDownLatch.countDown();System.out.printf("%-10d Exception %s %n", index, e);e.printStackTrace();}});} catch (Exception e) {e.printStackTrace();}}countDownLatch.await(5, TimeUnit.SECONDS);
参数 | 说明 |
topic_name | Topic 的名称，在控制台 Topic 管理页面复制。  |
TAG | 用来设置消息的 TAG。 |
for (int i = 0; i < 10; i++) {// 创建消息实例，设置topic和消息内容Message msg = new Message(topic_name, "TAG", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));// 发送单向消息producer.sendOneway(msg);}
参数 | 说明 |
topic_name | Topic 的名称，在控制台 Topic 管理页面复制。  |
TAG | 用来设置消息的 TAG。 |
步骤3：消费消息
1. 创建消费者
TDMQ RocketMQ 版支持 push 和 pull 两种消费模式。
// 实例化消费者DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(namespace,groupName,new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); //ACL权限// 设置NameServer的地址pushConsumer.setNamesrvAddr(nameserver);
参数 | 说明 |
namespace | 命名空间的名称，在控制台命名空间页面复制。如果您使用的是 4.x 通用集群，此处填写集群 ID 即可。  |
groupName | 生产者组名称，在控制台集群管理中 Group 页签中复制。 |
nameserver | 集群接入地址，在控制台集群基本信息页面的接入信息模块获取。  |
secretKey | 角色名称，在控制台的角色管理页面 SecretKey 列复制。  |
accessKey | 角色密钥，在控制台的角色管理页面 AccessKey 列复制。 |
// 实例化消费者DefaultLitePullConsumer pullConsumer = new DefaultLitePullConsumer(namespace,groupName,new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));// 设置NameServer的地址pullConsumer.setNamesrvAddr(nameserver);// 设置从第一个偏移量开始消费pullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
参数 | 说明 |
namespace | 命名空间的名称，在控制台命名空间页面复制。如果您使用的是 4.x 通用集群，此处填写集群 ID 即可。  |
groupName | 生产者组名称，在控制台集群管理中 Group 页签中复制。 |
nameserver | 集群接入地址，在控制台集群基本信息页面的接入信息模块获取。  |
secretKey | 角色名称，在控制台的角色管理页面 SecretKey 列复制。  |
accessKey | 角色密钥，在控制台的角色管理页面 AccessKey 列复制。 |
说明：
2. 订阅消息
根据消费模式不同，订阅方式也有所区别。
// 订阅topicpushConsumer.subscribe(topic_name, "*");// 注册回调实现类来处理从broker拉取回来的消息pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {// 消息处理逻辑System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);// 标记该消息已经被成功消费, 根据消费情况，返回处理状态return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});// 启动消费者实例pushConsumer.start();
参数 | 说明 |
topic_name | Topic 的名称，在控制台 Topic 管理页面复制。  |
"*" | 订阅表达式如果为 null 或*表达式表示订阅全部，同时支持 "tag1 || tag2 || tag3" 标识订阅多个类型的 tag。 |
// 订阅topicpullConsumer.subscribe(topic_name, "*");// 启动消费者实例pullConsumer.start();try {System.out.printf("Consumer Started.%n");while (true) {// 拉取消息List<MessageExt> messageExts = pullConsumer.poll();System.out.printf("%s%n", messageExts);}} finally {pullConsumer.shutdown();}
参数 | 说明 |
topic_name | Topic 的名称，在控制台 Topic 管理页面复制。  |
"*" | 订阅表达式如果为 null 或*表达式表示订阅全部，同时支持 "tag1 || tag2 || tag3" 标识订阅多个类型的 tag。 |
文档反馈