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

合肥公司网站建设学校教育培训机构

news 2025/7/14 18:16:36
合肥公司网站建设,学校教育培训机构,电脑网页视频怎么下载,wordpress最新免费主题下载地址目录 前言一、情景介绍二、问题分析三、代码实现 前言 之前接到一个需求，我们项目的技术负责人希望通过配置的形式，在项目启动的时候自动根据配置生成对应的消费者 觉得还有点意思，随即记录一下~ 一、情景介绍 比如我这里有一个消费者 Mes…

目录

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

前言

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

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

一、情景介绍

比如我这里有一个消费者 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.dinnco.com/news/47029.html

相关文章：

  • 开发一个app软件的公司济南网站优化
  • 企业网站建设心得水果网络营销策划书
  • 校园网站建设资金来源有建网站需要什么
  • 怎样用一台电脑做代理 让别的电脑通过代理上几个网站百度竞价排名事件分析
  • 怎么做传奇私服网站品牌广告投放
  • 韶关公司做网站博客推广的方法与技巧
  • 聊城做网站的公司机构制作自己的网页
  • seo网站优化及网站推广百度网址大全下载安装
  • 网站如何做点击链接全网推广费用
  • 楚雄市城乡建设局网站app推广员好做吗
  • 广州网站建设设计哪家好培训学校机构有哪些
  • wordpress html 静态化免费seo营销优化软件下载
  • 龙岩小程序报价网站优化方案怎么写
  • 做门户网站用什么系统好托管竞价推广公司
  • 苏州企业网站建设网站seo优化报告
  • 百度怎么免费做网站百度移动端排名
  • 电商网站建设好么百度关键词怎么排名
  • 做游戏的网站天津百度seo排名优化软件
  • 网站英文专门制作小程序的公司
  • 网站建设原因分析苏州seo排名公司
  • 怎么做网站滑动图片部分h5怎样免费给自己的公司做网站
  • 网页版面设计包括什么内容青岛seo网站推广
  • 做试题的网站百度客服中心
  • 宁都网站建设百度网站官网入口
  • 天翼云电脑免费领取河北百度seo
  • 武威做网站it培训班真的有用吗
  • 工程网站建设谷歌搜索引擎入口2022
  • 可以下载各种软件的网站永久免费建个人网站
  • 卫浴外贸版网站案例赣州seo公司
  • 日照建设信息网站北京网站seo技术厂家