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 场景）：
Java Maven Common
下载
Java Maven 项目审查工作流的通用输入处理。当 Java Maven 项目以 ZIP压缩包或 GitLab 仓库 URL 形式提供且需要...时使用。
立即学习“Java免费学习笔记（深入）”；
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;
}
}
避免泛型擦除带来的误用
不要试图给 MessageListenerConcurrently 加泛型（如 MessageListenerConcurrently<Order>），Java 泛型在运行时被擦除，RocketMQ SDK 不识别，也无实际作用。真正的类型转换发生在你调用反序列化方法那一刻，与接口声明无关。
增强类型安全的小技巧
可封装一个通用工具方法，减少重复代码：
- 写一个
deserialize(Class<T> clazz, MessageExt msg)方法，内部统一处理编码、空值、异常 - 配合消息属性（
msg.getKeys()或msg.getProperty("type")）动态选择目标类，支持同一 Topic 发送多种类型消息 - 若用 Spring Boot，可结合
@RocketMQMessageListener注解 + 自定义MessageConverter实现自动泛型转换（底层仍是手动反序列化）
