RocketMQ api 生产者和消费者例子

avatar 2019年08月04日17:06:22 6 9693 views
博主分享免费Java教学视频，B站账号：Java刘哥 ，长期提供技术问题解决、项目定制：本站商品点此
上文介绍了 RocketMQ 和可视化控制台的安装和启动，本文介绍通过 RocketMQ 官方的 api实现一个简单的生产者和消费者例子，因为是直接使用官方API，所以不需要 springboot 项目内容，直接在 main 方法里调用接口。

一、准备


启动 NameServer 和 broker

nohup sh bin/mqnamesrv >mqnamesrv-log.txt &


nohup sh bin/mqbroker -n 127.0.0.1:9876 >mqbroker-log.txt &


启动控制台


mvn spring-boot:run


创建一个 topic名为 test_quick_topic



创建一个 Maven 项目，添加依赖


  1. <dependency>
  2.     <groupId>com.alibaba</groupId>
  3.     <artifactId>fastjson</artifactId>
  4.     <version>1.2.54</version>
  5. </dependency>
  7. <dependency>
  8.     <groupId>org.apache.rocketmq</groupId>
  9.     <artifactId>rocketmq-client</artifactId>
  10.     <version>4.3.0</version>
  11. </dependency>

二、生产者例子


1.例1：同步投递10条消息


  1. package com.liuyanzhao.rocketmq.demo.test;
  3. import org.apache.rocketmq.client.exception.MQBrokerException;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  6. import org.apache.rocketmq.client.producer.SendResult;
  7. import org.apache.rocketmq.common.message.Message;
  8. import org.apache.rocketmq.remoting.exception.RemotingException;
  10. /**
  11.  * @author 言曌
  12.  * @date 2019-08-03 16:19
  13.  */
  15. public class TestProducer {
  17.     public static final String NAMESRV_ADDR = "127.0.0.1:9876";
  20.     public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
  21.         DefaultMQProducer producer = new DefaultMQProducer("test_quick_producer_name");
  22.         producer.setNamesrvAddr(NAMESRV_ADDR);
  23.         producer.start();
  25.         for (int i = 0; i <10; i++) {
  26.             Message message = new Message("test_quick_topic",//主题
  27.                     "tagA", //标签
  28.                     "key" + i, //自定义key，唯一标识
  29.                     ("Hello RocketMQ=" + i).getBytes()); //消息内容实体 (byte[])
  31.             SendResult result = producer.send(message);
  32.             System.out.println("第" + i + "条消息发出，结果：" + result);
  33.         }
  35.         producer.shutdown();
  36.     }
  37. }

控制台输出





控制台也可以看到有10条消息





例2：异步投递消息、延迟投递消息和自定义投递到指定队列
  1. package com.liuyanzhao.rocketmq.demo.test;
  3. import org.apache.rocketmq.client.exception.MQBrokerException;
  4. import org.apache.rocketmq.client.exception.MQClientException;
  5. import org.apache.rocketmq.client.producer.DefaultMQProducer;
  6. import org.apache.rocketmq.client.producer.MessageQueueSelector;
  7. import org.apache.rocketmq.client.producer.SendCallback;
  8. import org.apache.rocketmq.client.producer.SendResult;
  9. import org.apache.rocketmq.common.message.Message;
  10. import org.apache.rocketmq.common.message.MessageQueue;
  11. import org.apache.rocketmq.remoting.exception.RemotingException;
  13. import java.util.List;
  15. /**
  16.  * @author 言曌
  17.  * @date 2019-08-03 16:19
  18.  */
  20. public class TestProducer {
  22.     public static final String NAMESRV_ADDR = "127.0.0.1:9876";
  25.     public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
  26.         DefaultMQProducer producer = new DefaultMQProducer("test_quick_producer_name");
  27.         producer.setNamesrvAddr(NAMESRV_ADDR);
  28.         producer.start();
  30.         for (int i = 0; i < 10; i++) {
  31.             //1.创建消息
  32.             Message message = new Message("test_quick_topic",//主题
  33.                     "tagA", //标签
  34.                     "key" + i, //自定义key，唯一标识
  35.                     ("Hello RocketMQ=" + i).getBytes()); //消息内容实体 (byte[])
  37.             //3.延迟投递消息
  38.             if (i == 2) {
  39.                 //延迟10s投递
  40.                 message.setDelayTimeLevel(3);
  41.             }
  43.             //4.自定义投递：将消息发送到指定的队列,如发到第2个队列
  44.             producer.send(message, new MessageQueueSelector() {
  45.                 @Override
  46.                 public MessageQueue select(List<MessageQueue> list, Message message, Object args) {
  47.                     Integer queueNumber = (Integer) args;
  48.                     return list.get(queueNumber);
  49.                 }
  50.             }, 2);  //这里的最后一个参数2就是传到前面匿名方法的 args 的值
  52.             //2.1 同步发生消息
  53.             // SendResult result = producer.send(message);
  54.             // System.out.println("第" + i + "条消息发出，结果：" + result);
  56.             //2.2 异步发送消息
  57.             producer.send(message, new SendCallback() {
  58.                 @Override
  59.                 public void onSuccess(SendResult sendResult) {
  60.                     System.out.println("msgId:" + sendResult.getMsgId() + ", status:" + sendResult.getSendStatus());
  61.                 }
  63.                 @Override
  64.                 public void onException(Throwable throwable) {
  65.                     throwable.printStackTrace();
  66.                     System.out.println("--发送失败--");
  67.                 }
  68.             });
  69.         }
  71.         //producer.shutdown();
  72.     }
  73. }




