个人中心
文章发布
退出
作者热门文章
- ubuntu12.04环境下使用kvm ioctl接口实现最简单的虚拟机
- Ubuntu 通过无线网络安装Ubuntu Server启动系统后连接无线网络的方法
- 在Ubuntu上搭建网桥的方法
- ubuntu 虚拟机上网方式及相关配置详解
26
4
CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.
这篇CFSDN的博客文章解决springboot集成rocketmq关于tag的坑由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.
新项目使用springboot的若依框架集成rocketmq,选择集成RocketMQTemplate这种方式实现消息的发送和接收.
此处回调方法里有些业务不用关注，只关心发送方法 。
@Componentpublic 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")); } }); } }
@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的值 * * */@Componentpublic 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开发 。
环境：阿里云+Centos8+RocketMQ+SpringBoot+Docker 。
启动：docker start rmqserver rmqbroker[因为RocketMQ安装在Docket容器中，所以这样启动] 。
brokerIP1=外网ipnamesrvAddr=外网ip:9876brokerName=broker_tanhuaautoCreateTopicEnable=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
application.properties:
# RocketMQ相关配置rocketmq.nameServer=外网IP:9876rocketmq.producer.group=tanhuarocketmq.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:9876rocketmq.producer.group=tanhuarocketmq.producer.send-message-timeout= 6000
以上为个人经验，希望能给大家一个参考，也希望大家多多支持我.
原文链接：https://blog.csdn.net/qq_35427677/article/details/103630648 。
最后此篇关于解决springboot集成rocketmq关于tag的坑的文章就讲到这里了,如果你想了解更多关于解决springboot集成rocketmq关于tag的坑的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。
26
4
0
什么是RocketMQ RocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。主要功能是异步解耦和流量削峰：。 常见的MQ主要有
1、说说你们公司线上生产环境用的是什么消息中间件? 见【2、多个mq如何选型？】 2、多个mq如何选型？ MQ 描述 Rabb
消费者从Broker拉取到消息之后，会将消息提交到线程池中进行消费，RocketMQ消息消费是批量进行的，如果一批消息的个数小于预先设置的批量消费大小，直接构建消费请求 ConsumeReques
全局有序 在RocketMQ中，如果使消息全局有序，可以为Topic设置一个消息队列，使用一个生产者单线程发送数据，消费者端也使用单线程进行消费，从而保证消息的全局有序，但是这种方式效率低，一
RocketMQ支持集群部署来保证高可用。它基于主从模式，将节点分为Master、Slave两个角色，集群中可以有多个Master节点，一个Master节点可以有多个Slave节点。Master节点
RocketMQ 4.5版本之前，可以采用主从架构进行集群部署，但是如果master节点挂掉，不能自动在集群中选举出新的Master节点，需要人工介入，在4.5版本之后提供了DLedger模式，使用
RocketMQ是通过 DefaultMQProducer 进行消息发送的，它实现了 MQProducer 接口， MQProducer 接口中定义了消息发送的方法，方法主要分为三大类：
当Broker收到生产者的消息发送请求时，会对请求进行处理，从请求中解析发送的消息数据，接下来以单个消息的接收为例，看一下消息的接收过程。 数据校验 封装消息 首先Broker会创
在上一讲中，介绍了消息的存储，生产者向Broker发送消息之后，数据会写入到CommitLog中，这一讲，就来看一下消费者是如何从Broker拉取消息的。 RocketMQ消息的消费以组为单位
NameServer是一个注册中心，提供服务注册和服务发现的功能。NameServer可以集群部署，集群中每个节点都是对等的关系（没有像ZooKeeper那样在集群中选举出一个Master节点），节
RocketMQ 4.5版本之前，可以采用主从架构进行集群部署，但是如果master节点挂掉，不能自动在集群中选举出新的Master节点，需要人工介入，在4.5版本之后提供了DLedger模式，使用
消息存储 在 【RocketMQ】消息的存储 一文中提到，Broker收到消息后会调用 CommitLog 的asyncPutMessage方法写入消息，在DLedger模式下使用的是
RocketMQ在集群模式下，同一个消费组内，一个消息队列同一时间只能分配给组内的某一个消费者，也就是一条消息只能被组内的一个消费者进行消费，为了合理的对消息队列进行分配，于是就有了负载均衡.
RocketMQ有两种获取消息的方式，分别为推模式和拉模式。 推模式 推模式在 【RocketMQ】消息的拉取 一文中已经讲过，虽然从名字上看起来是消息到达Broker后推送给消费者
主从同步的实现逻辑主要在 HAService 中，在 DefaultMessageStore 的构造函数中，对 HAService 进行了实例化，并在start方法中，启动了 HAService :
在 【RocketMQ】消息的拉取 一文中可知，消费者在启动的时候，会创建消息拉取API对象 PullAPIWrapper ，调用pullKernelImpl方法向Broker发送拉取消息的
全局有序 在RocketMQ中，如果使消息全局有序，可以为Topic设置一个消息队列，使用一个生产者单线程发送数据，消费者端也使用单线程进行消费，从而保证消息的全局有序，但是这种方式效率低，一
概述 RocketMQ 支持发送延迟消息，但不支持任意时间的延迟消息的设置，仅支持内置预设值的延迟时间间隔的延迟消息； 预设值的延迟时间间隔为： 1s、 5s、 10s、 30s、 1m
RocketMQ 支持定时消息，但是不支持任意时间精度，仅支持特定的 level，例如定时 5s， 10s， 1m 等。 其中，level=0 级表示不延时，level=1 表示 1 级延时，le
我尝试给 rockerMQ broker 加注星标，但我收到了错误消息: There is insufficient memory for the Java Runtime Environment t
我是一名优秀的程序员,十分优秀！