经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 数据库/运维 » RocketMQ » 查看文章
RocketMQ的简单使用
来源：cnblogs　　作者：Leo哥coding~　　时间：2023/5/4 9:20:13　　对本文有异议

大家好，我是Leo！今天来和大家分享RocketMQ的一些用法。

领域模型介绍

Producer: 用于生产消息的运行实体。

Topic: 主题，用于消息传输和存储的分组容器。

MessageQueue: 消息传输和存储的实际单元容器。

Message: 消息传输的最小单元。

ConsumerGroup: 消费者组。

Consumer: 消费者。

Subscription: 订阅关系，发布订阅模式中消息过滤、重试、消费进度的规则配置。

MQ的优势

MQ的明显优势有3个。

应用解耦： 以多服务为例，用户下单，需要通知订单服务和库存服务，我们可以通过MQ消息来解除下单和库存系统的耦合。

异步提速： 以秒杀为例，我们可以先返回秒杀结果，后续再通过MQ异步消息去插入记录和扣减库存等，减少调用的链路长度。

削峰填谷： 将某一时间内的请求量分摊到更多时间处理，比如系统A一秒只能处理10000个请求，但是我有100000个请求需要处理，我可以将请求发到MQ中，再分成10秒去消费这些请求。

当然MQ也有劣势，系统可用性降低，系统复杂度提高，一致性问题。

RocketMQ的主要角色

主要包括Producer、Broker、Consumer、NameServer Cluster。

一对多

可以通过设置不同的消费者组

不同组通过不同的消费者组既可以实现同时收到一样数量的消息，那同一个消费者组需要怎样才能收到同样数量的消息呢？

  1. // 消费者消费模式
  2. consumer.setMessageModel(MessageModel.BROADCASTING);

默认是集群模式CLUSTERING，设置成广播模式

既可以实现一对多的发送。

同步消息（普通消息）

同步消息需要阻塞等待消息发送结果的返回

  1. public class ProducerDemo {
  3.     public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
  5.         DefaultMQProducer producer = new DefaultMQProducer("group1");
  7.         producer.setNamesrvAddr("localhost:9876");
  8.         producer.start();
  9.         Message message = new Message();
  10.         message.setTopic("MQLearn");
  11.         message.setTags("1.0.0");
  12.         message.setBody("Hello MQ!".getBytes(StandardCharsets.UTF_8));
  13.         SendResult result = producer.send(message);
  14.         if (result.getSendStatus().equals(SendStatus.SEND_OK)) {
  15.             System.out.println(result);
  16.             System.out.println("发送成功：" + message);
  17.         }
  19.         producer.shutdown();
  20.     }
  21. }

异步消息

异步消息需要实现发送成功和失败的回调函数。

  1. public class Producer {
  3.     public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
  5.         DefaultMQProducer producer = new DefaultMQProducer("group1");
  7.         producer.setNamesrvAddr("192.168.246.140:9876");
  8.         producer.start();
  9.         // 异步消息
  10.         for (int i = 0; i < 10; i++) {
  11.             Message message = new Message();
  12.             message.setTopic("topic7");
  13.             message.setTags("1.0.0");
  15.             message.setBody(("Hello World !" + i).getBytes(StandardCharsets.UTF_8));
  16.             producer.send(message, new SendCallback() {
  17.                 @Override
  18.                 public void onSuccess(SendResult sendResult) {
  19.                     // 发送成功的回调方法
  20.                     System.out.println(sendResult);
  21.                 }
  23.                 @Override
  24.                 public void onException(Throwable e) {
  25.                     // 发送失败的回调方法
  26.                     System.out.println(e);
  27.                 }
  28.             });
  29.         }
  30.         TimeUnit.SECONDS.sleep(10);
  31.         System.out.println("异步发送完成！");
  32.     }
  33. }

单向消息

单向消息就类似UDP，只顾单向发送，不管是否发送成功，常用于日志收集等场景。

  1. public class SingleDirectionProducer {
  3.     public static void main(String[] args) throws Exception {
  4.         DefaultMQProducer producer = new DefaultMQProducer("group1");
  6.         producer.setNamesrvAddr("localhost:9876");
  7.         producer.start();
  8.         // 单向消息
  9.         for (int i = 0; i < 10; i++) {
  10.             Message message = new Message();
  11.             message.setTopic("topic8");
  12.             message.setTags("1.0.0");
  13.             message.setBody(("Hello World !" + i).getBytes(StandardCharsets.UTF_8));
  14.             producer.sendOneway(message);
  15.         }
  16.         System.out.println("带向发送完成！");
  17.     }
  18. }

延时（定时）消息

RocketMQ提供的定时消息并不能指定在什么时间点去投递消息。而是根据设定的等待时间，起到延时到达的缓冲作用在RocketMQ中,延时消息的delayTimeLevel支持以下级别:

1 1s 2 5s 3 10s 4 30s 5 1m 6 2m 7 3m 8 4m 9 5m 10 6m 11 7m 12 8m 13 9m 14 10m 15 20m 16 30m 17 1h 18 2h

  1. // 设置消息延时级别
  2. message.setDelayTimeLevel(3);

