RocketMQ第三章：手把手教老婆实现：普通消息（同步，异步和单向）的生产者和消费者

以你之姓@ 2022-11-17 15:00 161阅读 0赞

技术活，该赏
    关注+一键三连（点赞，评论，收藏）再看，养成好习惯

[RocketMQ使用教程相关系列 目录][RocketMQ_]

--------------------

**目录**

第一节：项目准备

第二节：普通消息-生产者和消息者步骤说明

普通消息生产者代码实现步骤

普通消息消费者代码实现步骤

第三节：普通消息-同步消息

生产者

效果：

消费者

效果：

第四节：普通消息-异步消息

生产者

效果：

消费者：

效果：

分析：

第五节：普通消息-单向消息

生产者

效果：

消费者

效果：

分析：

第六节：总结

--------------------

# 第一节：项目准备 #

rocketmq-client依赖

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

# 第二节：普通消息-生产者和消息者步骤说明 #

## 普通消息生产者代码实现步骤 ##

*  1.创建消息生产者producer，并制定生产者组名
 *  2.指定Nameserver地址
 *  3.启动producer
 *  4.创建消息对象 Message，指定主题Topic、Tag和消息体
 *  5.发送消息
 *  6.关闭生产者producer

## 普通消息消费者代码实现步骤 ##

*  1.创建消费者Consumer，制定消费者组名
 *  2.指定Nameserver地址
 *  3.订阅主题Topic和Tag
 *  4.设置回调函数，处理消息
 *  5.启动消费者consumer

注意：消费者的 Topic 和 Tag 需要和生产者保持一致

# 第三节：普通消息-同步消息 #

同步消息这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。

## 生产者 ##

public class ProducerSync {
    
       public static void main(String[] args) throws Exception {
          // 1.创建消息生产者producer，并制定生产者组名
          DefaultMQProducer producer = new DefaultMQProducer("demo_producer_group");
          // 2.指定Nameserver地址
          producer.setNamesrvAddr("192.168.88.131:9876");
          // 3.启动producer
          System.out.println("生产者启动");
          producer.start();
    
          for (int i = 0; i < 3; i++) {
             // 4.创建消息对象，指定主题Topic、Tag和消息体
             /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
             Message msg = new Message("Topic_demo_sync", "Tag_demo_sync",
                   ("Hello 虚竹，" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
             // 5.发送同步消息
             SendResult sendResult = producer.send(msg);
             System.out.println("发送结果：" + sendResult);
          }
    
          // 6.关闭生产者producer
          producer.shutdown();
          System.out.println("生产者关闭");
       }
    }

### 效果：  ###

![0][]

## 消费者  ##

public class ConsumerSync {
    
       public static void main(String[] args) throws Exception {
          // 1.创建消费者Consumer，制定消费者组名
          DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo_consumer_group");
          // 2.指定Nameserver地址
          consumer.setNamesrvAddr("192.168.88.131:9876");
    
          // 消息拉取最大条数
          consumer.setConsumeMessageBatchMaxSize(2);
          // 3.订阅主题Topic和Tag
          consumer.subscribe("Topic_demo_sync", "*");
    
          // 4.设置回调函数，处理消息
          consumer.registerMessageListener(new MessageListenerConcurrently() {
    
             // 接受消息内容
             @Override
             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
    
                   try {
                      // 获取主题
                      String topic = msg.getTopic();
                      // 获取标签
                      String tags = msg.getTags();
                      // 获取信息
                      byte[] body = msg.getBody();
                      String result = new String(body, RemotingHelper.DEFAULT_CHARSET);
                      System.out.println("Consumer消费信息：topic:" + topic + ",tags:" + tags + ",result：" + result);
    
                   } catch (UnsupportedEncodingException e) {
                      e.printStackTrace();
                   }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
             }
          });
          // 5.启动消费者consumer
          consumer.start();
       }
    }

### 效果： ###

![0][0 1]

# 第四节：普通消息-异步消息 #

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。

## 生产者 ##

public class ProducerASync {
    
       public static void main(String[] args) throws Exception {
          // 1.创建消息生产者producer，并制定生产者组名
          DefaultMQProducer producer = new DefaultMQProducer("demo_producer_group");
          // 2.指定Nameserver地址
          producer.setNamesrvAddr("192.168.88.131:9876");
          // 3.启动producer
          System.out.println("生产者启动");
          producer.start();
    
          for (int i = 0; i < 3; i++) {
             // 4.创建消息对象，指定主题Topic、Tag和消息体
             /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
             Message msg = new Message("Topic_demo_async", "Tag_demo_async",
                   ("Hello 虚竹，" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
             // 发送消息到一个Broker，异步发送没有返回值，需要使用 SendCallback 接收异步返回结果的回调
             producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                   System.out.println("发送成功：" + sendResult);
                }
    
                @Override
                public void onException(Throwable throwable) {
                   System.out.println("发送异常：" + throwable.getMessage());
                }
             });
    
          }
          Thread.sleep(2000);
          // 6.关闭生产者producer
          producer.shutdown();
          System.out.println("生产者关闭");
       }
    }

