基于阿里云RocketMQ封装库的集成技术示例整体架构新建一个私有库该库封装阿里云 RocketMQ 的三种协议接入方式通过 Spring Boot 自动配置根据 YAML 中的protocol字段选择不同的消费者客户端实现。application.yml │ ▼ RocketMqProperties (ConfigurationProperties) │ ├── producer: MqProperties → ONSFactory.createProducer() │ └── consumer: MapString, MqProperties │ ▼ (根据 protocol 字段路由) ┌────────────┬────────────┬────────────┐ │ TCP │ HTTP │ SELF │ ▼ ▼ ▼ ConsumerBean MQClient DefaultMqPush (阿里云ONS) (阿里云HTTP) ConsumerWrapper (开源RocketMQ)注博客https://blog.csdn.net/badao_liumang_qizhi配置结构YAML 配置模板rocketmq:producer:accessKey:xxxsecretKey:xxxendpoint:http://{instanceId}.{region}.mq.aliyuncs.com:80sendMsgTimeoutMillis:3000# 可选默认 3000consumer:my-consumer-1:# 自定义消费者名称keyaccessKey:xxxsecretKey:xxxendpoint:http://{instanceId}.{region}.mq-internal.aliyuncs.com:8080topic:TOPIC_NAME# 支持 | 分隔多个 topictag:my-taggroupId:GID-xxxprotocol:tcp# tcp / http / selfinstanceId:MQ_INST_xxx# 仅 HTTP 协议使用msgType:NORMAL# 仅 SELF 协议使用NORMAL / ORDERLY属性类 MqPropertiespublicclassMqProperties{privateStringendpoint;privateStringaccessKey;privateStringsecretKey;privateStringtopic;privateStringtag;privateStringgroupId;privateStringinstanceId;privateStringprotocol;privateStringmsgType;privateStringsendMsgTimeoutMillis;}属性绑定关系ConfigurationEnableConfigurationProperties(RocketMqProperties.class)publicclassRocketMqConfig{...}ConfigurationProperties(prefixrocketmq)publicclassRocketMqProperties{privateMqPropertiesproducer;privateMapString,MqPropertiesconsumer;}三种协议模式对比特性TCPHTTPSELF协议类型阿里云 ONS TCP阿里云 HTTP开源 Apache RocketMQ客户端类ConsumerBeanMQClientDefaultMqPushConsumerWrapper使用accessKey/secretKey✅✅❌使用endpoint✅ (作为 NAMESRV_ADDR)✅ (作为 HTTP endpoint)✅ (作为 namesrvAddr)使用groupId✅ (GROUP_ID)✅✅使用instanceId❌✅ (区分实例)❌使用msgType❌❌✅ (NORMAL/ORDERLY)多 topic 支持✅ (用 分隔)✅ (用 消息监听接口TcpMqListenerRocketMqHttpPullerSelfBuildMqListener/SelfBuildMqOrderlyListener生产者初始化流程// RocketMqConfig.producer() 等效逻辑BeanpublicProducerproducer(){MqPropertiespropsrocketMqProperties.getProducer();if(propsnull)returnnull;PropertiespropertiesnewProperties();properties.put(AccessKey,props.getAccessKey());properties.put(SecretKey,props.getSecretKey());properties.put(NAMESRV_ADDR,props.getEndpoint());properties.put(SendMsgTimeoutMillis,StringUtils.defaultIfBlank(props.getSendMsgTimeoutMillis(),3000));ProducerproducerONSFactory.createProducer(properties);producer.start();returnproducer;}消费者初始化流程路由逻辑在ApplicationReadyEvent事件触发时遍历所有消费者配置按protocol路由EventListenerpublicvoidhandleApplicationReady(ApplicationReadyEventevent){MapString,xxxRocketMqListenerbeanscontext.getBeansOfType(JshRocketMqListener.class);// 将 listeners 按 topic 分组为 listenerMapfor(MqPropertiesmqProperties:rocketMqProperties.getConsumer().values()){MqProtocolprotocolMqProtocol.valueOf(mqProperties.getProtocol().toUpperCase());switch(protocol){caseHTTP-clients.add(createHttpClient(mqProperties,listenerMap));caseTCP-clients.add(createTcpClient(mqProperties,listenerMap));caseSELF-clients.add(createSelfClient(mqProperties,listenerMap));}}// 启动所有客户端clients.forEach(xxxRocketMqConsumerClient::start);}TCP 模式 — createTcpClientprivatexxxRocketMqConsumerClientcreateTcpClient(MqPropertiesmqProperties,MapString,xxxRocketMqListenerlistenerMap){PropertiespropertiesnewProperties();properties.setProperty(GROUP_ID,mqProperties.getGroupId());properties.setProperty(AccessKey,mqProperties.getAccessKey());properties.setProperty(SecretKey,mqProperties.getSecretKey());properties.setProperty(NAMESRV_ADDR,mqProperties.getEndpoint());ConsumerBeanconsumerBeannewConsumerBean();consumerBean.setProperties(properties);xxxRocketMqConsumerClient clientnewxxxRocketMqConsumerClient(consumerBean);// 解析多 topic为每个 topic 创建订阅ListStringtopicsLists.newArrayList(StringUtils.split(mqProperties.getTopic(),|));for(Stringtopic:topics){StringkeyString.format(%s_%s,topic,StringUtils.defaultIfBlank(mqProperties.getTag(),));xxxRocketMqListener listenerlistenerMap.get(key);SubscriptionsubscriptionnewSubscription();subscription.setTopic(topic);subscription.setExpression(mqProperties.getTag());MapSubscription,MessageListenersubscriptionTablenewHashMap();subscriptionTable.put(subscription,newTcpMqListener(listener));consumerBean.setSubscriptionTable(subscriptionTable);}returnclient;}HTTP 模式 — createHttpClientprivatexxxRocketMqConsumerClientcreateHttpClient(MqPropertiesmqProperties,MapString,xxxRocketMqListenerlistenerMap){ListStringtopicsLists.newArrayList(StringUtils.split(mqProperties.getTopic(),|));MQClientmqClientnewMQClient(mqProperties.getEndpoint(),mqProperties.getAccessKey(),mqProperties.getSecretKey());MapMQConsumer,xxxRocketMqListenermapnewHashMap();for(Stringtopic:topics){StringkeyString.format(%s_%s,topic,StringUtils.defaultIfBlank(mqProperties.getTag(),));xxxRocketMqListener listenerlistenerMap.get(key);MQConsumerconsumer;if(StringUtils.isNoneBlank(mqProperties.getInstanceId())){// 带 instanceId 的 4 参数版本consumermqClient.getConsumer(mqProperties.getInstanceId(),topic,mqProperties.getGroupId(),mqProperties.getTag());}else{// 不带 instanceId 的 3 参数版本consumermqClient.getConsumer(topic,mqProperties.getGroupId(),mqProperties.getTag());}map.put(consumer,listener);}returnnewxxxRocketMqConsumerClient(map);}SELF 模式 — createSelfClientprivatexxxRocketMqConsumerClientcreateSelfClient(MqPropertiesmqProperties,MapString,xxxRocketMqListenerlistenerMap){DefaultMqPushConsumerWrapperconsumernewDefaultMqPushConsumerWrapper(mqProperties.getGroupId());consumer.setNamesrvAddr(mqProperties.getEndpoint());consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);xxxRocketMqConsumerClient clientnewxxxRocketMqConsumerClient(consumer);ListStringtopicsLists.newArrayList(StringUtils.split(mqProperties.getTopic(),|));for(Stringtopic:topics){StringkeyString.format(%s_%s,topic,StringUtils.defaultIfBlank(mqProperties.getTag(),));xxxRocketMqListener listenerlistenerMap.get(key);consumer.subscribe(topic,mqProperties.getTag());}// 根据 msgType 注册不同类型的监听器MqTypemqTypeMqType.valueOf(mqProperties.getMsgType().toUpperCase());switch(mqType){caseNORMAL-consumer.registerMessageListener(newSelfBuildMqListener(listener));caseORDERLY-consumer.registerMessageListener(newSelfBuildMqOrderlyListener(listener));}returnclient;}业务消费者实现模板接口定义publicinterfacexxxRocketMqListener{voidconsume(byte[]bytes);Stringtopic();Stringtag();}实现示例ComponentSlf4jpublicclassOrderMqConsumerimplementsxxxRocketMqListener{Value(${rocketmq.consumer.order.tag})privateStringtag;Value(${rocketmq.consumer.order.topic})// 或直接引用业务 topic 变量privateStringtopic;Overridepublicvoidconsume(byte[]bytes){OrderDtoorderJsonUtil.deserializeObject(bytes,newTypeReferenceOrderDto(){});log.info(收到订单消息: {},order.getId());// 业务逻辑}OverridepublicStringtopic(){returntopic;}OverridepublicStringtag(){returntag;}}Listener 匹配机制库通过String.format(%s_%s, topic, tag)生成 key将xxxRocketMqListener的实现与 YAML 中的消费者配置关联起来。消费者实现类的topic()tag()返回值必须与配置一致。消息发送模板ComponentpublicclassOrderMqSender{AutowiredprivateProducerproducer;Value(${rocketmq.producer.tag})privateStringtag;Value(${rocketmq.consumer.order.topic})privateStringtopic;publicvoidsend(OrderDtoorder){byte[]bodyJsonUtil.serializeObject(order).getBytes();MessagemessagenewMessage(topic,tag,body);message.setKey(String.valueOf(order.getId()));producer.send(message);}}生命周期管理// 启动ApplicationReadyEvent 时启动所有 consumer clientclients.forEach(JshRocketMqConsumerClient::start);// 销毁ContextClosedEvent 时关闭所有 consumer clientEventListenerpublicvoidhandleContextClosed(ContextClosedEventevent){clients.forEach(client-{try{client.destroy();}catch(Exceptione){...}});}// xxxRocketMqConsumerClient 实现了 DisposableBeanpublicclassxxxRocketMqConsumerClientimplementsDisposableBean{publicvoidstart(){...}publicvoiddestroy(){...}}各属性使用总结属性ProducerTCP ConsumerHTTP ConsumerSELF ConsumeraccessKey✅✅✅❌secretKey✅✅✅❌endpoint✅✅✅✅groupId❌✅✅✅topic❌✅✅✅tag❌✅✅✅instanceId❌❌✅❌protocol❌———msgType❌❌❌✅sendMsgTimeoutMillis✅❌❌❌关键设计要点多消费者支持consumer是一个 Map支持配置多个独立消费者每个消费者可以有不同的协议、topic、groupId多 topic 订阅单个消费者配置支持|分隔的多个 topicListener 自动发现通过 Spring 的getBeansOfType(xxxRocketMqListener.class)自动收集所有实现类延迟启动消费者在ApplicationReadyEvent之后才启动确保所有 Bean 初始化完成优雅停机监听ContextClosedEvent关闭所有消费者连接