tio-boot 整合 RocketMQ

tio-boot-admin 提供了 RocketMQ 整合，和 Kafka 整合一样放在 tio-boot-admin-base 中：入口配置类是 nexus.io.tio.boot.admin.config.TioAdminRocketMqConfig，API 位于 nexus.io.tio.boot.extension.rocketmq 包。它直接基于官方 rocketmq-client，不需要 Spring，也不需要 starter：

  • new TioAdminRocketMqConfig().config() 从 app.properties 读取配置，配置项与常见的 rocketmq.* 写法一致，迁移现有项目时基本可以原样搬过来。
  • 发送用 RocketMqTemplate，覆盖同步、带 Key、topic:tag、自定义属性、延迟级别、5.x 定时消息、顺序、异步和单向发送。
  • 消费用 @RocketMqMessageListener + RocketMqListener<T>，消息体按泛型自动反序列化成实体；支持集群/广播、并发/顺序、Tag/SQL92 过滤、最大重试次数，Topic 和消费组里可以写 ${app.node-id} 这样的占位符。
  • 发送钩子和消费钩子可以在消息属性里透传租户、用户等上下文。
  • 生产者和消费者随 tio-boot 关闭自动停止。

RocketMQ 服务端的安装见 RocketMQ 安装（Docker Compose），IM 场景的完整案例见 RocketMQ 即时通讯实战。

依赖

rocketmq-client 在 tio-boot-admin-base 中是可选依赖，用到 RocketMQ 的应用自己加上：

