解决springboot集成rocketmq关于tag的坑解决springboot集成rocketmq关于t

springboot集成rocketmq关于tag的坑新项目使用springboot的若依框架集成rocketmq,选择集成RocketMQTemplate这种方式实现消息的发送和接收。
1.客户端发送代码
此处回调方法里有些业务不用关注，只关心发送方法

@Componentpublic class RocketMqHelper { Logger logger = LoggerFactory.getLogger(RocketMqHelper.class); @Resourceprivate RocketMQTemplate rocketMQTemplate; public void send(ReqMsg msg){rocketMQTemplate.asyncSend(msg.getMsg().getTopic()+":"+msg.getMsg().getTags(),msg.getMsg(),new SendCallback(){@Overridepublic void onSuccess(SendResult sendResult) {logger.debug("msgid:{} 发送成功" , sendResult.getMsgId()); logger.debug("发送mq成功后要执行的service:{}",msg.getMsg().getSendAfterMethod()); IsaveSendAfterMqLog saveSendAfterMqLog = SpringUtils.getBean(msg.getMsg().getSendAfterMethod()); saveSendAfterMqLog.saveSendAfterMqLog(new SendAfterLog(msg.getMsg(),sendResult,"0")); } @Overridepublic void onException(Throwable throwable) {logger.error("mq发送异常！{}",throwable.toString()); logger.debug("发送mq失败后执行的service: {}",msg.getMsg().getSendAfterMethod()); //异常描述截取500 length入库msg.getMsg().putUserProperty("exceptionDesc",throwable.toString()); IsaveSendAfterMqLog saveSendAfterMqLog = SpringUtils.getBean(msg.getMsg().getSendAfterMethod()); saveSendAfterMqLog.saveSendAfterMqLog(new SendAfterLog(msg.getMsg(),"1")); }}); }}

2.服务端监听消息

@Service@RocketMQMessageListener(topic = "${rocketmq.topic}", consumerGroup = "${rocketmq.consumer.group}",selectorExpression="${rocketmq.tags}")public class CbiRocketmqConsumer implements RocketMQListener { Logger logger = LoggerFactory.getLogger(CbiRocketmqConsumer.class); @Overridepublic void onMessage(CbiMsg message) {String msgBody = new String(message.getBody()); String serviceName = message.getTags(); logger.info("本次消费服务名称：{}",serviceName); AbSaveReceiveAfter saveReceiveAfter = SpringUtils.getBean(serviceName); saveReceiveAfter.saveReceiveAfter(new RecevieAfterLog(message,Constants.CONSUME_SUCCESS)); //默认消费成功}}

@RocketMQMessageListener这个注解里selectorExpression默认是*，接收topic下全部消息。想动态对tags进行配置。于是利用springboot获取yml配置。写死的时候没有问题，但是改成$表达式配置后怎么都收不到消息，经排查居然是selectorExpression这个不支持配置，会原封的按表达式进入MQ容器初始化。然而注解里面的topic,comsumerGroup都可以正常拿到配置值。
翻源码发现问题所在，项目启动时，在ListenerContainerConfiguration在这个类里初始化mq容器时，对配置进行赋值

private DefaultRocketMQListenerContainer createRocketMQListenerContainer(Object bean, RocketMQMessageListener annotation) {DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer(); container.setNameServer(rocketMQProperties.getNameServer()); container.setTopic(environment.resolvePlaceholders(annotation.topic())); container.setConsumerGroup(environment.resolvePlaceholders(annotation.consumerGroup())); container.setRocketMQMessageListener(annotation); container.setRocketMQListener((RocketMQListener) bean); container.setObjectMapper(objectMapper); return container; }

topic和comsumerGroup都在springboot环境里获取配置值了，唯独selectorExpression这个没有，直接默认注解里的。下面的问题就是需要自己在项目启动，springboot容器起来，但是rocketmq容器未起的时候，动态去改注解里配置的值。然后让Rocketmq启动。

** *因为RocketMQMessageListener不提供动态配置功能 *springboot初始化后rocket容器初始化前利用反射动态改变 *RocketMQMessageListener注解selectorExpression的值 * * */@Componentpublic class ChangeSelectorExpressionBeforeMqStart implements InitializingBean { @Value("${rocketmq.consumer.tags}")private String tags; @Overridepublic void afterPropertiesSet() throws Exception {RocketMQMessageListener annoTable = CbiRocketmqConsumer.class.getAnnotation(RocketMQMessageListener.class); // 获取代理处理器InvocationHandler invocationHandler = Proxy.getInvocationHandler(annoTable); // 获取私有 memberValues 属性Field f = invocationHandler.getClass().getDeclaredField("memberValues"); f.setAccessible(true); // 获取实例的属性mapMap memberValues = (Map) f.get(invocationHandler); // 修改属性值memberValues.put("selectorExpression", tags); } }

问题解决。。
SpringBoot集成RocketMQ及报错处理项目场景：
【说明】：
springBoot集成RocketMQ开发
【环境】：
阿里云+Centos8+RocketMQ+SpringBoot+Docker
【启动】：
docker start rmqserver rmqbroker[因为RocketMQ安装在Docket容器中，所以这样启动]
服务器broker.conf配置信息：

brokerIP1=外网ipnamesrvAddr=外网ip:9876brokerName=broker_tanhuaautoCreateTopicEnable=true

【说明】：
1.brokerIP1 当前broker监听的IP
2.Broker是RocketMq的核心，负责消息的传递(提供者=》消费者)以及消息的持久化存储，消息的HA机制以及服务器过滤功能。
3.autoCreateTopicEnable:自动创建Topic路由
问题一描述：
【说明】：
我第一次配置时，broker.conf配置文件中没有配置autoCreateTopicEnable，因此在程序运行时会提示没有路由信息：No route info of this topic: tanhua-sso-login
【说明】
【解决springboot集成rocketmq关于tag的坑】我发送消息路由名字是tanhua-sso-login

文章图片

错误信息：

No route info of this topic: tanhua-sso-login

【错误信息截图】：我没有截图网上找了一个，差不多

文章图片

解决方式：
【说明】：
我当时也在网上找了很多，有在启动时添加自动创建的也有说防火墙开启的原因，但是我感觉会这个的话应该都知道关防火墙。在启动时添加自动创建可能也好使，但是我没试过，因为我在搜索时发现问题统一指向说没有自动创建，因此我想的是直接在配置文件中进行修改，然后重启
【解决方式】：
在broker.conf配置文件中添加如下配置：
autoCreateTopicEnable=true
SpringBoot集成信息：
【application.properties】:

# RocketMQ相关配置rocketmq.nameServer=外网IP:9876rocketmq.producer.group=tanhuarocketmq.producer.send-message-timeout= 6000

【注】:这里配置的开通没有spring，我之前加spring怎么也连接不上
【pom.xml】:

org.apache.rocketmqrocketmq-spring-boot-starter2.0.4org.apache.rocketmqrocketmq-client4.5.1org.apache.rocketmqrocketmq-common4.5.1

问题二描述：
【说明】：我在修改上面的错误后，紧接着又报
【错误信息】：

RemotingTooMuchRequestException: sendDefaultImpl call timeout

【错误信息截图】：也是没有截图网上找了一个，差不多

文章图片

*【思路】：错误信息中提示call timeout，timeout一般想到到时连接或响应超时，因此在网上找到的是在发送MQ时出错，网上解决方案是：修改Mq配置文件中的sendMsgTimeout，因此想到修改可以修改SpringBoot连接MQ时的配置设置
【解决方案】：添加rocketmq.producer.send-message-timeout= 6000
【说明】：给大一点发送信息超时时间。
*【说明】：同时在SpringBoot集成RoctetMQ配置中没有sendMsgTimeout因此用rocketmq=>输入'.'=>输入sendtimeout=>查看有哪些关于这个的配置。
*【完整配置】：