脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → springboot集成rocketmq tag的坑

解决springboot集成rocketmq关于tag的坑

 更新时间：2021年08月10日 12:03:32   作者：Z_winyou  
这篇文章主要介绍了解决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

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

相关文章

  • 里氏代换原则_动力节点Java学院整理

    里氏代换原则_动力节点Java学院整理

    这篇文章主要为大家详细介绍了里氏代换原则的相关资料，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
    2017-08-08
  • Struts2单选按钮详解及枚举类型的转换代码示例

    Struts2单选按钮详解及枚举类型的转换代码示例

    这篇文章主要介绍了Struts2单选按钮详解及枚举类型的转换代码示例，分享了相关代码示例，小编觉得还是挺不错的，具有一定借鉴价值，需要的朋友可以参考下
    2018-02-02
  • Java实战之小蜜蜂扩音器网上商城系统的实现

    Java实战之小蜜蜂扩音器网上商城系统的实现

    这篇文章主要介绍了如何利用Java实现简单的小蜜蜂扩音器网上商城系统，文中采用到的技术有JSP、Servlet 、JDBC、Ajax等，感兴趣的可以动手试一试
    2022-03-03
  • spring boot拦截器注入不了java bean的原因

    spring boot拦截器注入不了java bean的原因

    这篇文章主要介绍了spring boot拦截器注入不了java bean的原因，帮助大家更好的理解和学习spring boot框架，感兴趣的朋友可以了解下
    2020-11-11
  • 一定要让你搞懂Java位运算符

    一定要让你搞懂Java位运算符

    这篇文章主要为大家介绍了Java位运算符，具有一定的参考价值，感兴趣的小伙伴们可以参考一下，希望能够给你带来帮助
    2022-01-01
  • 可视化定时任务quartz集成解析全过程

    可视化定时任务quartz集成解析全过程

    在开发中有很多定时任务都不是写死的而是可以人为配置并且写到数据库中的,下面这篇文章主要给大家介绍了关于可视化定时任务quartz集成解析的相关资料,需要的朋友可以参考下
    2022-10-10
  • mybatis参数String与Integer类型的判断方式

    mybatis参数String与Integer类型的判断方式

    这篇文章主要介绍了mybatis参数String与Integer类型的判断方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2022-03-03
  • Java ThreadPool的使用解析

    Java ThreadPool的使用解析

    这篇文章主要介绍了Java ThreadPool的使用解析，帮助大家更好的理解和使用Java，感兴趣的朋友可以了解下
    2020-10-10
  • 一文详解Java方法重载与递归应用

    一文详解Java方法重载与递归应用

    方法重载 允许在同一个类中定义多个具有相同名称的方法,但 参数列表 必须不同,递归 是一种让函数调用自身的技术,它提供了一种将复杂问题分解为简单问题的方法,这样更容易解决,本文详细介绍了Java方法重载与递归应用,需要的朋友可以参考下
    2024-02-02
  • 详解Java向服务端发送文件的方法

    详解Java向服务端发送文件的方法

    这篇文章主要为大家详细介绍了Java向服务端发送文件的方法，主要是IO流，感兴趣的小伙伴们可以参考一下
    2016-05-05

最新评论