rockerMq实现消费者动态上下线

叁歲伎倆 2022-11-29 13:05 374阅读 0赞

## 需求： ##

1.通过接口方式动态控制消费者的上线和下线，通过配置的方式，实现对接不同rocketmq系统的不同消息队列的消费；

2.从消息队列中读取到消息后，将对映值映射到实现配置好的数据模板中，转换为模板格式的数据；

## **直切主题，实现动态上下线的两个主要类：** ##

1.配置类，存储消费者连接对象，可以扩展为redis

<dependency>
    			<groupId>org.apache.rocketmq</groupId>
    			<artifactId>rocketmq-client</artifactId>
    			<version>4.3.0</version>
    		</dependency>

/**
     * 消费者相关配置
     *
     * @author qiaofeng
     */
    public class ConsumerConfig {
    
        /**
         * 存储消费者对象，可以扩展为redis
         */
        public static Map<String, DefaultMQPushConsumer> consumerMap = new ConcurrentHashMap<>();
    }

2.实现类

/**
         * 创建对应的消费者服务
         * @param addrIp 消息队列服务ip:端口
         * @param rocketTopicMsgModel 消费者需要消费的队列
         */
        private DefaultMQPushConsumer addConsumerIntoMap(String addrIp, RocketTopicMsgModel rocketTopicMsgModel) throws MQClientException {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketTopicMsgModel.getConsumerGroup());
            consumer.setNamesrvAddr(addrIp);
            consumer.subscribe(rocketTopicMsgModel.getTopicName(), rocketTopicMsgModel.getSubExpression());
            consumer.setMessageModel(rocketTopicMsgModel.getMsgModelType() == 1 ? MessageModel.CLUSTERING : MessageModel.BROADCASTING);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    try {
                        if (msgs.size() > 0){
                            MessageExt messageExt = msgs.get(0);
                            consumerMsg(rocketTopicMsgModel.getMsgModel(), new String(messageExt.getBody(), StandardCharsets.UTF_8));
                        }
                    }catch (Throwable e){
                        log.error("发生异常:{}", e.getMessage());
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
    
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 启动消费者实例
            consumer.start();
    ConsumerConfig.consumerMap.put(rocketTopicId, mQPushConsumer);
            return consumer;
        }

## **思路** ##

**将每次创建的consumer对象存起来，通过接口方式启动或停止；本人实现使用了数据库，可在数据库动态配置不同的rocketMq服务地址、不同服务器地址下的消费者信息、对映需要转换成的模板**

**git地址：**[https://github.com/pengxingjia/rocketMq-consumer.git][https_github.com_pengxingjia_rocketMq-consumer.git]

笔记完结！

[https_github.com_pengxingjia_rocketMq-consumer.git]: https://github.com/pengxingjia/rocketMq-consumer.git