初始化v1.2

2025-11-24 16:19:09 +08:00
parent c0f51ef007
commit de49404116
918 changed files with 4236 additions and 4239 deletions
@@ -0,0 +1,77 @@
+package com.tongran.rocketmq.config;
+
+import com.tongran.rocketmq.consumer.RocketMsgListener;
+import com.tongran.rocketmq.enums.MessageTopic;
+import com.tongran.rocketmq.model.ConsumerMode;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.context.config.annotation.RefreshScope;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.List;
+
+/**
+ * 消费者配置
+ */
+@RefreshScope
+@Configuration
+@Slf4j
+public class ConsumerConfig {
+
+    @Autowired
+    private ConsumerMode consumerMode;
+
+    @Autowired
+    private RocketMsgListener rocketMsgListener;
+
+    @Bean
+    public DefaultMQPushConsumer getRocketMQConsumer() {
+        //构建客户端连接
+        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerMode.getAgentGroup());
+        //
+        consumer.setNamesrvAddr(consumerMode.getNamesrvAddr());
+        consumer.setConsumeThreadMin(consumerMode.getConsumeThreadMin());
+        consumer.setConsumeThreadMax(consumerMode.getConsumeThreadMax());
+        consumer.registerMessageListener(rocketMsgListener);
+        /**
+         * 1. CONSUME_FROM_LAST_OFFSET：第一次启动从队列最后位置消费，后续再启动接着上次消费的进度开始消费
+         * 2. CONSUME_FROM_FIRST_OFFSET：第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费
+         * 3. CONSUME_FROM_TIMESTAMP：第一次启动从指定时间点位置消费，后续再启动接着上次消费的进度开始消费
+         *  以上所说的第一次启动是指从来没有消费过的消费者，如果该消费者消费过，那么会在broker端记录该消费者的消费位置，如果该消费者挂了再启动，那么自动从上次消费的进度开始
+         */
+        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
+        /**
+         * CLUSTERING (集群模式) ：默认模式，同一个ConsumerGroup(groupName相同)每个consumer只消费所订阅消息的一部分内容，同一个ConsumerGroup里所有的Consumer消息加起来才是所
+         *  订阅topic整体，从而达到负载均衡的目的
+         * BROADCASTING (广播模式) ：同一个ConsumerGroup每个consumer都消费到所订阅topic所有消息，也就是一个消费会被多次分发，被多个consumer消费。
+         * 需要注意的是，在广播模式下，每个Consumer都会独立地处理相同的消息副本。这可能会导致一些潜在的问题，例如消息重复处理或者资源浪费。因此，在使用广播模式时，请确保消息的处理逻辑是幂等的，并仔细考虑系统资源的消耗。
+         */
+        //        consumer.setMessageModel(MessageModel.BROADCASTING);
+
+        consumer.setVipChannelEnabled(false);
+        consumer.setConsumeMessageBatchMaxSize(consumerMode.getConsumeMessageBatchMaxSize());
+        try {
+            /**
+             * 订阅topic，可以对指定消息进行过滤，例如："TopicTest","tagl||tag2||tag3",*或null表示topic所有消息
+             * 由于官方并没有给直接订阅全部消息示例 所以使用list列表循环订阅所有topic
+             */
+            // 获取所有topic列表
+            MessageTopic messageTopic = new MessageTopic();
+            List<String> allTopics = messageTopic.RocketMQTopicList();
+            //订阅所有topic
+            for (String topic : allTopics) {
+                consumer.subscribe(topic,"*");
+            }
+            consumer.start();
+            log.info("消费者初始化成功：{}", consumer);
+        } catch (MQClientException e) {
+            e.printStackTrace();
+            log.error("消费者初始化失败：{}",e.getMessage());
+        }
+        return consumer;
+    }
+}
@@ -0,0 +1,56 @@
+package com.tongran.rocketmq.config;
+
+import com.tongran.rocketmq.model.ProducerMode;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+
+/**
+ * mq搭建地址连接
+ * 生产者初者连接信息 具体看nacos配置
+ */
+@Configuration
+@Slf4j
+public class ProducerConfig {
+
+    /**
+     * 远程调用连接信息
+     */
+    public static DefaultMQProducer producer;
+
+    /**
+     * 连接客户端信息配置  具体看nacos配置
+     */
+    @Autowired
+    private ProducerMode producerMode;
+
+    @Bean
+    public DefaultMQProducer getRocketMQProducer() {
+        producer = new DefaultMQProducer(producerMode.getAgentGroup());
+        producer.setNamesrvAddr(producerMode.getNamesrvAddr());
+        //如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
+        if(producerMode.getMaxMessageSize()!=null){
+            producer.setMaxMessageSize(producerMode.getMaxMessageSize());
+        }
+        if(producerMode.getSendMsgTimeout()!=null){
+            producer.setSendMsgTimeout(producerMode.getSendMsgTimeout());
+        }
+        //如果发送消息失败，设置重试次数，默认为2次
+        if(producerMode.getRetryTimesWhenSendFailed()!=null){
+            producer.setRetryTimesWhenSendFailed(producerMode.getRetryTimesWhenSendFailed());
+        }
+        producer.setVipChannelEnabled(false);
+        try {
+            producer.start();
+            log.info("生产者初始化成功：{}",producer.toString());
+        } catch (MQClientException e) {
+            log.error("生产者初始化失败：{}",e.getMessage());
+        }
+        return producer;
+    }
+
+}