解决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

以上为个人经验,希望能给大家一个参考,也希望大家多多支持我们。

时间: 2021-08-09

Springboot RocketMq实现过程详解

首先,在虚拟机上安装rocketmq和rocketMq可视化控制,安装不做描述. 1.pom.xml文件添加依赖 mq的版本与连接的rocketmq版本保持一致 <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-remoting</artifactId> <version>4.4.0</version> </depende

springBoot整合RocketMQ及坑的示例代码

版本: JDK:1.8 springBoot:1.5.10 rocketMQ:4.2.0 pom 配置: <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.10.RELEASE</version> </parent> <d

解决SpringBoot整合RocketMQ遇到的坑

应用场景 在实现RocketMQ消费时,一般会用到@RocketMQMessageListener注解定义Group.Topic以及selectorExpression(数据过滤.选择的规则)为了能支持动态筛选数据,一般都会使用表达式,然后通过apollo或者cloud config进行动态切换. 引入依赖 <!-- RocketMq Spring Boot Starter--> <dependency> <groupId>org.apache.rocketmq<

浅谈Springboot整合RocketMQ使用心得

一.阿里云官网---帮助文档 https://help.aliyun.com/document_detail/29536.html?spm=5176.doc29535.6.555.WWTIUh 按照官网步骤,创建Topic.申请发布(生产者).申请订阅(消费者) 二.代码 1.配置: public class MqConfig { /** * 启动测试之前请替换如下 XXX 为您的配置 */ public static final String PUBLIC_TOPIC = "test"

浅谈springboot项目中定时任务如何优雅退出

在一个springboot项目中需要跑定时任务处理批数据时,突然有个Kill命令或者一个Ctrl+C的命令,此时我们需要当批数据处理完毕后才允许定时任务关闭,也就是当定时任务结束时才允许Kill命令生效. 启动类 启动类上我们获取到相应的上下文,捕捉相应命令.在这里插入代码片 @SpringBootApplication /**指定mapper对应包的路径*/ @MapperScan("com.youlanw.kz.dao") /**开启计划任务*/ @EnableScheduling

浅谈springboot 属性定义

本文介绍了浅谈springboot 属性定义,分享给大家.具体如下: 简单属性自定义 一般属性可以定义在通用的配置文件application.properties里面 # 自定义属性 boot.userName = yuxi 如何获取呢? 按照spring的获取方式就可以了,很简单 @Value(value = "${boot.userName}") private String userName; 复杂属性自定义 在配置里配置属性 # 复杂属性 test.id=1 test.name

浅谈SpringBoot处理url中的参数的注解

1.介绍几种如何处理url中的参数的注解 @PathVaribale 获取url中的数据 @RequestParam 获取请求参数的值 @GetMapping 组合注解,是 @RequestMapping(method = RequestMethod.GET) 的缩写 (1)PathVaribale 获取url中的数据 看一个例子,如果我们需要获取Url=localhost:8080/hello/id中的id值,实现代码如下: @RestController public class Hello

浅谈spring-boot的单元测试中,@Before不被执行的原因

我们先来看下笔者的单元测试的依赖版本: <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.2.6.RELEASE</version> <relativePath/> <!-- lookup parent from reposi

浅谈SpringBoot主流读取配置文件三种方式

读取配置SpringBoot配置文件三种方式 一.利用Bean注解中的Value(${})注解 @Data @Component public class ApplicationProperty { @Value("${application.name}") private String name; } 该方式可以自动读取当前配置文件appliation.yml  或者application.properties中的配置值 区别在于读取yml文件时候支持中文编码,peoperties需

浅谈springboot中tk.mapper代码生成器的用法说明

问:什么是tk.mapper? 答:这是一个通用的mapper框架,相当于把mybatis的常用数据库操作方法封装了一下,它实现了jpa的规范,简单的查询更新和插入操作都可以直接使用其自带的方法,无需写额外的代码. 而且它还有根据实体的不为空的字段插入和更新的方法,这个是非常好用的哈. 而且它的集成非常简单和方便,下面我来演示下使用它怎么自动生成代码. pom中引入依赖,这里引入tk.mybatis.mapper的版本依赖是因为在mapper-spring-boot-starter的新版本中没有

浅谈springboot内置tomcat和外部独立部署tomcat的区别

前两天,我去面了个试,面试官问了我个问题,独立部署的tomcat跟springboot内置的tomcat有什么区别,为什么存在要禁掉springboot的tomcat然后将项目部署到独立的tomcat当中? 我就想,不都一个样?独立部署的tomcat可以配置优化?禁AJP,开多线程,开nio?而且springboot内置的tomcat多方便,部署上服务器写个java脚本运行即可.现在考虑下有什么条件能优于内置tomcat的. 1.tomcat的优化配置多线程?内置的也可以配置多线程 server

浅谈Springboot之于Spring的优势

Spring在Java EE开发中是实际意义上的标准,但我们在开发Spring的时候可能会遇到以下令人头疼的问题: (1)大量配置文件的定义; (2)与第三方软件整合的技术问题,Spring每个新版本的推出都以减少配置作为自己的主要目标,例如: (a)推出@Component, @Service, @Repository, @Controller等注解在类上声明Bean; (b)推出@Configuration, @Bean的Java配置来替代Xml配置. 在脚本语言和敏捷开发大行其道的时代,J

浅谈SpringBoot之事务处理机制

一.Spring的事务机制 所有的数据访问技术都有事务处理机制,这些技术提供了API用来开启事务.提交事务来完成数据操作,或者在发生错误的时候回滚数据. 而Spring的事务机制是用统一的机制来处理不同数据访问技术的事务处理.Spring的事务机制提供了一个PlatformTransactionManager接口,不同的数据访问技术的事务使用不同的接口实现: 在程序中定义事务管理器的代码如下: @Bean public PlatformTransactionManager transaction