当前位置： 首页 > news >正文

RocketMQ 自动注入消费者

news 2024/11/8 3:52:08

目录

  • 前言
  • 一、情景介绍
  • 二、问题分析
  • 三、代码实现

前言

之前接到一个需求，我们项目的技术负责人希望通过配置的形式，在项目启动的时候自动根据配置生成对应的消费者

觉得还有点意思，随即记录一下~

一、情景介绍

比如我这里有一个消费者 MessageConsumer：

@Slf4j
@Service
@RocketMQMessageListener(consumerGroup = "mike-group",topic = "mike-message",selectorExpression = "TAG_MESSAGE_CONSUMER",consumeThreadMax = 6,consumeTimeout = 60L)
public class MessageConsumer implements RocketMQListener<NotifyMessage> {@Overridepublic void onMessage(NotifyMessage notifyMessage) {System.err.println("我收到啦~~");System.err.println("message = " + notifyMessage);}
}

在项目启动的时候会根据 @RocketMQMessageListener 注解上的配置生成一个消费者

假如我还需要一个 MessageConsumer 消费者，其 selectorExpression 的配置为 TAG_MESSAGE_CONSUMER_01，consumeThreadMax 要设置为 8

通常情况下我们会再复制一个 MessageConsumer 命名为 MessageConsumer_01，然后在新的消费者上改对应的配置，例如：

@Slf4j
@Service
@RocketMQMessageListener(consumerGroup = "mike-group-01",topic = "mike-message",selectorExpression = "TAG_MESSAGE_CONSUMER_01",consumeThreadMax = 8,consumeTimeout = 60L)
public class MessageConsumer_01 implements RocketMQListener<NotifyMessage> {@Overridepublic void onMessage(NotifyMessage notifyMessage) {System.err.println("我收到啦~~");System.err.println("message = " + notifyMessage);}
}

这样做虽然没啥问题，只是这两个类除了配置不一样，其它的代码都是一摸一样的，倘若之后还要有一个 selectorExpression = TAG_MESSAGE_CONSUMER_02 的消费者，那我又得再复制一个 MessageConsumer，这样就造成了大量的代码冗余

所以就希望通过读取配置文件生成对应配置的消费者

二、问题分析

要如何实现这个功能，可以去看下 RocketMQ 的源码，看 Spring 是如何创建 RocketMQ 的消费者的

源码如下：

org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer#initRocketMQPushConsumer

在这里插入图片描述

在该方法中可以看到 Spring 是如何初始化消费者，参照这个方法，只需要在项目启动完成后，将初始化从注解上获取消费者配置的地方换成从配置文件上获取就可以了

通过实现 ApplicationListener<ApplicationReadyEvent> 可以监听项目是否启动完成

三、代码实现

因为消费者是需要通过配置文件的配置来自动生成，那么可以将需要自动生成的消费者（比如 MessageConsumer）其 @RocketMQMessageListener 的配置注释掉

@Slf4j
@Service
//@RocketMQMessageListener(
//        consumerGroup = "mike-group",
//        topic = "mike-message",
//        selectorExpression = "TAG_MESSAGE_CONSUMER",
//        consumeThreadMax = 6,
//        consumeTimeout = 60L)
public class MessageConsumer implements RocketMQListener<NotifyMessage> {@Overridepublic void onMessage(NotifyMessage notifyMessage) {System.err.println("我收到啦~~");System.err.println("message = " + notifyMessage);}
}

配置文件上自动注入消费者的配置最好和 @RocketMQMessageListener 的属性相同，并且可以配置多个自动注入的消费者，那么对应的映射文件可以这么写

AutoConsumerProperties.java

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Configuration;import java.util.List;@Data
@RefreshScope
@Configuration
@ConfigurationProperties(prefix = "auto-consumer")
public class AutoConsumerProperties {private List<AutoConsumer> messageConsumer;
}

AutoConsumer.java

