SpringBoot整合RocketMQ实现消息队列

妖狐艹你老母 2022-11-28 00:51 270阅读 0赞

## [RocketMQ][] ##

Apache RockerMQ是一个分布式消息传递引擎，轻量级的数据处理平台。  
它具有以下的特性:

*  低延迟： 高压下在1毫米内响应延迟超过99.6%。
 *  财务导向：具有跟踪和审核功能的高可用性。
 *  产业可持续发展：万亿级消息容量。
 *  大数据友好：具有通用集成功能的批处理传输实现大批量吞吐。
 *  兼容性好：提供开放式分布式消息传递和流传输标准。
 *  万亿级容量：给定足够的磁盘空间，可以在不损失性能的情况下累计邮件。

## RocketMQ搭建与配置 ##

RocketMQ搭建与配置可参考我的博客：[RocketMQ搭建与配置][RocketMQ 1]

## SpringBoot整合RocketMQ实现消息队列 ##

**引入依赖**

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

**JmsConfig rockerMQ配置类**

public class JmsConfig { 
    
        //RocketMQServer地址
        public static final String NAME_SERVER = "127.0.0.1:9876";
        /** * 主题名称 */
        public static final String TOPIC = "topic_family";
    }

**Producer生产者实体类**

@Slf4j
    @Component
    public class Producer { 
        private String producerGroup = "test_producer";
        private DefaultMQProducer producer;
    
    
        public Producer(){ 
            //示例生产者
            producer = new DefaultMQProducer(producerGroup);
            //不开启vip通道 开通口端口会减2
            producer.setVipChannelEnabled(false);
            //绑定name server
            producer.setNamesrvAddr(JmsConfig.NAME_SERVER);
            start();
        }
        /** * 对象在使用之前必须要调用一次，只能初始化一次 */
        public void start(){ 
            try { 
                this.producer.start();
            } catch (MQClientException e) { 
                e.printStackTrace();
            }
        }
    
    
        public DefaultMQProducer getProducer(){ 
            return this.producer;
        }
        /** * 一般在应用上下文，使用上下文监听器，进行关闭 */
        public void shutdown(){ 
            this.producer.shutdown();
        }
    }

**Consumer消费者实体类**

@Slf4j
    @Component
    public class Consumer { 
    
    
        /** * 消费者实体对象 */
        private DefaultMQPushConsumer consumer;
        /** * 消费者组 */
        public static final String CONSUMER_GROUP = "test_consumer";
        /** * 通过构造函数 实例化对象 */
        public Consumer() throws MQClientException { 
    
    
            consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
            consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
            //消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            //订阅主题和 标签（ * 代表所有标签)下信息
            consumer.subscribe(JmsConfig.TOPIC, "*");
            // //注册消费的监听 并在此监听中消费信息，并返回消费的状态信息
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { 
                // msgs中只收集同一个topic，同一个tag，并且key相同的message
                // 会把不同的消息分别放置到不同的队列中
                try { 
                    for (Message msg : msgs) { 
    
    
                        //消费者获取消息 这里只输出 不做后面逻辑处理
                        String body = new String(msg.getBody(), "utf-8");
                        log.info("Consumer-获取消息-主题topic为={}, 消费消息为={}", msg.getTopic(), body);
                    }
                } catch (UnsupportedEncodingException e) { 
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
    
    
            consumer.start();
            System.out.println("消费者 启动成功=======");
        }
    }

**UserController消息验证**

@RestController
    public class UserController { 
    
        private static Logger logger = LogManager.getLogger(LogManager.ROOT_LOGGER_NAME);
        @Autowired
        Producer producer;
    
        @RequestMapping("/hello")
        public void hello()throws Exception{ 
    
    
            for(int i=1;i<10;i++){ 
                //创建生产信息
                Message message = new Message(JmsConfig.TOPIC, "testtag", ("Hello World"+i).getBytes());
                //发送
                SendResult sendResult = producer.getProducer().send(message);
                logger.info("输出生产者信息={}",sendResult);
            }
        }
    }

**访问：[localhost:8080/hello][localhost_8080_hello] 获取一下结果**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 1]

[RocketMQ]: http://rocketmq.apache.org/
[RocketMQ 1]: https://blog.csdn.net/weixin_40990818/article/details/107997923
[localhost_8080_hello]: http://localhost:8080/hello
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/3dcf587b377445c08d2f1148773b8e68.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDk5MDgxOA_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221124/334a9e7aede94b29afbf3292c7111933.png