• 消息中间件-RocketMQ

            RocketMQ是阿里巴巴开源的消息分布中间件，在阿里内部使用非常更广泛，已经经过了“双11”这种万亿级的应用场景考验。

    1.安装

            下载地址：http://rocketmq.apache.org/release_notes/release-notes-4.4.0/

            下载完成后解压缩安装包到指定目录。

    2.配置

               （1）环境变量：ROCKETMQ_HOME

                  15cc86086691442bb67803f7f0e9f1e0.png​

               （2）Path

                  9c392ee87366476887da013e1fe6cb32.png​

    3.启动

              切换到当前下载RocketMQ的bin目录下

            （1）启动NameServer ---> start mqnamesrv.cmd

    2a3aea30ac1d4813b6176b516b2a5405.png​

                  7300e1a85d5b41f6a6cd98004311fe5f.png​

            (2）启动Broker --->   start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

    91f08c10361a4758b1e988029b56fbd8.png​

    4.RocketMQ的架构及概念

            看下面这张图

            26ce0840bfc043a2a063417689cdd66b.png​

     

            RocketMQ整体分为4个角色，NameServer、Broker、Producer、Consumer、

            Broker：为RocketMQ的核心，负责消息的接受，存储，投递等功能。

            NameServer：消息队列的协商者，Broker向它注册路由信息，同时Producer和Consumer向其获取路由信息。

            Producer：消息的生产者，需要从NameServer获取Borker信息，然后与Broker建立连接，向Broker发送消息。

            Consumer：消息的消费者，需要从NameServer获取Broker信息，然后与Broker建立连接，从Broker获取消息。

            另外还包括别的组件，

            Topic：用来区分不同的消息类型，发送和接收消息前都要先创建Topic，针对Topic来发送和接收消息。

            Message Queue：消息队列，一个Topic可以设置一个或多个MessageQueue，这样消息就可以并行往各个Message Queue发送消息，消费者也可以并行的从多个Message Queue读取消息，提高性能和吞吐量。

            Message：消息的载体。

    5.消息发送和接收（应用）

           搭建好SpringBoot项目后，先导入RocketMQ所需的依赖：

    1.  
    2.         <dependency>
    3.             <groupId>org.apache.rocketmqgroupId>
    4.             <artifactId>rocketmq-clientartifactId>
    5.             <version>4.4.0version>
    6.         dependency>

    5.1 同步发送        

            同步发送方式比较可靠，应用也比较广泛，比如：重要的消息通知，短信通知。

    消息发送方：

    1. //发送同步消息
    2. public class RocketMQSendTest1 {
    3.     public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
    4.         //1.创建消息生产者，指定生产者所属的组名
    5.         DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
    6.         //2.指定Nameserver地址
    7.         producer.setNamesrvAddr("127.0.0.1:9876");
    8.         //3.启动生产者
    9.         producer.start();
    10.         for (int i = 0; i < 10; i++) {
    11.             Message message = new Message("myTopic", "myTag", ("十行代码九个错误八个警告竟敢说七日精通六天学会五湖四海也不见如此三心二意之程序简直一等下流" + i).getBytes());
    12.             //发送消息
    13.             SendResult sendResult = producer.send(message);
    14.             System.out.println(sendResult);
    15.         }
    16.  
    17.         //关闭生产者
    18.         producer.shutdown();
    19.     }
    20. }

    消息接收方：

    1. //接收消息
    2. public class RocketMQReceiveTest1 {
    3.     public static void main(String[] args) throws MQClientException {
    4.         //创建消息消费者。指定消费者所属的组名
    5.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumer-group");
    6.         //指定Nameserver地址
    7.         consumer.setNamesrvAddr("127.0.0.1:9876");
    8.         //指定消费者订阅的主题和标签
    9.         consumer.subscribe("myTopic", "*");
    10.  
    11.         //设置回调函数，编写处理消息的方法
    12.         consumer.registerMessageListener(new MessageListenerConcurrently() {
    13.             @Override
    14.             public ConsumeConcurrentlyStatus consumeMessage(List list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
    15.                 System.out.println("获取到的消费数据：" + list);
    16.                 System.out.println(new String(list.get(0).getBody()));
    17.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    18.             }
    19.         });
    20.  
    21.         //启动消息消费者
    22.         consumer.start();
    23.         System.out.println("Consumer Starting.");
    24.     }
    25. }

            启动时，切记先启动消费者（接收方），再启动生产者（发送方）。

            启动测试，同步发送消息结果： 

            ff8315dd84b94c708fb27135329444ed.png​

    5.2 异步消息

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

    发送方：

    1. //发送异步消息
    2. //异步消息比较浪费性能，经常会失败，所以多发送几次并且让线程休眠几秒
    3. public class RocketMQSendTest2 {
    4.  
    5.     public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException {
    6.         //1.创建消息生产者，指定生产者所属的组名
    7.         DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
    8.         //2.设置NameServer地址
    9.         producer.setNamesrvAddr("127.0.0.1:9876");
    10.         //3.启动生产者
    11.         producer.start();
    12.         for (int i = 0; i < 10; i++) {
    13.             //4.创建消息对象，制定主题、标签、消息体
    14.             Message message = new Message("myTopic", "myTag2", ("十行代码九个错误八个警告竟敢说七日精通六天学会五湖四海也不见如此三心二意之程序简直一等下流").getBytes());
    15.             //5.发送消息
    16.             producer.send(message, new SendCallback() {
    17.                 @Override
    18.                 public void onSuccess(SendResult sendResult) {
    19.                     System.out.println("发送成功：" + sendResult);
    20.                 }
    21.  
    22.                 @Override
    23.                 public void onException(Throwable e) {
    24.                     System.out.println("发送异常：" + e);
    25.                 }
    26.             });
    27.  
    28.             //休眠
    29.             TimeUnit.SECONDS.sleep(3);
    30.         }
    31.  
    32.         //6.关闭生产者
    33.         producer.shutdown();
    34.  
    35.     }
    36. }

    接收方：

    1. public class RocketMQReceiveTest2 {
    2.     public static void main(String[] args) throws MQClientException {
    3.         //1.创建消息消费者，指定消费者所属的组名
    4.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumer-group");
    5.  
    6.         //2.指定Nameserver地址
    7.         consumer.setNamesrvAddr("127.0.0.1:9876");
    8.         //3.指定消费者订阅的主题和标签
    9.         consumer.subscribe("myTopic", "*");
    10.         consumer.setMessageModel(MessageModel.BROADCASTING);
    11.         //4.设置回调函数，编写处理请求的方法
    12.         consumer.registerMessageListener(new MessageListenerConcurrently() {
    13.             @Override
    14.             public ConsumeConcurrentlyStatus consumeMessage(List list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
    15.  
    16.                 System.out.println(new String(list.get(0).getBody()));
    17.                 //返回消费状态
    18.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    19.             }
    20.         });
    21.         //5.启动消息消费者
    22.         consumer.start();
    23.         System.out.println("Consumer Starting.");
    24.  
    25.     }
    26. }

    启动测试，异步发送消息结果： 

    04a1bcc45c5e4302888c4f16a0340ef5.png​

    5.3 单行发送消息

          该方式用在不关注发送结果的场景，比如日志发送。

    1. //单行发送消息
    2. public class RocketMQSendTest3 {
    3.     public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
    4.         DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
    5.         producer.setNamesrvAddr("127.0.0.1:9876");
    6.         producer.start();
    7.         for (int i = 0; i < 10; i++) {
    8.             Message message = new Message("myTopic", "myTag3", ("芜湖").getBytes());
    9.             producer.sendOneway(message);
    10.  
    11.             TimeUnit.SECONDS.sleep(3);
    12.         }
    13.         producer.shutdown();
    14.     }
    15.  
    16. }

    启动两个消费者接收10条消息，结果如下：

    4ac62d7e77994809babf36bd1e0b13bf.png​

    c4890f8db72d4ebe905bd574a3471b09.png​

    存在消息丢失的情况。

    以上就是对RocketMQ的初步认识啦！

     

  • 相关阅读:
    技术分享 | 自动驾驶的春晚—Tesla AI Day
    前端工具——01-VS Code的使用
    Linux parted 分区命令详解
    大多数计算机专业研究生的三年是怎么过的？
    使用注解新开事务 @Transactional
    相机与相机模型（针孔/鱼眼/全景相机）
    【MySQL】用户管理&权限控制
    在Linux中安装中文包
    Windows环境下Springboot3+Graalvm+Idea 打包成原生镜像 踩坑
    用EasyV实现地图下钻，快速查看省市区热力值，老板看了都说好！
  • 原文地址：https://blog.csdn.net/weixin_62226624/article/details/132841974