解决springboot集成rocketmq关于tag的坑

解决springboot集成rocketmq关于tag的坑

springboot集成rocketmq关于tag的坑

新项目使用springboot的若依框架集成rocketmq,选择集成RocketMQTemplate这种方式实现消息的发送和接收。

1.客户端发送代码

此处回调方法里有些业务不用关注，只关心发送方法

@Component

public class RocketMqHelper {

Logger logger = LoggerFactory.getLogger(RocketMqHelper.class);

@Resource

private RocketMQTemplate rocketMQTemplate;

public void send(ReqMsg msg){

rocketMQTemplate.asyncSend(msg.getMsg().getTopic()+":"+msg.getMsg().getTags(),

msg.getMsg(),

new SendCallback(){

@Override

public 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"KzTybNtu));

}

@Override

public 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);

@Override

public 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());

http://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的值

*

*

*/

@Component

public class ChangeSelectorExpressionBeforeMqStart implements InitializingBean {

@Value("${rocketmq.consumer.tags}")

private String tags;

@Override

public void aftehttp://rPropertiesSet() throws Exception {

RocketMQMessageListener annoTable =

CbiRocketmqConsumer.class.getAnnotation(RocketMQMessageListener.class);

// 获取代理处理器

InvocationHandler invocationHandler = Proxy.getInvocationHandler(annoTable);

// 获取私有 memberValues 属性

Field f = invocationHandler.getClass().getDeclaredField("memberValues");

f.setAccessible(true);

// 获取实例的属性map

Map 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=外网ip

namesrvAddr=外网ip:9876

brokerName=broker_tanhua

autoCreateTopicEnable=true

【说明】：

1.brokerIP1 当前broker监听的IP

2.Broker是RocketMq的核心，负责消息的传递(提供者=》消费者)以及消息的持久化存储，消息的HA机制以及服务器过滤功能。

3.autoCreateTopicEnable:自动创建Topic路由

问题一描述：

【说明】：

我第一次配置时，broker.conf配置文件中没有配置autoCreateTopicEnable，因此在程序运行时会提示没有路由信息：No route info of this topic: tanhua-sso-login

【说明】

我发送消息路由名字是tanhua-sso-login

错误信息：

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

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

解决方式：

【说明】：

我当时也在网上找了很多，有在启动时添加自动创建的也有说防火墙开启的原因，但是我感觉会这个的话应该都知道关防火墙。在启动时添加自动创建可能也好使，但是我没试过，因为我在搜索时发现问题统一指向说没有自动创建，因此我想的是直接在配置文件中进行修改，然后重启

【解决方式】：

在broker.conf配置文件中添加如下配置：

autoCreateTopicEnable=true

SpringBoot集成信息：

【application.properties】:

# RocketMQ相关配置

rocketmq.nameServer=外网IP:9876

rocketmq.producer.group=tanhua

rocketmq.producer.send-message-timeout= 6000

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

【pom.xml】:

org.apache.rocketmq

rocketmq-spring-boot-starter

2.0.4

org.apache.rocketmq

rocketmq-client

4.5.1

org.apache.rocketmq

rocketmq-common

4.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=>查看有哪些关于这个的配置。

*【完整配置】：

# RocketMQ相关配置

rocketmq.nameServer=外网IP:9876

rocketmq.producer.group=tanhua

rocketmq.producer.send-message-timeout= 6000

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。