【MQ-02】基于SpringBoot的RocketMQ简单Demo(附源码)

左手的ㄟ右手 2023-06-23 03:59 54阅读 0赞

### 源码 ###

*  生产者：[https://github.com/HelloSummer5/RocketMQProducerDemo][https_github.com_HelloSummer5_RocketMQProducerDemo]
 *  消费者：[https://github.com/HelloSummer5/RocketMQConsumerDemo][https_github.com_HelloSummer5_RocketMQConsumerDemo]
 *  运行本Demo的前提是RocketMQ已经启动，详情参考[RocketMQ的安装][RocketMQ]

### 官方文档 ###

*  在csdn或者度娘search走两天弯路不如好好看看官方文档，以下是基于官方文档出的Demo。
 *  RocketMQ作为国产骄傲，支持下。  
    [http://rocketmq.apache.org/docs/quick-start/][http_rocketmq.apache.org_docs_quick-start]

### Maven依赖 ###

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

### 开发生产者Producer ###

*  在此为了方便查看效果，生产者消费者是分开两个项目的。

package com.example.demo.service;
    
    import io.netty.util.internal.ThreadExecutorMap;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.PreDestroy;
    
    /** * 生产者 */
    @Service
    public class ProducerService { 
    
        /** * 初始化 */
        public static void main(String[] args) throws Exception { 
            // 创建一个带组名的生产者实例
            DefaultMQProducer producer = new
                    DefaultMQProducer("please_rename_unique_group_name");
            // 指定MQ服务ID地址/域名
            producer.setNamesrvAddr("localhost:9876");
            // 启动生产者实例
            producer.start();
            for (int i = 0; i < 10; i++) { 
                // 创建消息实例, 指定topic, tag和消息体.
                Message msg = new Message("TopicTest",
                        "TagA",
                        ("Hello RocketMQ " +
                                i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                );
                // 调用发送消息方法来传递消息到其中一个broker
                SendResult sendResult = producer.send(msg);
    // System.out.printf("%s%n", sendResult);
                System.out.printf("发送消息的ID：%s, ---发送消息的状态：%s%n", sendResult.getMsgId(), sendResult.getSendStatus());
            }
            // 不常使用的话，关闭生产者实例
            producer.shutdown();
        }
    
    }

### 消费者 ###

package com.example.demo.servcie;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.PreDestroy;
    import java.util.List;
    
    /** * 消费者 */
    @Service
    public class ConsumerService { 
    
        public static void main(String[] args) throws MQClientException { 
            // 创建一个带组名的消费者实例
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
    
            // 指定MQ服务IP地址/域名
            consumer.setNamesrvAddr("localhost:9876");
    
            // 为消费者订阅一个或多个topics
            consumer.subscribe("TopicTest", "*");
    
            // 从broker拿到的消息之后执行的方法，do sth.
            consumer.registerMessageListener(new MessageListenerConcurrently() { 
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) { 
    // System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    for (MessageExt msg: msgs ) { 
                            System.out.println("收到消息:" + new String(msg.getBody()));
                        }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    
            // 启动消费者实例
            consumer.start();
    
            System.out.printf("消费者已准备就绪.%n");
        }
    
    }

### 运行 ###

*  先启动消费者，再启动生产者。

#### 消费者 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMTA2NjQ3_size_16_color_FFFFFF_t_70]

#### 生产者 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMTA2NjQ3_size_16_color_FFFFFF_t_70 1]

#### 启动生产者之后的消费者 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMTA2NjQ3_size_16_color_FFFFFF_t_70 2]

[https_github.com_HelloSummer5_RocketMQProducerDemo]: https://github.com/HelloSummer5/RocketMQProducerDemo
[https_github.com_HelloSummer5_RocketMQConsumerDemo]: https://github.com/HelloSummer5/RocketMQConsumerDemo
[RocketMQ]: https://blog.csdn.net/qq_32106647/article/details/103575221
[http_rocketmq.apache.org_docs_quick-start]: http://rocketmq.apache.org/docs/quick-start/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMTA2NjQ3_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191218174022221.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMTA2NjQ3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMTA2NjQ3_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191218174123263.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMTA2NjQ3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMTA2NjQ3_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191218174234611.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyMTA2NjQ3,size_16,color_FFFFFF,t_70