### 效果： ###

![0][0 2]

## 消费者：  ##

public class ConsumerASync {
    
       public static void main(String[] args) throws Exception {
          // 1.创建消费者Consumer，制定消费者组名
          DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo_consumer_group");
          // 2.指定Nameserver地址
          consumer.setNamesrvAddr("192.168.88.131:9876");
    
          // 消息拉取最大条数
          consumer.setConsumeMessageBatchMaxSize(2);
          // 3.订阅主题Topic和Tag
          consumer.subscribe("Topic_demo_async", "*");
    
          // 4.设置回调函数，处理消息
          consumer.registerMessageListener(new MessageListenerConcurrently() {
    
             // 接受消息内容
             @Override
             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
    
                   try {
                      // 获取主题
                      String topic = msg.getTopic();
                      // 获取标签
                      String tags = msg.getTags();
                      // 获取信息
                      byte[] body = msg.getBody();
                      String result = new String(body, RemotingHelper.DEFAULT_CHARSET);
                      System.out.println("Consumer消费信息：topic:" + topic + ",tags:" + tags + ",result：" + result);
    
                   } catch (UnsupportedEncodingException e) {
                      e.printStackTrace();
                   }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
             }
          });
          // 5.启动消费者consumer
          consumer.start();
       }
    }

### 效果： ###

![0][0 3]

### 分析： ###

从消费者代码中可以看出：异步消费者的实现方式和同步消费者实现方式并无区别。

从消费者的结果来看：可以看出异步消费者接收的结果是无序的。

# 第五节：普通消息-单向消息 #

这种方式主要用在不特别关心发送结果的场景，例如日志发送。

## 生产者 ##

public class Producer {
    
       public static void main(String[] args) throws Exception {
          // 1.创建消息生产者producer，并制定生产者组名
          DefaultMQProducer producer = new DefaultMQProducer("demo_producer_group");
          // 2.指定Nameserver地址
          producer.setNamesrvAddr("192.168.88.131:9876");
          // 3.启动producer
          producer.start();
          System.out.println("生产者启动");
          for (int i = 0; i < 20; i++) {
             // 4.创建消息对象，指定主题Topic、Tag和消息体
             /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
             Message msg = new Message("Topic_demo", "Tag_demo",
                   ("Hello 虚竹，" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
             // 5.发送单向消息
             producer.sendOneway(msg);
             System.out.println("发送结果：" + msg);
             // 线程睡1秒
          }
    
          // 6.关闭生产者producer
          producer.shutdown();
          System.out.println("生产者关闭");
       }
    }

### 效果： ###

![0][0 4]

## 消费者  ##

public class Consumer {
    
       public static void main(String[] args) throws Exception {
          // 1.创建消费者Consumer，制定消费者组名
          DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo_consumer_group");
          // 2.指定Nameserver地址
          consumer.setNamesrvAddr("192.168.88.131:9876");
    
          // 消息拉取最大条数
          consumer.setConsumeMessageBatchMaxSize(2);
          // 3.订阅主题Topic和Tag
          consumer.subscribe("Topic_demo", "*");
    
          // 4.设置回调函数，处理消息
          consumer.registerMessageListener(new MessageListenerConcurrently() {
    
             // 接受消息内容
             @Override
             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
    
                   try {
                      // 获取主题
                      String topic = msg.getTopic();
                      // 获取标签
                      String tags = msg.getTags();
                      // 获取信息
                      byte[] body = msg.getBody();
                      String result = new String(body, RemotingHelper.DEFAULT_CHARSET);
                      System.out.println("Consumer消费信息：topic:" + topic + ",tags:" + tags + ",result:" + result);
    
                   } catch (UnsupportedEncodingException e) {
                      e.printStackTrace();
                   }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
             }
          });
          // 5.启动消费者consumer
          consumer.start();
       }
    }

### 效果： ###

![0][0 5]

### 分析： ###

从消费结果来看，消费的顺序是无序的

# 第六节：总结 #

同步消息、异步消息和单向消息的消费者实现方式是一样的。

同步消息、异步消息和单向消息的区别在于消息的发送方。

异步消息生产者没有返回值，需要使用 SendCallback 接收异步返回结果的回调。

异步消息生产者，在关闭实例之前，建议进行休眠。

单向消息也是没有返回值的，并且它的消费者也是无序消费。

单向消息和异步消息的区别是单向消息不需要 SendCallback 来接收异步返回结果的回调。

[RocketMQ_]: https://blog.csdn.net/shi_hong_fei_hei/article/details/115470425
[0]: /images/20221022/5cd820b3991442128f8ad381f8e2c188.png
[0 1]: /images/20221022/325784fc3edd4d4f87cea5a5bb62c082.png
[0 2]: /images/20221022/df45a45bf2bb4ed3bff2e65679e47c5a.png
[0 3]: /images/20221022/2ea3b0109c3043b6a7617c46ae0b7ff8.png
[0 4]: /images/20221022/619520cfd7c8427eb3dd9505f6dde8a2.png
[0 5]: /images/20221022/73a5848abee749ab96f36bd478d7aa80.png