rocketmq的messagelistenerconcurrently不支持泛型,需在consumemessage中手动反序列化message.getbody(),常见用json.parseobject(bytes, clazz.class);须与发送端序列化格式一致,避免泛型擦除误用,推荐封装通用反序列化工具。

RocketMQ 的 MessageListenerConcurrently 本身不支持泛型,它只接收原始的 MessageExt 对象,因此“解析泛型消息体”不是接口层面的能力,而是你消费时主动反序列化的逻辑。关键在于:消息体(message.getBody())是字节数组,你需要按发送方约定的格式(如 JSON、Protobuf)手动转成目标类型。
确认消息体序列化格式
发送方决定消息体内容和格式,消费端必须一致。常见方式:
- JSON 字符串(最常用):发送时用
JSON.toJSONString(obj),消费时用JSON.parseObject(bytes, YourClass.class) - Protobuf 二进制:需提前定义 .proto 文件并生成 Java 类,用
YourProtoClass.parseFrom(message.getBody()) - 纯字符串或自定义二进制协议:需按约定规则逐字段解析
在 consumeMessage 中做反序列化
MessageListenerConcurrently 的 consumeMessage 方法参数是 List<messageext></messageext>,你需要遍历并逐条处理:
- 调用
message.getBody()获取byte[] - 根据格式选择反序列化方式(推荐统一用 JSON,兼容性好)
- 捕获反序列化异常(如 JSON 格式错误、类不存在),避免影响其他消息消费
示例(JSON 场景):
public class OrderMessageListener implements MessageListenerConcurrently {
@Override
public ConsumeConcurrentlyStatus consumeMessage(
List<messageext> msgs, ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
try {
String bodyStr = new String(msg.getBody(), StandardCharsets.UTF_8);
Order order = JSON.parseObject(bodyStr, Order.class);
// 处理 order 业务逻辑
} catch (Exception e) {
// 记录日志,不要抛出运行时异常(否则会触发重试)
log.error("解析消息失败,msgId={}", msg.getMsgId(), e);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 或根据策略返回 RECONSUME_LATER
}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}</messageext>
避免泛型擦除带来的误用
不要试图给 MessageListenerConcurrently 加泛型(如 MessageListenerConcurrently<order></order>),Java 泛型在运行时被擦除,RocketMQ SDK 不识别,也无实际作用。真正的类型转换发生在你调用反序列化方法那一刻,与接口声明无关。
增强类型安全的小技巧
可封装一个通用工具方法,减少重复代码:
- 写一个
deserialize(Class<t> clazz, MessageExt msg)</t>方法,内部统一处理编码、空值、异常 - 配合消息属性(
msg.getKeys()或msg.getProperty("type"))动态选择目标类,支持同一 Topic 发送多种类型消息 - 若用 Spring Boot,可结合
@RocketMQMessageListener注解 + 自定义MessageConverter实现自动泛型转换(底层仍是手动反序列化)
Java免费学习笔记:立即使用
解锁 Java 大师之旅:从入门到精通的终极指南