批量消息

批量消息支持一次发送多条消息。

注意：

  • 批量消息需要有相同的topic

  • 不能是延时消息

  • 消息内容不能超过4M，可以通过producer.setMaxMessageSize()和broker进行设置设置（可以通过拆分多次发送）

  1. public class BatchProducer {
  3.     public static void main(String[] args) throws Exception {
  4.         DefaultMQProducer producer = new DefaultMQProducer("group1");
  6.         producer.setNamesrvAddr("localhost:9876");
  7.         producer.start();
  8.         List<Message> list = new ArrayList<>();
  9.         // 批量消息
  10.         for (int i = 0; i < 10; i++) {
  11.             Message message = new Message();
  12.             message.setTopic("topic10");
  13.             message.setTags("1.0.0");
  14.             message.setBody(("Hello World !" + i).getBytes(StandardCharsets.UTF_8));
  15.             list.add(message);
  16.         }
  17.         SendResult result = producer.send(list);
  18.         System.out.println(result);
  19.         TimeUnit.SECONDS.sleep(2);
  20.         System.out.println("发送完成！");
  21.     }
  22. }

顺序消息

顺序消息支持按照消息的发送消息先后获取消息。

比如：我的一笔订单有多个流程需要处理，比如创建->付款->推送->完成。

通过同一笔订单放到一个队列中，这样就可以解决消费的无序问题。

通过实现MessageQueueSelector来选择一个队列。

  1. public class Producer {
  3.     public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
  5.         DefaultMQProducer producer = new DefaultMQProducer("group1");
  7.         producer.setNamesrvAddr("localhost:9876");
  8.         producer.start();
  10.             Message message = new Message();
  11.             // 模拟业务ID
  12.             int step = 10;
  13.             message.setTopic("topic12");
  14.             message.setTags("1.0.0");
  15.             message.setBody(("Hello World !").getBytes(StandardCharsets.UTF_8));
  16.             producer.send(message, new MessageQueueSelector() {
  17.                 @Override
  18.                 public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
  19.                     // 队列数
  20.                     int size = mqs.size();
  21.                     // 取模
  22.                     int orderId = step;
  23.                     return mqs.get(orderId % size);
  24.                 }
  25.             }, null);
  27.         System.out.println("发送完成！");
  28.     }
  29. }
  1. public class Consumer {
  3.     public static void main(String[] args) throws Exception{
  5.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
  6.         consumer.setConsumerGroup("group1");
  7.         consumer.setNamesrvAddr("localhost:9876");
  8.         consumer.subscribe("topic12", "*");
  9.         // 消费者，起一个顺序监听，一个线程，只监听一个队列
  10.         consumer.registerMessageListener(new MessageListenerOrderly() {
  11.             @Override
  12.             public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
  13.                 for (MessageExt msg : msgs) {
  14.                     System.out.println(msg);
  15.                     byte[] body = msg.getBody();
  16.                     System.out.println(new String(body));
  17.                 }
  18.                 return ConsumeOrderlyStatus.SUCCESS;
  19.             }
  20.         });
  21.         consumer.start();
  22.         System.out.println("消费者启动了！");
  24.     }
  25. }

事务消息

RocketMQ中的事务消息支持在分布式场景下消息生产和本地事务的最终一致性。

大致流程为，

  1. 生产者先将消息发送至RocketMQ。

  2. RocketMQBroker将消息持久化成功后，向生产者返回ACK消息确认已经返回成功，消息状态为暂时不能投递状态。

  3. 执行本地事务逻辑。

  4. 生产者根据事务执行结果向Broker提交commit或者rollback结果。

  5. 如果在断网或者重启情况下，未收到4的结果，或者返回Unknown未知状态，在固定时间对消息进行回查。

  6. 生产者收到消息回查后，需要本地事务执行的最终结果。

  7. 生产者对本地事务状态进行二次提交或确认。

  1. public class Producer {
  3.     public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
  4.         TransactionMQProducer producer = new TransactionMQProducer("group1");
  5.         producer.setNamesrvAddr("localhost:9876");
  6.         // 设置事务监听
  7.         producer.setTransactionListener(new TransactionListener() {
  8.             // 正常事务监听
  9.             @Override
  10.             public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
  11.                 // 把消息保存到mysql数据库
  12.                 boolean ok = false;
  14.                 if (ok) {
  15.                     System.out.println("正常执行事务过程");
  16.                     return LocalTransactionState.COMMIT_MESSAGE;
  17.                 } else {
  18.                     System.out.println("事务补偿过程");
  19.                     return LocalTransactionState.UNKNOW;
  20.                     //return LocalTransactionState.ROLLBACK_MESSAGE;
  21.                 }
  23.             }
  24.             // 事务补偿事务
  25.             @Override
  26.             public LocalTransactionState checkLocalTransaction(MessageExt msg) {
  27.                 System.out.println("事务补偿过程");
  28.                 // sql select
  29.                 if (true) {
  31.                 } else {
  33.                 }
  34.                 return LocalTransactionState.COMMIT_MESSAGE;
  35.             }
  36.         });
  37.         producer.start();
  38.         String msg = "Hello Transaction Message!";
  39.         Message message = new Message("topic13", "tag", msg.getBytes(StandardCharsets.UTF_8));
  40.         TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(message, null);
  41.         TimeUnit.SECONDS.sleep(2);
  42.         System.out.println(transactionSendResult);
  43.         System.out.println("发送完成！");
  44.     }
  45. }