import lombok.Data;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.SelectorType;@Data
public class AutoConsumer {private String consumerGroup;private String topic;private SelectorType selectorType = SelectorType.TAG;private String selectorExpression = "*";private ConsumeMode consumeMode = ConsumeMode.CONCURRENTLY;private MessageModel messageModel = MessageModel.CLUSTERING;private int consumeThreadMin = 64;private int consumeThreadMax = 64;private long consumeTimeout = 15L;private String accessKey;private String secretKey;private boolean enableMsgTrace;private String customizedTraceTopic;private String nameServer;private String accessChannel;
}

核心代码：

ConsumerStarted.java

import cn.hutool.core.collection.CollUtil;
import com.mike.common.core.utils.JacksonUtil;
import com.mike.server.message.config.properties.AutoConsumer;
import com.mike.server.message.config.properties.AutoConsumerProperties;
import com.mike.server.message.consumer.MessageConsumer;
import com.mike.server.message.domain.entity.NotifyMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Objects;@Slf4j
@Component
public class ConsumerStarted implements ApplicationContextAware, ApplicationListener<ApplicationReadyEvent> /* , InitializingBean, SmartLifecycle */ {@Value("${rocketmq.name-server:}")private String nameServer;@Value("${rocketmq.consumer.topic:}")private String topic;@Value("${rocketmq.consumer.access-key:}")private String accessKey;@Value("${rocketmq.consumer.secret-key:}")private String secretKey;@Resourceprivate AutoConsumerProperties autoConsumerProperties;@Resourceprivate MessageConsumer messageConsumer;private ApplicationContext applicationContext;private final static boolean enableMsgTrace = true;private final static String customizedTraceTopic = null;@Override@SuppressWarnings("all")public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {// 需要等到程序启动完全之后再去启动initConsumer();}public void initConsumer() {List<AutoConsumer> messageConsumers = autoConsumerProperties.getMessageConsumer();if (CollUtil.isEmpty(messageConsumers)) return;final RocketMQListener<NotifyMessage> messageConsumerListener = messageConsumer;this.autoGenerateConsumer(messageConsumers, messageConsumerListener, NotifyMessage.class);}@SuppressWarnings("all")private <R> void autoGenerateConsumer(List<AutoConsumer> autoConsumers, RocketMQListener<R> rocketMQListener, Class<R> objClass) {// 根据 tag 自动生成对应的消费者for (AutoConsumer autoConsumer : autoConsumers) {String consumerGroup = autoConsumer.getConsumerGroup();String nameServer = getValueOrDefault(autoConsumer.getNameServer(), this.nameServer);String topic = getValueOrDefault(autoConsumer.getTopic(), this.topic);String accessKey = getValueOrDefault(autoConsumer.getAccessKey(), this.accessKey);String secretKey = getValueOrDefault(autoConsumer.getSecretKey(), this.secretKey);try {Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");Assert.notNull(nameServer, "Property 'nameServer' is required");Assert.notNull(topic, "Property 'topic' is required");DefaultMQPushConsumer consumer;RPCHook rpcHook = RocketMQUtil.getRPCHookByAkSk(this.applicationContext.getEnvironment(), accessKey, secretKey);if (Objects.nonNull(rpcHook)) {consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook, new AllocateMessageQueueAveragely(), enableMsgTrace, customizedTraceTopic);consumer.setVipChannelEnabled(false);} else {consumer = new DefaultMQPushConsumer(consumerGroup, enableMsgTrace, customizedTraceTopic);}consumer.setInstanceName(RocketMQUtil.getInstanceName(this.nameServer));consumer.setNamesrvAddr(this.nameServer);consumer.setAccessChannel(AccessChannel.LOCAL);consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.setConsumeThreadMin(autoConsumer.getConsumeThreadMin());consumer.setConsumeThreadMax(autoConsumer.getConsumeThreadMax());if (consumer.getConsumeThreadMax() < consumer.getConsumeThreadMin()) {consumer.setConsumeThreadMin(consumer.getConsumeThreadMax());}switch (autoConsumer.getMessageModel()) {case BROADCASTING:consumer.setMessageModel(MessageModel.BROADCASTING);break;case CLUSTERING:consumer.setMessageModel(MessageModel.CLUSTERING);break;default:throw new IllegalArgumentException("Property 'messageModel' was wrong.");}switch (autoConsumer.getSelectorType()) {case TAG:consumer.subscribe(topic, autoConsumer.getSelectorExpression());break;case SQL92:consumer.subscribe(topic, MessageSelector.bySql(autoConsumer.getSelectorExpression()));break;default:throw new IllegalArgumentException("Property 'selectorType' was wrong.");}switch (autoConsumer.getConsumeMode()) {case ORDERLY:consumer.setMessageListener(new DefaultMessageListenerOrderly(autoConsumer, rocketMQListener, objClass));break;case CONCURRENTLY:consumer.setMessageListener(new DefaultMessageListenerConcurrently(autoConsumer, rocketMQListener, objClass));break;default:throw new IllegalArgumentException("Property 'consumeMode' was wrong.");}consumer.start();log.info("Consumer Start Success: {}:{}", topic, autoConsumer.getSelectorExpression());} catch (MQClientException e) {e.printStackTrace();log.info("Consumer Start Failed: {}:{}", topic, autoConsumer.getSelectorExpression());}}}private String getValueOrDefault(String value, String defaultValue) {return StringUtils.isNotBlank(value)? value: defaultValue;}@Override@SuppressWarnings("all")public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}public class DefaultMessageListenerOrderly<T> implements MessageListenerOrderly {private final AutoConsumer autoConsumer;private final RocketMQListener<T> rocketMQListener;private final Class<T> objClass;public DefaultMessageListenerOrderly(AutoConsumer autoConsumer, RocketMQListener<T> rocketMQListener, Class<T> objClass) {this.autoConsumer = autoConsumer;this.rocketMQListener = rocketMQListener;this.objClass = objClass;}public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgList, ConsumeOrderlyContext context) {for (MessageExt messageExt : msgList) {log.info("group[{}]-tag[{}] consume start ->>>", autoConsumer.getConsumerGroup(), autoConsumer.getSelectorExpression());log.debug("received msg: {}", messageExt);try {long now = System.currentTimeMillis();this.rocketMQListener.onMessage(doConvertMessage(messageExt, this.objClass));long costTime = System.currentTimeMillis() - now;log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);} catch (Exception var9) {log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), var9);final long suspendCurrentQueueTimeMillis = 1000L;context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis);return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;}}return ConsumeOrderlyStatus.SUCCESS;}}public class DefaultMessageListenerConcurrently <T> implements MessageListenerConcurrently {private final AutoConsumer autoConsumer;private final RocketMQListener<T> rocketMQListener;private final Class<T> objClass;public DefaultMessageListenerConcurrently(AutoConsumer autoConsumer, RocketMQListener<T> rocketMQListener, Class<T> objClass) {this.autoConsumer = autoConsumer;this.rocketMQListener = rocketMQListener;this.objClass = objClass;}public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext context) {for (MessageExt messageExt : msgList) {log.info("group[{}]-tag[{}] consume start ->>>", autoConsumer.getConsumerGroup(), autoConsumer.getSelectorExpression());log.debug("received msg: {}", messageExt);try {long now = System.currentTimeMillis();this.rocketMQListener.onMessage(doConvertMessage(messageExt, objClass));long costTime = System.currentTimeMillis() - now;log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);} catch (Exception var9) {log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), var9);final int delayLevelWhenNextConsume = 0;context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);return ConsumeConcurrentlyStatus.RECONSUME_LATER;}}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}}@SuppressWarnings("unchecked")private <T> T doConvertMessage(MessageExt messageExt, Class<T> objClass) {if (Objects.equals(objClass, MessageExt.class)) {return (T)messageExt;} else {String str = new String(messageExt.getBody(), StandardCharsets.UTF_8);if (Objects.equals(objClass, String.class)) {return (T)str;} else {if (objClass != null) {return JacksonUtil.fromJson(str, objClass);} else {log.info("convert failed. str:{}, msgType:{}", str, null);throw new RuntimeException("cannot convert message to " + null);}}}}
}

配置文件 yml 新增自动注入消费者的配置

auto-consumer:message-consumer:- consumer-group: mico-grouptopic: mike-messageselector-expression: TAG_MESSAGE_CONSUMERconsume-thread-max: 6- consumer-group: mike-group-01topic: mike-messageselector-expression: TAG_MESSAGE_CONSUMER_01consume-thread-max: 8- consumer-group: mike-group-02topic: mike-messageselector-expression: TAG_MESSAGE_CONSUMER_02consume-thread-max: 10

如果是配置在 properties 文件中，配置如下：

auto-consumer.message-consumer[0].consumer-group = mico-group
auto-consumer.message-consumer[0].topic = mike-message
auto-consumer.message-consumer[0].selector-expression = TAG_MESSAGE_CONSUMER
auto-consumer.message-consumer[0].consume-thread-max = 6auto-consumer.message-consumer[1].consumer-group = mico-group-01
auto-consumer.message-consumer[1].topic = mike-message
auto-consumer.message-consumer[1].selector-expression = TAG_MESSAGE_CONSUMER_01
auto-consumer.message-consumer[1].consume-thread-max = 8auto-consumer.message-consumer[2].consumer-group = mico-group-02
auto-consumer.message-consumer[2].topic = mike-message
auto-consumer.message-consumer[2].selector-expression = TAG_MESSAGE_CONSUMER_02
auto-consumer.message-consumer[2].consume-thread-max = 10

启动项目进行验证，观察是否有三个消费者被创建

在这里插入图片描述

从日志上看确实根据配置文件自动创建了三个不同的消费者


http://www.mrgr.cn/news/68113.html

相关文章：

