【RocketMQ】第一篇-高性能消息中间件RocketMQ的初体验

谁借莪１个温暖的怀抱￠ 2023-10-09 14:59 76阅读 0赞

#### RocketMQ ####

*  一、What is RocketMQ？
 *   *  1、消息中间件的应用场景
     *  2、RocketMQ发展演进历程
     *  3、常见的MQ产品比较
 *  二、RocketMQ运行环境
 *  三、RocketMQ管理控制台
 *   *  1、环境安装
     *  2、运维菜单
     *  3、驾驶舱菜单
     *  4、集群菜单
     *  5、主题菜单
     *  6、消费者菜单
 *  四、RocketMQ快速体验
 *  四、RocketMQ Client消息传递开发
 *   *  1、 发送同步消息
     *  2、 发送异步消息
     *  3、SpringBoot集成RocketMQ
 *  五、接受、消费模式
 *   *  接收消息-负载均衡模式
     *  接收消息-广播模式
     *  消息发送-延时消息
     *  消息发送-批量消息
     *  消息过滤 -过滤消息
     *  一个消费者订阅多个Topic

## 一、What is RocketMQ？ ##

官网：[http://rocketmq.apache.org/][http_rocketmq.apache.org]  
Apache RocketMQ是一个统一的消息引擎，轻巧的数据处理平台；  
特点是低延迟：高压力下99.6%的响应在1毫秒内；（经受了阿里双11的挑战）

### 1、消息中间件的应用场景 ###

![在这里插入图片描述][6966bd45f0294a488ab8386535ff357b.png]

*  1、异步处理  
    同步是阻塞的（会造成等待），异步是非阻塞的（不会等待）；  
    大流量高并发请求、批量数据传递，就可以采用异步处理，提升系统吞吐量；
 *  2、系统解耦  
    多个系统之间，不需要直接交互，通过消息进行业务流转；
 *  3、流量削峰  
    高负载请求/任务的缓冲处理
 *  4日志处理  
    日志处理是指将消息队列用于在日志处理中，比如Kafka解决大量日志传输的问题；（日志不落地，就是log.info(…) 这是写入磁盘，改为 发送一个消息）
 *  5消息通讯  
    消息队列一般都内置了高效的通信机制，因此也可以用于单纯的消息通讯，比如实现点对点消息队列或者聊天室等；

### 2、RocketMQ发展演进历程 ###

RocketMQ是阿里巴巴开发并开源的消息中间件，采用Java语言开发的；  
2007年，阿里启动了“五彩石”项目，异步信息传递采用自己研发的第一代消息引擎Notify；  
2010年，阿里开始大规模使用ActiveMQ，并打造了自己的消息引擎Napoli，还有之前的消息引擎Notify；  
2011年，LinkedIn开源了自己的分布式消息引擎Kafka，阿里采用Java重写了Kafka的核心逻辑，并取名为MetaQ进行开源，并不断升级迭代版本到MetaQ2.x；  
2012年，阿里完全重写MetaQ2.0，版本升级到MetaQ3.0，并重新命名RocketMQ3.0；  
2016年11月，阿里将RocketMQ捐赠给Apache（版本升级为RocketMQ4.0）  
2017年，不到10个月的时间，RocketMQ 从 Apache 毕业，成为Apache下的顶级项目；

### 3、常见的MQ产品比较 ###

![在这里插入图片描述][1dba290db1d64b7aaabdb5796899214e.png]

## 二、RocketMQ运行环境 ##

1、Linux上需要安装Java环境；（64bit JDK 1.8+）  
2、官网：[http://rocketmq.apache.org/][http_rocketmq.apache.org]  
3、下载：[http://rocketmq.apache.org/release\_notes/release-notes-4.7.1/][http_rocketmq.apache.org_release_notes_release-notes-4.7.1]

4、解压：`unzip rocketmq-all-4.7.1-bin-release.zip`  
解压后rocketmq即完成了安装；  
5、启动RocketMQ  
（1）启动Name Server：`./mqnamesrv &`  
检查进程：ps -ef | grep namesrv 或者 jps  
日志文件生成到用户名下 /root/logs/rocketmqlogs/namesrv.log  
Name Server启动后默认端口为9876；  
（2）启动Broker：`./mqbroker -n localhost:9876 &`  
指定配置文件启动：`./mqbroker -n localhost:9876 -c ../conf/broker.conf &`  
查看进程：ps -ef | grep mqbroker 或者jps  
日志文件生成到用户名下 /root/logs/rocketmqlogs/broker.log  
虚拟机下启动失败，原因：  
RocketMQ默认的JVM内存配置比较大，如果当前机器内存不够就会导致启动失败，我们可以修改RocketMQ默认的JVM内存大小，修改如下两个文件：  
**runbroker.sh  
runserver.sh**  
6、关闭RocketMQ

./mqshutdown broker
    ./mqshutdown namesrv

## 三、RocketMQ管理控制台 ##

### 1、环境安装 ###

RocketMQ有一个扩展的开源项目  
[https://github.com/apache/rocketmq-externals][https_github.com_apache_rocketmq-externals]  
这个项目中有一个子模块叫rocketmq-console，这个就是管理控制台项目，将该项目拉取到本地，对rocketmq-console进行编译打包；

git clone https://github.com/apache/rocketmq-externals
    cd rocketmq-console
    mvn clean package -Dmaven.test.skip=true

注：打包前在rocketmq-console中配置namesrv集群地址：

rocketmq.config.namesrvAddr=192.168.112.132:9876;192.18.172.132:9876

启动rocketmq-console：

java -jar rocketmq-console-ng-1.0.0.jar

启动成功后，就可以通过浏览器访问http://localhost:8080进入控制台界面，将看到很多菜单；  
![在这里插入图片描述][6248d96995f04e1294600ccd914905e8.png]

### 2、运维菜单 ###

可以修改这个服务使用的navesvr的地址；  
可以修改这个服务是否使用VIPChannel(如果你的mq server版本小于3.5.8，请设置不使用)

### 3、驾驶舱菜单 ###

查看broker的消息量（总量/5分钟图）  
查看单一主题的消息量（总量/趋势图）

### 4、集群菜单 ###

查看集群的分布情况  
cluster与broker关系  
broker  
查看broker具体信息/运行信息  
查看broker配置信息

### 5、主题菜单 ###

展示所有的主题，可以通过搜索框进行过滤；  
筛选 普通/重试/死信 主题；  
添加/更新主题；  
clusterName 创建在哪几个cluster上；  
brokerName 创建在哪几个broker上；  
topicName 主题名；  
writeQueueNums 写队列数量；  
readQueueNums 读队列数量；  
perm //2是写 4是读 6是读写；  
状态 查询消息投递状态（投递到哪些broker/哪些queue/多少量等）  
路由 查看消息的路由（现在你发这个主题的消息会发往哪些broker，对应broker的queue信息）  
CONSUMER管理（这个topic都被哪些group消费了，消费情况何如）  
topic配置（查看变更当前的配置）  
发送消息（向这个主题发送一个测试消息）  
重置消费位点(分为在线和不在线两种情况，不过都需要检查重置是否成功)  
删除主题 （会删除掉所有broker以及namesvr上的主题配置和路由信息）

### 6、消费者菜单 ###

展示所有的消费组，可以通过搜索框进行过滤；  
刷新页面/每隔五秒定时刷新页面；  
按照订阅组/数量/TPS/延迟 进行排序；  
添加/更新消费组；  
clusterName 创建在哪几个集群上；  
brokerName 创建在哪几个broker上；  
groupName 消费组名字；  
consumeEnable //是否可以消费 FALSE的话将无法进行消费；  
consumeBroadcastEnable //是否可以广播消费；  
retryQueueNums //重试队列的大小；  
brokerId //正常情况从哪消费；  
whichBrokerWhenConsumeSlowly//出问题了从哪消费；  
终端 在线的消费客户端查看，包括版本订阅信息和消费模式；  
消费详情 对应消费组的消费明细查看，这个消费组订阅的所有Topic的消费情况，每个queue对应的消费client查看（包括Retry消息）；  
配置 查看变更消费组的配置；  
删除 在指定的broker上删除消费组；

## 四、RocketMQ快速体验 ##

发送（生产者）

export NAMESRV_ADDR=localhost:9876
    ./tools.sh org.apache.rocketmq.example.quickstart.Producer

接收（消费者）

export NAMESRV_ADDR=localhost:9876
    ./tools.sh org.apache.rocketmq.example.quickstart.Consumer

## 四、RocketMQ Client消息传递开发 ##

添加依赖

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

### 1、 发送同步消息 ###

这是比较常见的发送方式，同步进行消息发送，发送时会等待消息服务器Broker的响应；

/**
     * 消息生产者
     *
     */
    public class SyncProducer {
        
    
        public static void main(String[] args) throws Exception {
        
            //消息发送
            sendMsg();
        }
    
        public static void sendMsg() throws Exception {
        
            // 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("producer-group");
            // 设置NameServer的地址
            
            producer.setNamesrvAddr("127.0.0.1:9876");
            // 启动Producer实例
            producer.start();
    
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message("MyTopic","TagA",("Hello RocketMQ").getBytes(RemotingHelper.DEFAULT_CHARSET));
    
            // 发送消息到一个Broker (同步发送)
            SendResult sendResult = producer.send(msg);
    
            // 通过sendResult返回消息是否成功送达
            System.out.println(sendResult);
    
            // 如果不再发送消息，关闭Producer实例
            producer.shutdown();
        }
    }
    
    /**
     * 消息消费者
     *
     */
    public class SyncConsumer {
        
    
        public static void main(String[] args) throws Exception {
        
            //接收消息
            receiveMsg();
        }
    
        public static void receiveMsg() throws Exception {
        
            // 实例化消息生产者，指定组名
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
            // 指定Namesrv地址信息.
            
            consumer.setNamesrvAddr("127.0.0.1:9876");
            // 订阅Topic
            consumer.subscribe("MyTopic", "*");
            //负载均衡模式消费
            consumer.setMessageModel(MessageModel.BROADCASTING); //可以理解为：单个消费者接收
            // 注册回调函数，处理消息
            consumer.registerMessageListener(
                (List<MessageExt> msgs, ConsumeConcurrentlyContext context) -> {
        
                    for (MessageExt ext : msgs) {
        
                        System.out.println("接收到的消息为：" + new String(ext.getBody()));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            );
            //启动消息者
            consumer.start();
            System.out.println("Consumer Started.");
        }
    }

### 2、 发送异步消息 ###

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

/**
     * 消息生产者
     *
     */
    public class AsyncProducer {
        
    
        public static void main(String[] args) throws Exception {
        
            //消息发送
            sendMsg();
        }
    
        public static void sendMsg() throws Exception {
        
            // 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("producer-group");
            // 设置NameServer的地址
               consumer.setNamesrvAddr("127.0.0.1:9876");
            // 启动Producer实例
            producer.start();
    
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message("MyTopic","TagA", ("Hello RocketMQ").getBytes(RemotingHelper.DEFAULT_CHARSET));
    
            // 异步发送消息到一个Broker,SendCallback接收异步返回结果的回调
            producer.send(msg, new SendCallback() {
        
                @Override
                public void onSuccess(SendResult sendResult) {
        
                    System.out.println("发送结果：" + sendResult.getSendStatus());
                }
                @Override
                public void onException(Throwable e) {
        
                    e.printStackTrace();
                }
            });
            // 如果不再发送消息，关闭Producer实例。
            Thread.sleep(5000);
    
            producer.shutdown();
        }
    }
    
    /**
     * 消息消费者
     *
     */
    public class AsyncConsumer {
        
    
        public static void main(String[] args) throws Exception {
        
            receiveMsg();
        }
    
        public static void receiveMsg() throws Exception {
        
            // 实例化消息生产者,指定组名
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group4");
            // 指定Namesrv地址信息.
             consumer.setNamesrvAddr("127.0.0.1:9876");
            // 订阅Topic
            consumer.subscribe("MyTopic", "*");
            //负载均衡模式消费
    
            consumer.setMessageModel(MessageModel.CLUSTERING); //集群
            // 注册回调函数，处理消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {
        
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        
                    for (MessageExt ext : msgs) {
        
                        System.out.println("接收到的消息为：" + new String(ext.getBody()));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            //启动消息者
            consumer.start();
            System.out.println("Consumer Started.");
        }
    }

### 3、SpringBoot集成RocketMQ ###

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.1.0</version>
    </dependency>

## 五、接受、消费模式 ##

RocketMQ消息发送

### 接收消息-负载均衡模式 ###

消费者采用负载均衡方式消费消息，多个消费者共同消费队列消息，每个消费者消费不同的消息，每个消费者是竞争关系，同一个消息不能被多个消费者消费；

### 接收消息-广播模式 ###

消费者采用广播的方式消费消息，每个消费者消费的消息都是相同的，同一个消息可以被多个不同的消费者消费；

### 消息发送-延时消息 ###

比如用户提交了一个订单就可以发送一个延时消息，30分钟后去检查这个订单的状态，如果还未支付就取消该订单并释放库存；

// org/apache/rocketmq/store/config/MessageStoreConfig.java
    private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

RocketMQ不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18；  
具体参考样例代码；

### 消息发送-批量消息 ###

批量发送消息能显著提高传递小消息的性能，限制是这些批量消息应该有相同的topic，相同的waitStoreMsgOK，而且不能是延时消息，此外这一批消息的总大小不应超过4MB；

### 消息过滤 -过滤消息 ###

在大多数情况下，可以采用TAG来进行消息过滤，例如：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
    consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

消费者将接收包含TAGA或TAGB或TAGC的消息，但是限制是一个消息只能有一个标签，这对于复杂的场景可能不满足需求，在这种情况下，可以使用SQL表达式筛选消息，SQL特性可以通过发送消息时的属性来进行计算；

![在这里插入图片描述][985dc1c5ce064b028331ae143ccc5003.png]

RocketMQ只定义了一些基本语法来支持这个特性，也可以扩展它；  
数值比较，比如：>，>=，<，<=，BETWEEN，=；  
字符比较，比如：=，<>，IN；  
IS NULL 或者 IS NOT NULL；  
逻辑符号 AND，OR，NOT；  
常量支持类型为：  
数值，比如：123，3.1415；  
字符，比如：‘abc’，必须用单引号包裹起来；  
NULL，特殊的常量  
布尔值，TRUE 或 FALSE  
消费者使用SQL92标准的sql语句，接口如下：

public void subscribe(finalString topic, final MessageSelector messageSelector)

如果报错不支持SQL过滤，则在broker的配置文件broker.conf添加 enablePropertyFilter = true 来支持SQL92方式过滤消息；  
启动的时候需要指定配置文件启动`：./mqbroker -n localhost:9876 -c ../conf/broker.conf &`

### 一个消费者订阅多个Topic ###

在springboot整合下，通过监听器的方式，不能实现监听多个topic，我们可以采用变通的办法，在springboot中配置一个消费者的bean，消费者的bean里面订阅多个topic，然后让消费者的bean注册一个监听器监听消息；

@Slf4j
    @Configuration
    public class RocketMQConfig {
        
    
        @Value("${rocketmq.name-server}")
        private String namesrvAddr;
    
        @Value("${rocketmq.consumer.groupName}")
        private String groupName;
    
        @Value("${rocketmq.consumer.topics}")
        private String topics;
    
        @Bean
        public DefaultMQPushConsumer defaultMQPushConsumer () {
        
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
            consumer.setNamesrvAddr(namesrvAddr);
            consumer.setMessageModel(MessageModel.CLUSTERING);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
        
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        
                    //接收到消息
                    msgs.forEach((MessageExt ext) -> {
        
                        System.out.println("topic=" + ext.getTopic() + ", tag=" + ext.getTags() + ", msg=" + new String(ext.getBody()));
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            try {
        
                //设置该消费者订阅的主题和tag，如果是订阅该主题下的所有tag，则tag使用*；
                //如果需要指定订阅该主题下的某些tag，则使用||分割，例如tag1||tag2||tag3
                String[] topicsArr = topics.split(",");
                for (String topic : topicsArr) {
        
                    //TODO 消费者循环订阅topic，目前订阅了3个topic
                    consumer.subscribe(topic, "*");
                }
                consumer.start();
                log.info("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}",groupName,topics,namesrvAddr);
            }catch (MQClientException e){
        
                log.error("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}",groupName,topics,namesrvAddr,e);
                throw new RuntimeException(e);
            }
            return consumer;
        }
    }

[http_rocketmq.apache.org]: http://rocketmq.apache.org/
[6966bd45f0294a488ab8386535ff357b.png]: https://img-blog.csdnimg.cn/6966bd45f0294a488ab8386535ff357b.png
[1dba290db1d64b7aaabdb5796899214e.png]: https://img-blog.csdnimg.cn/1dba290db1d64b7aaabdb5796899214e.png
[http_rocketmq.apache.org_release_notes_release-notes-4.7.1]: http://rocketmq.apache.org/release_notes/release-notes-4.7.1/
[https_github.com_apache_rocketmq-externals]: https://github.com/apache/rocketmq-externals
[6248d96995f04e1294600ccd914905e8.png]: https://img-blog.csdnimg.cn/6248d96995f04e1294600ccd914905e8.png
[985dc1c5ce064b028331ae143ccc5003.png]: https://img-blog.csdnimg.cn/985dc1c5ce064b028331ae143ccc5003.png