消息的过滤

在RocketMQ中的消息过滤功能能通过生产者和消费者对消息的属性和Tag进行定义，在消费端可以根据过滤条件进行筛选匹配，将符合条件的消息投递给消费者进行消费。

支持两种方式：Tag标签过滤和SQL属性过滤。

  1. Message message = new Message();
  2. message.setTopic("topic11");
  3. message.setTags("tag");
  4. message.setBody(("Hello World !" + "tag").getBytes(StandardCharsets.UTF_8));
  5. message.putUserProperty("name", "zhangsan");
  6. message.putUserProperty("age", "16");
  7. SendResult result = producer.send(message);

subscribe方法subExpression参数也支持Tag过滤

  1. DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
  2. consumer.setConsumerGroup("group1");
  3. consumer.setNamesrvAddr("112.74.125.184:9876");
  4. consumer.subscribe("topic11", MessageSelector.bySql("age > 16"));

SpringBoot整合RocketMQ的使用

在SpringBoot项目中主要通过RocketMQTemplate进行消息的发送。

  1. // 普通消息
  2. rocketMQTemplate.convertAndSend("topic10", user);
  3. rocketMQTemplate.send("topic10", MessageBuilder.withPayload(user).
  4. SendResult result = rocketMQTemplate.syncSend("topic10", user);
  5. // 异步消息
  6. rocketMQTemplate.asyncSend("topic10", user, new SendCallback() {
  7.     @Override
  8.     public void onSuccess(SendResult sendResult) {
  9.         System.out.println("成功！");
  10.     }
  11.     @Override
  12.     public void onException(Throwable e) {
  13.         System.out.println(e);
  14.     }
  15. }, 1000L);
  16. // 单向消息
  17. rocketMQTemplate.sendOneWay("topic10", user);
  18. // 延时消息
  19. rocketMQTemplate.syncSend("topic10", MessageBuilder.withPayload(user).build(), 2000L, 3); 
  20. // 批量消息
  21. rocketMQTemplate.syncSend("topic10", list, 1000);

消费者：在注解中可以实现根据Tag和SQL进行属性的过滤。

  1. @Service
  2. //@RocketMQMessageListener(
  3. //        consumerGroup = "group1",
  4. //        topic = "topic10",
  5. //        selectorExpression = "tag1 || tag2"
  6. //)
  7. @RocketMQMessageListener(
  8.         consumerGroup = "group1",
  9.         topic = "topic10",
  10.         selectorType = SelectorType.SQL92,
  11.         selectorExpression = "age > 16",
  12.         messageModel = MessageModel.BROADCASTING
  13. )
  14. public class UserConsumer implements RocketMQListener<User> {
  17.     @Override
  18.     public void onMessage(User message) {
  20.     }
  21. }

总结

今天主要分享了一下RocketMQ的一些基础使用，包括各种类型的消息的使用，偏向于代码实现部分，对于原理篇没有过多涉及。

原文链接：https://www.cnblogs.com/Leo-coding/p/17367503.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

RocketMQ热门文章

  • RocketMq开启安全认证ACL-解决服务器系统安全漏洞
  • SpringBoot整合RocketMQ，老鸟们都是这么玩的！
  • RocketMQ 消息重试与死信队列
  • 【RocketMQ】Dledger日志复制源码分析
  • Prometheus 监测 RocketMQ 最佳实践
  • RocketMQ - 消费者Rebalance机制
  • RocketMq深入分析讲解两种削峰方式
  • 基于 RocketMQ 的 Dubbo-go 通信新范式
  • RocketMQ 在网易云音乐的实践
  • Redis+Hbase+RocketMQ 实际使用问题案例分享

    • RocketMQ推荐文章

  • 【主流技术】聊一聊消息队列 RocketMQ 的基本结构与概念
  • RocketMq开启安全认证ACL-解决服务器系统安全漏洞
  • RocketMQ—RocketMQ发送同步、异步、单向、延迟、批量、顺序、批量消息、带标签消息
  • 一文讲透消息队列RocketMQ实现消费幂等
  • 【RocketMQ】数据的清理机制
  • RocketMQ 消息重试与死信队列
  • 【RocketMQ】【源码】延迟消息实现原理
  • RocketMQ 系列(五）高可用与负载均衡
  • 【RocketMQ】消息的拉取总结
  • 在.NET Framework中使用RocketMQ（阿里云版）实战【第一章】

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号