当前位置 博文首页 > 文章内容

    解决springboot集成rocketmq关于tag的坑

    作者:shunshunshun18 栏目:未分类 时间:2021-08-10 14:45:23

    本站于2023年9月4日。收到“大连君*****咨询有限公司”通知
    说我们IIS7站长博客,有一篇博文用了他们的图片。
    要求我们给他们一张图片6000元。要不然法院告我们

    为避免不必要的麻烦,IIS7站长博客,全站内容图片下架、并积极应诉
    博文内容全部不再显示,请需要相关资讯的站长朋友到必应搜索。谢谢!

    另祝:版权碰瓷诈骗团伙,早日弃暗投明。

    相关新闻:借版权之名、行诈骗之实,周某因犯诈骗罪被判处有期徒刑十一年六个月

    叹!百花齐放的时代,渐行渐远!



    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

    以上为个人经验,希望能给大家一个参考,也希望大家多多支持IIS7站长之家博文。