  • jenkins使用slave节点进行node打包报错问题处理
  • 面试题分享11月7日
  • windows使用 cmd 进行批量删除文件夹下，全部小于5k的文件、txt文件、gif文件
  • rabbitMq怎么保证消息不丢失？消费者没有接收到消息怎么处理
  • idea database连接数据库后看不到表解决方法、格式化sql快捷键
  • U8C表体存货或编码相关的字段赋值不上
  • python-23-一篇文章帮你理解Python推导式
  • 【大模型】通过Crew AI 公司的崛起之路学习 AI Agents 的用法
  • 加入鲲鹏社区，领取精美礼品！
  • uniapp组件实现省市区三级联动选择
  • 用uio正确映射读写内存mmap
  • 如何在 Java 中使用 Canal 同步 MySQL 数据到 Redis
  • 数据结构——快速排序
  • 带你用Go实现二维码小游戏（下）
  • 一文了解git TAG
  • 基于Python+Vue开发的蛋糕商城管理系统
  • C++ 判断是不是平衡二叉树
  • 【fiddler】用fiddler实现手机抓包
  • 华为OD机试 - 学生排名（Java 2024 E卷 100分）
  • LLMs之PDF：zeroX(一款PDF到Markdown 的视觉模型转换工具)的简介、安装和使用方法、案例应用之详细攻略
  • move_base
  • D365 无法在数据被选择或插入到另一个事务作用域中的缓冲区上调用 NEXT、update（） 或 delete（）
  • Visual Studio Code从安装到正常使用
  • 在鱼皮的模拟面试里面学习有感
  • 代码中的设计模式-策略模式
  • LLMs之RAG：《LightRAG: Simple and Fast Retrieval-Augmented Generation》翻译与解读