二、消费者例子


1.例1：消费上面生产者生产的10条消息
  1. package com.liuyanzhao.rocketmq.demo.test;
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  5. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  6. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  7. import org.apache.rocketmq.client.exception.MQClientException;
  8. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  9. import org.apache.rocketmq.common.message.MessageExt;
  10. import org.apache.rocketmq.remoting.common.RemotingHelper;
  12. import java.util.List;
  14. /**
  15.  * @author 言曌
  16.  * @date 2019-08-04 13:31
  17.  */
  19. public class TestConsumer {
  21.     public static final String NAMESRV_ADDR = "127.0.0.1:9876";
  24.     public static void main(String[] args) throws MQClientException {
  25.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_quick_consumer_name");
  26.         consumer.setNamesrvAddr(NAMESRV_ADDR);
  27.         //从最后开始消费
  28.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  29. //        consumer.subscribe("test_quick_topic","tagA"); //过滤：消费tag为tagA的消息
  30.         consumer.subscribe("test_quick_topic", "*"); //消费所有的
  32.         consumer.registerMessageListener(new MessageListenerConcurrently() {
  33.             @Override
  34.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
  35.                 MessageExt messageExt = list.get(0);
  36.                 try {
  37.                     String topic = messageExt.getTopic();
  38.                     String tags = messageExt.getTags();
  39.                     String keys = messageExt.getKeys();
  40.                     String msgBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
  41.                     System.out.println("topic: " + topic + ",tags: " + tags + ", keys: " + keys + ", body: " + msgBody);
  42.                 } catch (Exception e) {
  43.                     e.printStackTrace();
  44.                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;
  45.                 }
  46.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  47.             }
  48.         });
  50.         consumer.start();
  51.         System.out.println("comsumer start");
  53.     }
  54. }

控制台输出





2.例2，模拟消费时发生异常
  1. package com.liuyanzhao.rocketmq.demo.test;
  3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
  5. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
  6. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  7. import org.apache.rocketmq.client.exception.MQClientException;
  8. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  9. import org.apache.rocketmq.common.message.MessageExt;
  10. import org.apache.rocketmq.remoting.common.RemotingHelper;
  12. import java.util.List;
  14. /**
  15.  * @author 言曌
  16.  * @date 2019-08-04 13:31
  17.  */
  19. public class TestConsumer {
  21.     public static final String NAMESRV_ADDR = "127.0.0.1:9876";
  24.     public static void main(String[] args) throws MQClientException {
  25.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_quick_consumer_name");
  26.         consumer.setNamesrvAddr(NAMESRV_ADDR);
  27.         //从最后开始消费
  28.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
  29. //        consumer.subscribe("test_quick_topic","tagA"); //过滤：消费tag为tagA的消息
  30.         consumer.subscribe("test_quick_topic", "*"); //消费所有的
  32.         consumer.registerMessageListener(new MessageListenerConcurrently() {
  33.             @Override
  34.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
  35.                 MessageExt messageExt = list.get(0);
  36.                 try {
  38.                     String id = messageExt.getMsgId();
  39.                     String topic = messageExt.getTopic();
  40.                     String tags = messageExt.getTags();
  41.                     String keys = messageExt.getKeys();
  42.                     String msgBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
  45.                     //模拟失败
  46.                     if ("key5".equals(keys)) {
  47.                         System.out.println(1 / 0);
  48.                     }
  50.                     System.out.println("topic: " + topic + ",tags: " + tags + ", id:" + id + ",keys: " + keys + ", body: " + msgBody);
  53.                 } catch (Exception e) {
  54.                     //失败后，默认会隔 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 重试
  56.                     e.printStackTrace();
  57.                     int reconsumeTimes = messageExt.getReconsumeTimes();//失败重试次数
  58.                     if (reconsumeTimes == 3) {
  59.                         //记录日志 ...
  60.                         System.out.println("消息ID:" + messageExt.getMsgId() + "失败三次，执行补偿策略");
  61.                         //做补偿处理
  62.                         //TODO
  63.                         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  65.                     }
  66.                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;
  67.                 }
  68.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  69.             }
  70.         });
  72.         consumer.start();
  73.         System.out.println("comsumer start");
  75.     }
  76. }



然后重新启动生产者生产10条消息，然后启动消费者

消费的顺序是随机的，当消费到 key 为 key5的消息，会抛一个异常，进入 catch 代码块

reconsumeTimes 是重试次数，mq自己记录，返回 ConsumeConcurrentlyStatus.RECONSUME_LATER; 表示需要重试

重试机制是每隔  1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 重试一次

这里我们判断当重试第3次时，做补偿处理，记录日志，然后直接返回消费成功

控制台输出





历史上的今天
八月
04日
  • 微信
  • 交流学习，资料分享
  • weinxin
  • 个人淘宝
  • 店铺名：言曌博客咨询部

  • (部分商品未及时上架淘宝)
avatar

发表评论

avatar 登录者：匿名
匿名评论，评论回复后会有邮件通知

  

已通过评论：0   待审核评论数：0