<dependency>
  <groupId>nexus.io</groupId>
  <artifactId>tio-boot-admin-base</artifactId>
  <version>${tio-boot-admin.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-client</artifactId>
  <version>5.5.0</version>
</dependency>

rocketmq-client 自带 ACL 与消息轨迹，不需要再加其他 RocketMQ 依赖。

配置

# 总开关，默认 true；设为 false 或不配置 name-server 时整合不启动
rocketmq.enabled=true
rocketmq.name-server=127.0.0.1:9876

# 生产者；不配置 group 则只消费不发送
rocketmq.producer.group=order_producer_group
rocketmq.producer.send-message-timeout=3000
rocketmq.producer.compress-message-body-threshold=4096
rocketmq.producer.max-message-size=4194304
rocketmq.producer.retry-times-when-send-failed=2
rocketmq.producer.retry-times-when-send-async-failed=2
rocketmq.producer.retry-next-server=false

# ACL（Broker 开启 aclEnable 时配置）
rocketmq.producer.access-key=
rocketmq.producer.secret-key=
rocketmq.consumer.access-key=
rocketmq.consumer.secret-key=

# 消息轨迹
rocketmq.producer.enable-msg-trace=false
rocketmq.producer.customized-trace-topic=RMQ_SYS_TRACE_TOPIC
rocketmq.consumer.enable-msg-trace=false
rocketmq.consumer.customized-trace-topic=RMQ_SYS_TRACE_TOPIC

# 按"消费组.Topic"关闭某个监听，不改代码
rocketmq.consumer.listeners.order_paid_group.order_topic=false

配置通过 EnvUtils 读取，所以同样可以用环境变量或启动参数覆盖，比如 --rocketmq.name-server=10.0.0.5:9876。

启动

在配置类里用 TioAdminRocketMqConfig 注册钩子和监听器，最后调用 config()：

import com.example.order.hook.TenantConsumeMessageHook;
import com.example.order.hook.TenantSendMessageHook;
import com.example.order.listener.ConfigRefreshListener;
import com.example.order.listener.OrderCreatedListener;
import com.example.order.listener.OrderPaidListener;
import com.example.order.listener.OrderStatusListener;

import nexus.io.annotation.AConfiguration;
import nexus.io.annotation.Initialization;
import nexus.io.tio.boot.admin.config.TioAdminRocketMqConfig;

@AConfiguration
public class RocketMqConfiguration {

  @Initialization
  public void config() {
    new TioAdminRocketMqConfig()
        .addSendMessageHook(new TenantSendMessageHook())
        .addConsumeMessageHook(new TenantConsumeMessageHook())
        .addListener(new OrderCreatedListener())
        .addListener(new OrderPaidListener())
        .addListener(new OrderStatusListener())
        .addListener(new ConfigRefreshListener())
        .config();
  }
}

config() 读取 rocketmq.* 配置，启动生产者和全部消费者，tio-boot 关闭时自动停止。没有配置 rocketmq.name-server 或 rocketmq.enabled=false 时只记录一行日志并跳过，本地开发没有 MQ 时应用照常启动。

启动日志：

INFO nexus.io.tio.boot.extension.rocketmq.RocketMqTemplate - RocketMQ producer started, group=order_producer_group, nameServer=127.0.0.1:9876
INFO nexus.io.tio.boot.extension.rocketmq.RocketMqListenerContainer - RocketMQ consumer started, topic=order_topic, group=order_created_group, model=CLUSTERING, mode=CONCURRENTLY, listener=com.example.order.listener.OrderCreatedListener
  • new TioAdminRocketMqConfig().properties(new RocketMqProperties().setNameServer("127.0.0.1:9876").setProducerGroup("g")) 可以不用配置文件。
  • TioAdminRocketMqConfig 内部调用的是 RocketMq 入口类，也可以直接使用：RocketMq.init()、addSendMessageHook、addConsumeMessageHook、addListener、start()。
  • start() 之后再调用 RocketMq.addListener 的监听器会立即启动，适合运行时才知道 Topic 的场景。
  • RocketMq.close() 手动停止全部生产者和消费者，之后可以重新初始化。

发送消息

目标地址写成 topic 或 topic:tag。消息体可以是任意对象（转成 JSON）、String（UTF-8 文本）或 byte[]（原样发送）；需要 Key、属性、延迟时用 RocketMqMessage 包一层。

下面的 Service 演示了全部发送方式：

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.example.order.model.OrderCreated;

import nexus.io.tio.boot.extension.rocketmq.RocketMq;
import nexus.io.tio.boot.extension.rocketmq.RocketMqMessage;
import nexus.io.tio.boot.extension.rocketmq.RocketMqTemplate;

public class OrderMessageService {
  private static final Logger log = LoggerFactory.getLogger(OrderMessageService.class);

  /** 同步发送，消息体转成 JSON */
  public SendResult created(OrderCreated event) {
    return RocketMq.send("order_topic", event);
  }

  /** 带业务 Key，控制台可以按 Key 查询 */
  public SendResult createdWithKey(OrderCreated event) {
    return RocketMq.send("order_topic", event, event.getOrderId());
  }

  /** topic:tag + 自定义属性 */
  public SendResult paid(OrderCreated event) {
    RocketMqMessage message = RocketMqMessage.of(event)
        .keys(event.getOrderId())
        .property("channel", "wechat");
    return RocketMq.getTemplate().send("order_topic:paid", message);
  }

  /** 延迟级别 3：Broker messageDelayLevel 中的第 3 个时长 */
  public SendResult closeIfUnpaid(OrderCreated event) {
    return RocketMq.getTemplate().syncSend("order_close_topic", event, 3000, 3);
  }

  /** 定时消息：任意秒数 */
  public SendResult remindIn(OrderCreated event, long seconds) {
    return RocketMq.getTemplate().syncSendDelayTimeSeconds("order_remind_topic", event, seconds);
  }

  /** 顺序发送：同一订单的消息进入同一个队列 */
  public SendResult status(OrderCreated event, String status) {
    return RocketMq.getTemplate().syncSendOrderly("order_status_topic:" + status, event,
        String.valueOf(event.getOrderId()));
  }

  /** 异步发送，回调在客户端线程中执行 */
  public void audit(OrderCreated event) {
    RocketMqTemplate template = RocketMq.getTemplate();
    template.asyncSend("order_audit_topic", event, new SendCallback() {
      @Override
      public void onSuccess(SendResult sendResult) {
        log.info("audit sent {}", sendResult.getMsgId());
      }

      @Override
      public void onException(Throwable e) {
        log.error("audit send failed", e);
      }
    });
  }

  /** 单向发送：不等结果、不重试，适合日志和指标 */
  public void metric(String line) {
    RocketMq.getTemplate().sendOneWay("metric_topic", line);
  }
}

OrderCreated 是普通实体类（orderId、userId、amount 三个 Long 字段加 getter/setter），后面的监听示例也用它。

发送方法一览：

  • send(destination, payload) / syncSend(destination, payload, timeout)：同步发送，返回 SendResult。
  • send(destination, payload, keys)：同步发送并设置 Key。
  • syncSend(destination, payload, timeout, delayLevel)：延迟级别消息。
  • syncSendDelayTimeSeconds / syncSendDelayTimeMills / syncSendDeliverTimeMills：定时消息，按秒、毫秒或指定投递时间点。
  • syncSendOrderly(destination, payload, hashKey)：顺序发送。
  • asyncSend / asyncSendOrderly：异步发送。
  • sendOneWay / sendOneWayOrderly：单向发送。
  • createMessage(destination, payload)：只构造原生 Message，需要调用 DefaultMQProducer 的其他能力时使用，getProducer() 可以拿到原生生产者。

RocketMqMessage 上可设置 keys、tags（覆盖地址里的 tag）、property、delayLevel、delaySeconds、delayMillis、deliverTimeMillis，所有发送方法都接受它。发送失败统一抛出 RocketMqException，原始异常在 getCause() 中。

监听消息

实现 RocketMqListener<T>，用 @RocketMqMessageListener 声明订阅。泛型决定消息体的转换方式：

  • 实体类或 List<实体>：按 JSON 反序列化。
  • String：UTF-8 文本。
  • byte[]：原始字节。
  • MessageExt：原生消息，可以读 Key、Tag、属性和重试次数。

集群消费

默认是集群消费（CLUSTERING）：同一个消费组的多个实例分摊消息，每条消息只被其中一个实例处理。

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.example.order.model.OrderCreated;

import nexus.io.tio.boot.extension.rocketmq.RocketMqListener;
import nexus.io.tio.boot.extension.rocketmq.RocketMqMessageListener;

@RocketMqMessageListener(topic = "order_topic", consumerGroup = "order_created_group", maxReconsumeTimes = 5)
public class OrderCreatedListener implements RocketMqListener<OrderCreated> {
  private static final Logger log = LoggerFactory.getLogger(OrderCreatedListener.class);

  @Override
  public void onMessage(OrderCreated event) {
    log.info("order {} created by user {}", event.getOrderId(), event.getUserId());
  }
}

onMessage 正常返回即消费成功；抛出异常则稍后重新投递，超过 maxReconsumeTimes 后进入死信队列 %DLQ%消费组，在控制台的 DLQMessage 页可以查看和重发。maxReconsumeTimes 默认 -1，表示使用客户端默认值 16 次。消息可能重复投递，业务处理需要按消息 Key 或业务 ID 做幂等。

按 Tag 过滤并读取原生消息

import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.example.order.model.OrderCreated;

import nexus.io.tio.boot.extension.rocketmq.RocketMqListener;
import nexus.io.tio.boot.extension.rocketmq.RocketMqMessageListener;
import nexus.io.tio.utils.json.JsonUtils;

@RocketMqMessageListener(topic = "order_topic", consumerGroup = "order_paid_group", selectorExpression = "paid")
public class OrderPaidListener implements RocketMqListener<MessageExt> {
  private static final Logger log = LoggerFactory.getLogger(OrderPaidListener.class);

  @Override
  public void onMessage(MessageExt message) {
    OrderCreated event = JsonUtils.parse(message.getBody(), OrderCreated.class);
    log.info("order {} paid, key={}, channel={}, reconsumeTimes={}", event.getOrderId(), message.getKeys(),
        message.getUserProperty("channel"), message.getReconsumeTimes());
  }
}

selectorExpression 支持 *、单个 Tag 和 paidA || paidB。

广播消费

广播消费（BROADCASTING）下，消费组里的每个实例都会收到每一条消息，适合刷新本地缓存、通知所有 WebSocket 节点等场景：

import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import nexus.io.tio.boot.extension.rocketmq.RocketMqListener;
import nexus.io.tio.boot.extension.rocketmq.RocketMqMessageListener;

@RocketMqMessageListener(topic = "config_refresh_topic", consumerGroup = "config_refresh_group",
    messageModel = MessageModel.BROADCASTING)
public class ConfigRefreshListener implements RocketMqListener<String> {
  private static final Logger log = LoggerFactory.getLogger(ConfigRefreshListener.class);

  @Override
  public void onMessage(String key) {
    log.info("refresh local cache of {}", key);
  }
}

广播模式的消费进度保存在各实例本地，实例首次启动时从队列最新位置开始消费，启动之前发出的消息不会补收。

顺序消费

和 syncSendOrderly 配合使用，同一个队列内的消息由一个线程按顺序处理：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.example.order.model.OrderCreated;

import nexus.io.tio.boot.extension.rocketmq.RocketMqConsumeMode;
import nexus.io.tio.boot.extension.rocketmq.RocketMqListener;
import nexus.io.tio.boot.extension.rocketmq.RocketMqMessageListener;

@RocketMqMessageListener(topic = "order_status_topic", consumerGroup = "order_status_group",
    consumeMode = RocketMqConsumeMode.ORDERLY)
public class OrderStatusListener implements RocketMqListener<OrderCreated> {
  private static final Logger log = LoggerFactory.getLogger(OrderStatusListener.class);

  @Override
  public void onMessage(OrderCreated event) {
    log.info("status change of order {}", event.getOrderId());
  }
}

顺序消费失败时会暂停当前队列 suspendCurrentQueueTimeMillis（默认 1000 毫秒）后重试，不会跳过这条消息去处理后面的消息。

SQL92 过滤

按消息属性过滤，Broker 需要开启 enablePropertyFilter=true：

import com.example.order.model.OrderCreated;

import nexus.io.tio.boot.extension.rocketmq.RocketMqListener;
import nexus.io.tio.boot.extension.rocketmq.RocketMqMessageListener;
import nexus.io.tio.boot.extension.rocketmq.RocketMqSelectorType;

@RocketMqMessageListener(topic = "order_topic", consumerGroup = "order_vip_group",
    selectorType = RocketMqSelectorType.SQL92, selectorExpression = "level = 'vip' AND amount > 1000")
public class VipOrderListener implements RocketMqListener<OrderCreated> {

  @Override
  public void onMessage(OrderCreated event) {
  }
}

占位符

topic、consumerGroup、selectorExpression、nameServer、accessKey、secretKey 中可以写 ${key} 或 ${key:默认值}，值从 tio-boot 配置中读取。多节点部署时，每个节点只订阅属于自己的 Topic：

@RocketMqMessageListener(topic = "im_push_${app.node-id}", consumerGroup = "im_push_group_${app.node-id}")

节点 1 配置 app.node-id=1，实际订阅 im_push_1。完整用法见 RocketMQ 即时通讯实战。

在代码中注册

Topic 运行时才确定，或者想用 Lambda 写监听器时，用 RocketMqConsumerOptions。Lambda 拿不到泛型信息，需要显式传入消息体类型：

import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

import com.example.order.model.OrderCreated;

import nexus.io.tio.boot.extension.rocketmq.RocketMq;
import nexus.io.tio.boot.extension.rocketmq.RocketMqConsumerOptions;
import nexus.io.tio.boot.extension.rocketmq.RocketMqListener;
import nexus.io.tio.utils.environment.EnvUtils;

public class ExampleListenerRegistration {

  public void register() {
    String nodeId = EnvUtils.get("app.node-id", "1");
    RocketMqConsumerOptions options = RocketMqConsumerOptions.of("node_cmd_" + nodeId, "node_cmd_group_" + nodeId)
        .messageModel(MessageModel.CLUSTERING)
        .maxReconsumeTimes(3)
        .consumeThreadNumber(8);
    RocketMqListener<OrderCreated> listener = event -> System.out.println(event.getOrderId());
    RocketMq.addListener(options, listener, OrderCreated.class);
  }
}

注解属性

  • topic、consumerGroup：必填。
  • selectorType / selectorExpression：TAG（默认，*）或 SQL92。
  • messageModel：CLUSTERING（默认）或 BROADCASTING。
  • consumeMode：CONCURRENTLY（默认）或 ORDERLY。
  • maxReconsumeTimes：最大重试次数，默认 -1 即 16 次。
  • consumeThreadNumber：消费线程数，默认 20。
  • consumeTimeout：单条消息消费超时，单位分钟，默认 15。
  • delayLevelWhenNextConsume：并发模式下失败后的重试延迟级别，默认 0 由 Broker 按重试次数递增。
  • suspendCurrentQueueTimeMillis：顺序模式失败后的暂停时间。
  • nameServer、accessKey、secretKey、enableMsgTrace、customizedTraceTopic：覆盖全局配置，单个监听器连接其他集群或使用独立账号时使用。

发送钩子与消费钩子

钩子直接使用 RocketMQ 原生的 SendMessageHook 和 ConsumeMessageHook。常见用法是把当前线程的租户 ID 写进消息属性，消费时再还原到消费线程，业务代码完全不用关心：

import org.apache.rocketmq.client.hook.SendMessageContext;
import org.apache.rocketmq.client.hook.SendMessageHook;

public class TenantSendMessageHook implements SendMessageHook {

  public static final String HEADER_TENANT_ID = "tenantId";

  @Override
  public String hookName() {
    return "TenantSendMessageHook";
  }

  @Override
  public void sendMessageBefore(SendMessageContext context) {
    String tenantId = TenantContext.get();
    if (tenantId != null) {
      context.getMessage().putUserProperty(HEADER_TENANT_ID, tenantId);
    }
  }

  @Override
  public void sendMessageAfter(SendMessageContext context) {
  }
}

import org.apache.rocketmq.client.hook.ConsumeMessageContext;
import org.apache.rocketmq.client.hook.ConsumeMessageHook;
import org.apache.rocketmq.common.message.MessageExt;

public class TenantConsumeMessageHook implements ConsumeMessageHook {

  @Override
  public String hookName() {
    return "TenantConsumeMessageHook";
  }

  @Override
  public void consumeMessageBefore(ConsumeMessageContext context) {
    MessageExt message = context.getMsgList().get(0);
    String tenantId = message.getUserProperty(TenantSendMessageHook.HEADER_TENANT_ID);
    if (tenantId != null) {
      TenantContext.set(tenantId);
    }
  }

  @Override
  public void consumeMessageAfter(ConsumeMessageContext context) {
    TenantContext.clear();
  }
}

TenantContext 是一个持有 ThreadLocal<String> 的工具类，提供 get、set、clear。钩子通过 TioAdminRocketMqConfig 的 addSendMessageHook、addConsumeMessageHook 注册，见上文"启动"。

消费钩子与监听器在同一个线程中执行，所以监听器里读到的就是这条消息的租户。监听器里再发送的消息也会被发送钩子带上同一个租户，上下文可以跨多跳传递。在控制台的消息详情中可以看到属性 tenantId：

按 Key 查询到的消息详情

开关

  • rocketmq.enabled=false 或不配置 rocketmq.name-server：TioAdminRocketMqConfig.config() 只记录日志并跳过，RocketMq.isEnabled() 返回 false。
  • rocketmq.consumer.listeners.<消费组>.<Topic>=false：只关闭某一个监听器，addListener 返回 null。

消费线程与业务调用

监听器运行在 RocketMQ 的消费线程中，可以直接用 Aop.get(XxxService.class) 获取 Service：

import com.example.order.model.OrderCreated;
import com.example.order.service.OrderService;

import nexus.io.jfinal.aop.Aop;
import nexus.io.tio.boot.extension.rocketmq.RocketMqListener;
import nexus.io.tio.boot.extension.rocketmq.RocketMqMessageListener;

@RocketMqMessageListener(topic = "order_topic", consumerGroup = "order_points_group")
public class OrderPointsListener implements RocketMqListener<OrderCreated> {

  @Override
  public void onMessage(OrderCreated event) {
    OrderService orderService = Aop.get(OrderService.class);
    orderService.afterCreated(event);
  }
}

消费线程数由 consumeThreadNumber 控制，耗时较长的任务可以适当调大，或者在监听器内只做轻量处理、把重活交给业务线程池。