解决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")); } @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<CbiMsg> { 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()); 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 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); // 获取实例的属性map Map<String, Object> memberValues = (Map<String, Object>) 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:
<!--RocketMQ相关--> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.0.4</version> </dependency> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.5.1</version> </dependency> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-common</artifactId> <version>4.5.1</version> </dependency>
问题二描述:
我在修改上面的错误后,紧接着又报
错误信息:
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
以上为个人经验,希望能给大家一个参考,也希望大家多多支持猪先飞。
相关文章
解决springboot使用logback日志出现LOG_PATH_IS_UNDEFINED文件夹的问题
这篇文章主要介绍了解决springboot使用logback日志出现LOG_PATH_IS_UNDEFINED文件夹的问题,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-04-28- 这篇文章主要为大家详细介绍了SpringBoot实现excel文件生成和下载,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...2021-02-09
- 这篇文章主要介绍了详解springBoot启动时找不到或无法加载主类解决办法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-09-16
- 这篇文章主要介绍了SpringBoot集成Redis实现消息队列的方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-02-10
- 这篇文章主要介绍了解决Springboot get请求是参数过长的情况,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-09-17
- 这篇文章主要介绍了Springboot+TCP监听服务器搭建过程,本文通过图文并茂的形式给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2020-10-28
Spring Boot项目@RestController使用重定向redirect方式
这篇文章主要介绍了Spring Boot项目@RestController使用重定向redirect方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-09-02- 这篇文章主要介绍了springBoot 项目排除数据库启动方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-09-10
springboot中使用@Transactional注解事物不生效的坑
这篇文章主要介绍了springboot中使用@Transactional注解事物不生效的原因,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-01-26详解SpringBoot之访问静态资源(webapp...)
这篇文章主要介绍了详解SpringBoot之访问静态资源(webapp...),文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-09-14- 这篇文章主要介绍了SpringBoot接口接收json参数解析,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-10-19
- 这篇文章主要介绍了vue+element-ui表格封装tag标签使用插槽,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-06-19
- 这篇文章主要介绍了springboot多模块包扫描问题的解决方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-09-16
Springboot mybatis plus druid多数据源解决方案 dynamic-datasource的使用详解
这篇文章主要介绍了Springboot mybatis plus druid多数据源解决方案 dynamic-datasource的使用,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2020-11-18- 这篇文章主要介绍了Springboot实现多线程注入bean的工具类操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-08-27
Springboot+MDC+traceId日志中打印唯一traceId
本文主要介绍了Springboot+MDC+traceId日志中打印唯一traceId,文中通过示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...2021-10-17SpringBoot部署到Linux读取resources下的文件及遇到的坑
本文主要给大家介绍SpringBoot部署到Linux读取resources下的文件,在平时业务开发过程中,很多朋友在获取到文件内容乱码或者文件读取不到的问题,今天给大家分享小编遇到的坑及处理方案,感兴趣的朋友跟随小编一起看看吧...2021-06-21- 这篇文章主要介绍了springboot中nacos动态路由的配置方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-09-11
SpringBoot高版本修改为低版本时测试类报错的解决方案
这篇文章主要介绍了SpringBoot高版本修改为低版本时测试类报错的解决方案,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-09-18解决Springboot整合shiro时静态资源被拦截的问题
这篇文章主要介绍了解决Springboot整合shiro时静态资源被拦截的问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-01-26