普通消息

普通消息是 Apache RocketMQ 中不具备特殊功能的消息。它们与顺序消息、定时消息和事务消息等具有特殊功能的消息不同。本主题介绍普通消息的场景、工作机制、使用方法及使用注意事项。

适用场景

普通消息通常用于微服务解耦、数据集成和事件驱动场景。这些场景大多对消息处理的时效性或顺序性要求较低，主要关注可靠的消息传输通道。

场景 1：微服务异步解耦

上图显示了一个在线电子商务交易场景。在该场景中，上游订单系统将下单和支付封装为独立的普通消息，并将其发送到 Apache RocketMQ Broker。下游系统根据需求从 Broker 订阅消息，并基于本地消费逻辑处理任务。消息之间相互独立，无需关联。

场景 2：数据集成传输

上图以离线日志收集为例。使用埋点组件收集前端应用程序的操作日志，并将日志转发给 Apache RocketMQ。每条消息都是一段日志数据，不需要 Apache RocketMQ 进行处理。Apache RocketMQ 仅需将日志数据发送到下游存储和分析系统即可。后端应用程序负责后续的处理任务。

工作原理

普通消息定义

普通消息是 Apache RocketMQ 中的基础功能消息。普通消息支持生产者与消费者之间的异步解耦和通信。

普通消息的生命周期

• 初始化：生产者构建并初始化消息，准备发送至 Broker。

• 就绪（Ready）：消息已发送至 Broker，对消费者可见并可消费。

• 处理中（Inflight）：消息被消费者获取，并按照本地业务逻辑进行处理。

  在此期间，Broker 等待消费者提交结果。若超时未收到反馈，Apache RocketMQ 会对该消息进行重试。详情请参考消费重试。

• 确认（Acked）：消费者完成消费并提交结果，Broker 标记消息消费状态。

  默认情况下，Apache RocketMQ 保留所有消息。消费确认后仅进行逻辑标记，不会立即物理删除。因此，在消息过期或空间不足被清理前，消费者可以回溯消费。

• 删除：当消息过期或存储空间不足时，Apache RocketMQ 会按滚动方式删除最早的物理文件。详情请参考消息存储与清理。

使用限制

普通消息仅支持 MessageType 为 Normal 的主题（Topic）。

示例

创建主题（Topic）

在 Apache RocketMQ 5.0 中，建议使用 mqadmin 工具创建主题。注意，需要添加消息类型作为属性参数。以下是示例：

sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=NORMAL

发送消息

您可以设置索引键（Index Key）和过滤标签（Tag）来过滤或搜索普通消息。以下示例代码展示了如何在 Java 中发送和接收普通消息：

// Send a normal message. 
  MessageBuilder messageBuilder = new MessageBuilder();
  Message message = messageBuilder.setTopic("topic")
  // Specify the message index key so that you can accurately search for the message by using a keyword.
  .setKeys("messageKey")
  // Specify the message tag so that the consumer can filter the message based on the specified tag.
  .setTag("messageTag")
  // Message body. 
  .setBody("messageBody".getBytes())
  .build();
  try {
    // Send the message. You need to pay attention to the sending result and capture exceptions such as failures. 
    SendReceipt sendReceipt = producer.send(message);
    System.out.println(sendReceipt.getMessageId());
  } catch (ClientException e) {
      e.printStackTrace();
  }
  // Consumption example 1: When you consume a normal message as a push consumer, you need only to process the message in the message listener. 
  MessageListener messageListener = new MessageListener() {
      @Override
      public ConsumeResult consume(MessageView messageView) {
          System.out.println(messageView);
          // Return the status based on the consumption result. 
          return ConsumeResult.SUCCESS;
      }
  };
  // Consumption example 2: When you consume a normal message as a simple consumer, you must obtain and consume the message, and submit the consumption result. 
  List<MessageView> messageViewList = null;
  try {
      messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
      messageViewList.forEach(messageView -> {
          System.out.println(messageView);
          // After consumption is complete, you must invoke ACK to submit the consumption result. 
          try {
              simpleConsumer.ack(messageView);
          } catch (ClientException e) {
              e.printStackTrace();
          }
      });
      } catch (ClientException e) {
      // If the pull fails due to system traffic throttling or other reasons, you must re-initiate the request to obtain the message. 
      e.printStackTrace();
  }

使用说明

设置全局唯一的索引键，以便于故障排查

您可以在 Apache RocketMQ 中设置自定义索引键（即 Message Key）。在查询和追踪消息时，索引键可以帮助您高效、准确地找到这些消息。

因此，我们建议您在发送消息时，使用业务的唯一信息（例如订单 ID 和用户 ID）作为索引。这将有助于您日后快速查找消息。