tencent cloud

发送与接收普通消息
下载PDF
最后更新时间：2025-07-15 18:32:07
发送与接收普通消息
最后更新时间： 2025-07-15 18:32:07
下载PDF

操作场景

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

前提条件

完成资源创建与准备
安装1.8或以上版本 JDK
安装2.5或以上版本 Maven
下载 Demo 或者前往 GitHub 项目

操作步骤

步骤1：安装 Java 依赖库

在 Java 项目中引入相关依赖，以 maven 工程为例，在 pom.xml 添加以下依赖：
说明：
依赖版本要求 ≥ 4.9.3, 当前建议为4.9.4。
<!-- in your <dependencies> block -->
  <dependency>
      <groupId>org.apache.rocketmq</groupId>
      <artifactId>rocketmq-client</artifactId>
      <version>4.9.4</version>
  </dependency>
  
  <dependency>
      <groupId>org.apache.rocketmq</groupId>
      <artifactId>rocketmq-acl</artifactId>
      <version>4.9.4</version>
  </dependency>

步骤2：生产消息

创建消息生产者

// 实例化消息生产者Producer
   DefaultMQProducer producer = new DefaultMQProducer(
       groupName,
       new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)) // ACL权限
   );
   // 设置NameServer的地址
   producer.setNamesrvAddr(nameserver);
   // 启动Producer实例
   producer.start();

参数
说明
groupName
生产者组名称，建议使用对应的 topic 名字。
accessKey
角色密钥，在控制台的角色管理页面 AccessKey 列复制。

secretKey
角色名称，在控制台的角色管理页面 SecretKey 列复制。
nameserver
集群接入地址，控制台集群基本信息页面的接入信息模块获取。


发送消息

发送消息有多种方式：同步发送、异步发送、单向发送等。
同步发送
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 管理页面复制。
4.x虚拟集群/专享集群：此处需拼接命名空间名称，格式为namespace全称%topic名称，例如 MQ_INSTxxx_aaa%TopicTest。
4.x通用集群/5.x集群：此处无需拼接，填写 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() {
                  @Override
                  public void onSuccess(SendResult sendResult) {
                      // 消息发送成功逻辑
                      countDownLatch.countDown();
                      System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                  }
      
                  @Override
                  public 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 管理页面复制。
4.x虚拟集群/专享集群：此处需拼接命名空间名称，格式为namespace全称%topic名称，例如 MQ_INSTxxx_aaa%TopicTest。
4.x通用集群/5.x集群：此处无需拼接，填写 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 管理页面复制。
4.x虚拟集群/专享集群：此处需拼接命名空间名称，格式为namespace全称%topic名称，例如 MQ_INSTxxx_aaa%TopicTest。
4.x通用集群/5.x集群：此处无需拼接，填写 Topic 名称即可。

TAG
用来设置消息的 TAG。
说明：
批量发送及其他情况可参见 GitHub Demo 或 RocketMQ 官方文档。

步骤3：消费消息

创建消费者

TDMQ RocketMQ 版支持 push 和 pull 两种消费模式。推荐Push消费模式
// 实例化消费者
        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(
            groupName,                                              
            new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); //ACL权限
        // 设置NameServer的地址
        pushConsumer.setNamesrvAddr(nameserver);
参数
说明
groupName
Group 的名称，在控制台 Group 管理页面复制。
4.x虚拟集群/专享集群：此处需拼接命名空间名称，格式为namespace全称%group名称，例如 MQ_INSTxxx_aaa%GroupTest。
4.x通用集群/5.x集群：此处无需拼接，填写 Group 名称即可。

nameserver
集群接入地址，控制台集群基本信息页面的接入信息模块获取。

secretKey
角色名称，在控制台的角色管理页面 SecretKey 列复制。
accessKey
角色密钥，在控制台的角色管理页面 AccessKey 列复制。



订阅消息

根据消费模式不同，订阅方式也有所区别。
// 订阅topic
      pushConsumer.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 管理页面复制。
4.x虚拟集群/专享集群：此处需拼接命名空间名称，格式为namespace全称%topic名称，例如 MQ_INSTxxx_aaa%TopicTest。
4.x通用集群/5.x集群：此处无需拼接，填写 Topic 名称即可。

"*"
订阅表达式如果为 null 或*表达式表示订阅全部，同时支持 "tag1 || tag2 || tag3" 标识订阅多个类型的 tag。

步骤4：查看消费详情

发送完成消息后会得到一个消息 ID （messageID），您可以在控制台的消息查询 > 综合查询页面查询刚刚发送的消息，以及该消息的详情和轨迹等信息。



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

本页内容是否解决了您的问题？
您也可以 联系销售 或 提交工单 以寻求帮助。
是
否

文档反馈