RabbitMQ是基于AMQP协议的一款消息管理系统，是部署最广泛的开源消息中间件，是最受欢迎的开源消息中间件之一：官网： http://www.rabbitmq.com/官方教程：http://www.rabbitmq.com/getstarted.htmlAMQP 协议：AMQP（advanced message queuing protocol）在2003年时被提出，最早用于解决金融领不同平台之间的消息传递交互问题。顾名思义，AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。以下是AMQP协议模型；不同MQ的区别：ActiveMQ：ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。它是一个完全支持JMS规范的的消息中间件。丰富的API,多种集群架构模式让ActiveMQ在业界成为老牌的消息中间件,在中小型企业颇受欢迎!Kafka：Kafka是LinkedIn开源的分布式发布-订阅消息系统，目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输。0.8版本开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务。RocketMQ：RocketMQ是阿里开源的消息中间件，它是纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起源于Kafka，但并不是Kafka的一个Copy，它对消息的可靠传输及事务性做了优化，目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。RabbitMQ：RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。总结：RabbitMQ比Kafka可靠，Kafka更适合IO高吞吐的处理，一般应用在大数据日志处理或对实时性（少量延迟），可靠性（少量丢数据）要求稍低的场景使用，比如ELK日志收集。2. Linux中下载和安装2.1 下载地址官网下载地址： https://www.rabbitmq.com/download.html​下载安装压缩包：百度云下载链接：https://pan.baidu.com/s/17hZab0ZKZLtoi-6q14R27Q，提取码：dwti​注意：这里的安装包是centos7安装的包！2.2 安装步骤2.2.1 将rabbitmq安装包上传到linux系统中这里我使用XFTP工具上传，使用XShell 连接云服务器，上传位置为：=/usr/local/src/software/rabbitMQ/，如果对应的文件夹不存在，自己创建一下！2.2.2 安装Erlang依赖包进入目标文件夹下：cd /usr/local/src/software/rabbitMQ/，并查看安装包信息：ls -l执行安装Erlang依赖包：rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm执行安装Erlang内存管理的依赖包：rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm2.2.3 安装RabbitMQ安装包(需要联网)执行安装rabbitmq：rpm -ivh rabbitmq-server-3.7.18-1.el7.noarch.rpm2.2.4 复制配置文件执行命令：cp /usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config，该命令是将rabbitmq.config.example 配置文件复制到/etc/rabbitmq/ 下，并该名为rabbitmq.config！注意：如果不知道rabbitmq.config.example 配置文件的具体位置，可以使用命令：find / -name rabbitmq.config.example去查找其位置！因为是从/目录开始查找可能会延迟几秒！2.2.5 查看/修改配置文件查看配置文件位置：ll /etc/rabbitmq/修改配置文件：vim /etc/rabbitmq/rabbitmq.config开放来宾账户权限loopback_users：把注释开放，注意尾部的逗号也去掉：3. 相关命令3.2 开启web界面管理工具命令：rabbitmq-plugins enable rabbitmq_management3.2 启动、停止命令服务启动：systemctl start rabbitmq-server服务重启：systemctl restart rabbitmq-server服务关闭：systemctl stop rabbitmq-server查看服务运行状态：systemctl status rabbitmq-server4. RabbitMQ可视化管理页面4.1 开放端口如果是本地虚拟机开放防火墙对应端口，如果是云服务器开放安全组端口：4.2 访问管理页面浏览器访问：http://服务器IP地址:15672用户名和密码默认是我们在配置文件中开启的来宾账户：guest/guest，输入账号密码后进入管理页：有上图可知，为了之后使用代码整合RabbitMq，我们还需要在安全组中开放另外两个端口：4.3 管理界面导航介绍4.4 Admin用户和虚拟主机管理4.4.1 添加用户上面的Tags选项，其实是指定用户的角色，可选的有以下几个：超级管理员(administrator)可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。监控者(monitoring)可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)策略制定者(policymaker)可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。普通管理者(management)仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。其他无法登陆管理控制台，通常就是普通的生产者和消费者。4.4. 2 添加虚拟主机点击给虚拟主机leyou 设置访问用户以及其权限：接下来退出登录，并以leyou用户登录即可！5. Java API 操作RabbitMQ5 种消息模型：简单消息模型：1个生产者 + 1个队列 + 1个消费者；工作队列消息模型：1个生产者 + 1个队列 + 多个消费者，一条消息只能被消费一次；订阅消息模型之 fanout：1个生产者 + 1个交换机 + 多个队列 + 多个消费者，一条消息可以被多个消费者消费；订阅消息模型之durect/router：1个生产者 + 1个交换机 + 多个队列 + 多个消费者 ，routingKey ，一条消息发送给符合 routingKey 的队列；订阅消息模型之topic：通配符，#：匹配一个或者多个 *：一个词；官网教程地址：https://www.rabbitmq.com/getstarted.html​​参考传智播客的Demo 案例：代码案例，基本上是参考官网的quickStart的，可以快速上手！​​6. Spring Boot 整合RabbiMQSpring AMQP 官网地址：https://spring.io/projects/spring-amqpSpringBoot整合官方文档地址：https://docs.spring.io/spring-boot/docs/current/reference/html/spring-boot-features.html#boot-features-messaging6.1 pom.xml 引入依赖<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> 6.2 application.yml 中进行配置# Spring 相关配置 spring: # RabbitMQ 相关配置 rabbitmq: # 主机ip地址 host: 8.xxx.xx.xx6 # 授权用户名 username: leyou # 授权用户密码 password: leyou # 授权虚拟主机名称 virtual-host: leyou 6.3 RabbitMQ 监听器(消息消费者)组件注入IOC/** * RabbitMQ 监听器组件：相当于消费者 */ @Component public class Listener { @RabbitListener(bindings = @QueueBinding( // value = "spring.test.queue" 队列名称 // durable = "true" 队列消息持久化 value = @Queue(value = "spring.test.queue", durable = "true"), // value = "spring.test.exchange" 交换机名称 // durable = "true" 交换机消息持久化 // ignoreDeclarationExceptions = "true" 忽略声明异常 // type = ExchangeTypes.TOPIC 交换机类型：TOPIC exchange = @Exchange( value = "spring.test.exchange", ignoreDeclarationExceptions = "true", type = ExchangeTypes.TOPIC // durable = "true" ), // 消息接收的路由表达式 key = {"#.#"})) public void listen(String msg) { System.out.println("接收到消息：" + msg); } } 6.4 Test 测试模拟消息生产者@RunWith(SpringRunner.class) @SpringBootTest(classes = Application.class) public class MqDemo { // 注入AmqpTemplate模板 @Autowired private AmqpTemplate amqpTemplate; @Test public void testSend() throws InterruptedException { String msg = "hello, Spring boot amqp"; // 通过AmqpTemplate模板发送消息 // 参数1：交换机 // 参数2：a.b是否符合监听组件(消息消费者)中定义的消息接收的路由表达式 // 参数3：消息内容 this.amqpTemplate.convertAndSend("spring.test.exchange","a.b", msg); // 等待10秒后再结束 Thread.sleep(10000); } } 6.5 执行测试可以看到消息生产后，被消息监听器(消费者)接收成功！​​

RabbitMQ是基于AMQP协议的一款消息管理系统，是部署最广泛的开源消息中间件，是最受欢迎的开源消息中间件之一：官网： http://www.rabbitmq.com/官方教程：http://www.rabbitmq.com/getstarted.htmlAMQP 协议：AMQP（advanced message queuing protocol）在2003年时被提出，最早用于解决金融领不同平台之间的消息传递交互问题。顾名思义，AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。以下是AMQP协议模型；不同MQ的区别：ActiveMQ：ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。它是一个完全支持JMS规范的的消息中间件。丰富的API,多种集群架构模式让ActiveMQ在业界成为老牌的消息中间件,在中小型企业颇受欢迎!Kafka：Kafka是LinkedIn开源的分布式发布-订阅消息系统，目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输。0.8版本开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务。RocketMQ：RocketMQ是阿里开源的消息中间件，它是纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起源于Kafka，但并不是Kafka的一个Copy，它对消息的可靠传输及事务性做了优化，目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。RabbitMQ：RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。总结：RabbitMQ比Kafka可靠，Kafka更适合IO高吞吐的处理，一般应用在大数据日志处理或对实时性（少量延迟），可靠性（少量丢数据）要求稍低的场景使用，比如ELK日志收集。2. Linux中下载和安装2.1 下载地址官网下载地址： https://www.rabbitmq.com/download.html​下载安装压缩包：百度云下载链接：https://pan.baidu.com/s/17hZab0ZKZLtoi-6q14R27Q，提取码：dwti​注意：这里的安装包是centos7安装的包！2.2 安装步骤2.2.1 将rabbitmq安装包上传到linux系统中这里我使用XFTP工具上传，使用XShell 连接云服务器，上传位置为：=/usr/local/src/software/rabbitMQ/，如果对应的文件夹不存在，自己创建一下！2.2.2 安装Erlang依赖包进入目标文件夹下：cd /usr/local/src/software/rabbitMQ/，并查看安装包信息：ls -l执行安装Erlang依赖包：rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm执行安装Erlang内存管理的依赖包：rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm2.2.3 安装RabbitMQ安装包(需要联网)执行安装rabbitmq：rpm -ivh rabbitmq-server-3.7.18-1.el7.noarch.rpm2.2.4 复制配置文件执行命令：cp /usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config，该命令是将rabbitmq.config.example 配置文件复制到/etc/rabbitmq/ 下，并该名为rabbitmq.config！注意：如果不知道rabbitmq.config.example 配置文件的具体位置，可以使用命令：find / -name rabbitmq.config.example去查找其位置！因为是从/目录开始查找可能会延迟几秒！2.2.5 查看/修改配置文件查看配置文件位置：ll /etc/rabbitmq/修改配置文件：vim /etc/rabbitmq/rabbitmq.config开放来宾账户权限loopback_users：把注释开放，注意尾部的逗号也去掉：3. 相关命令3.2 开启web界面管理工具命令：rabbitmq-plugins enable rabbitmq_management3.2 启动、停止命令服务启动：systemctl start rabbitmq-server服务重启：systemctl restart rabbitmq-server服务关闭：systemctl stop rabbitmq-server查看服务运行状态：systemctl status rabbitmq-server4. RabbitMQ可视化管理页面4.1 开放端口如果是本地虚拟机开放防火墙对应端口，如果是云服务器开放安全组端口：4.2 访问管理页面浏览器访问：http://服务器IP地址:15672用户名和密码默认是我们在配置文件中开启的来宾账户：guest/guest，输入账号密码后进入管理页：有上图可知，为了之后使用代码整合RabbitMq，我们还需要在安全组中开放另外两个端口：4.3 管理界面导航介绍4.4 Admin用户和虚拟主机管理4.4.1 添加用户上面的Tags选项，其实是指定用户的角色，可选的有以下几个：超级管理员(administrator)可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。监控者(monitoring)可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)策略制定者(policymaker)可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。普通管理者(management)仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。其他无法登陆管理控制台，通常就是普通的生产者和消费者。4.4. 2 添加虚拟主机点击给虚拟主机leyou 设置访问用户以及其权限：接下来退出登录，并以leyou用户登录即可！5. Java API 操作RabbitMQ5 种消息模型：简单消息模型：1个生产者 + 1个队列 + 1个消费者；工作队列消息模型：1个生产者 + 1个队列 + 多个消费者，一条消息只能被消费一次；订阅消息模型之 fanout：1个生产者 + 1个交换机 + 多个队列 + 多个消费者，一条消息可以被多个消费者消费；订阅消息模型之durect/router：1个生产者 + 1个交换机 + 多个队列 + 多个消费者 ，routingKey ，一条消息发送给符合 routingKey 的队列；订阅消息模型之topic：通配符，#：匹配一个或者多个 *：一个词；官网教程地址：https://www.rabbitmq.com/getstarted.html​​参考传智播客的Demo 案例：代码案例，基本上是参考官网的quickStart的，可以快速上手！​​6. Spring Boot 整合RabbiMQSpring AMQP 官网地址：https://spring.io/projects/spring-amqpSpringBoot整合官方文档地址：https://docs.spring.io/spring-boot/docs/current/reference/html/spring-boot-features.html#boot-features-messaging6.1 pom.xml 引入依赖<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> 6.2 application.yml 中进行配置# Spring 相关配置 spring: # RabbitMQ 相关配置 rabbitmq: # 主机ip地址 host: 8.xxx.xx.xx6 # 授权用户名 username: leyou # 授权用户密码 password: leyou # 授权虚拟主机名称 virtual-host: leyou 6.3 RabbitMQ 监听器(消息消费者)组件注入IOC/** * RabbitMQ 监听器组件：相当于消费者 */ @Component public class Listener { @RabbitListener(bindings = @QueueBinding( // value = "spring.test.queue" 队列名称 // durable = "true" 队列消息持久化 value = @Queue(value = "spring.test.queue", durable = "true"), // value = "spring.test.exchange" 交换机名称 // durable = "true" 交换机消息持久化 // ignoreDeclarationExceptions = "true" 忽略声明异常 // type = ExchangeTypes.TOPIC 交换机类型：TOPIC exchange = @Exchange( value = "spring.test.exchange", ignoreDeclarationExceptions = "true", type = ExchangeTypes.TOPIC // durable = "true" ), // 消息接收的路由表达式 key = {"#.#"})) public void listen(String msg) { System.out.println("接收到消息：" + msg); } } 6.4 Test 测试模拟消息生产者@RunWith(SpringRunner.class) @SpringBootTest(classes = Application.class) public class MqDemo { // 注入AmqpTemplate模板 @Autowired private AmqpTemplate amqpTemplate; @Test public void testSend() throws InterruptedException { String msg = "hello, Spring boot amqp"; // 通过AmqpTemplate模板发送消息 // 参数1：交换机 // 参数2：a.b是否符合监听组件(消息消费者)中定义的消息接收的路由表达式 // 参数3：消息内容 this.amqpTemplate.convertAndSend("spring.test.exchange","a.b", msg); // 等待10秒后再结束 Thread.sleep(10000); } } 6.5 执行测试可以看到消息生产后，被消息监听器(消费者)接收成功！​​

三、TubeMQ的存储模式与管控措施MQ里最核心的是它的存储模式，如下图所示，右边的存储方案列表是一位叫陈大白的知乎用户给我们提供的，左边是TubeMQ的存储方案。TubeMQ采用的是按Topic组织内存+文件的存储实例方案来实现的，数据先写到主内存，主内存写满之后切为备份内存，数据异步从备份内存刷到文件完成数据写入。它通过消费的偏移度来决定它是由主内存还是备内存还是文件消费数据，这样的话可以减轻系统的负担，提高它的存储量。大家从右边的存储图上面可以看到：Kafka、RocketMQ和京东2019年发布JMQ，其实相差并不大。但是需要注意的是，每一种存储模式在不同的资源要求下，它的性能指标和相应的量级是不一样的。因为我们做的是有损服务，有损的服务是怎么回事呢？就是我们在机器断电、没有存储或者没有刷盘的情况下，数据就会丢失，在磁盘RAID10都无法恢复的磁盘故障情况下数据也会丢失。除了这两种情况，其他的情况都不会丢。因为上述两类故障随时可能发生，因而TubeMQ其实不适合用做持久的反复消费、而又需要前后数据完全一致的场景。那么，我们为什么要这么做呢？我们是不是做不到多副本呢？其实也不是的。问题就在于成本方面的考量。我们这样做，如果横向作比较，大家知道我们能够省多少台机器吗？换算成金钱的话能省多少吗？在这里给大家提供一个数据：2019年11月8日，开源Kafka项目的LinkIn公司发表了一篇文章，他们7万多条数据用了4000台机器，这个信息大家网上可以查到。另一个是我们国内做大数据与应用相关场景公司的例子，采用原生Kafka做大数据接入，在2018年底也达到了7、8万亿的数据量，花了1500台万兆机。说回来，我们这种模式下需要多少台机器呢？我们现在达到35万亿的数据量用的也是1500台机器，在相同的前提下，我们对比外部MQ，使用的机器数量只有它们的1/4、1/5。换算成人民币的话是多少？一台商用机大概是10万左右，仅仅机器成本我们就可以节约到几个亿，这就是为什么要采用这个方案的原因。跟Kafka异步节点复制方案相比，我们只需要1/4左右的机器量。当然，即便是用单复本，我们的性能也会比Kafka强很多，可以节约不少机器，相关数据可以看我们的测试报告。TubeMQ所有的管控逻辑包括所有的API接口都是围绕着它的存储来做的，包括它的Topic的配置和流控的处理和数据的查询、API的库存等等。下图所示的是TubeMQ最核心的API接口定义，主要分为4个章节。如果只是使用的话，直接通过管控台操控就可以了，但如果你要精细化地去调控系统，就需要去了解API的定义了。TubeMQ的管控模块Master，是基于BDB嵌入式数据库进行集群的Broker节点管理。各个Broker配置的Topic信息的数据存储，只要在标红的操作栏里操作，就会有一个状态告知操作者目前处于什么样的过程，是基于执行操作还是只读只写或者是可读写的情况。还可以通过这个页面查询。本系统在Windows上面就可以运行起来，欢迎大家去试用。TubeMQ的认证授权设计和传统的也不太一样，因为我们把TubeMQ的认证机做了重新的定义，具体可见下图。四、为什么选择开源？第一，基于公司的开源政策要求：对内开源协同，对外形成技术影响力，所以我们选择了开源。第二，从我们掌握的信息来看，我们认为在这一领域开源TubeMQ，是可以对有需要的同学们产生实际价值的。第三，我们觉得开源是在打破壁垒。在世界不同的角落，很多人都在研究这一问题，就像平行宇宙一样，大家都在各自的宇宙里面去研究和分析，相互之间没有太多的交流，我们相信肯定有人比我们做得更好，有值得我们学习的地方，所以我们把它开源出来，形成一个大家都了解、可以相互学习的状态，这样对自己也是一种促进。基于以上这三点，我们最终选择了开源。在已经开源情况下为什么还要去贡献过给Apache呢？其实我也理解有很多做开发的同学不敢去用一些开源项目，因为有很多公司开源了一个项目，用着用着结果发现没有人维护了。为了解决这个问题，我们希望把它捐献给一个中立的基金会，通过它已经成文的标准化流程，使项目成为一个大家可以接受的成熟项目，包括它的文档化和多种接入的情况。即便原创团队最后不接手这个项目了，后面也有人去接手它，使这个项目能够持续向前改进。所以我们把它捐献给了基金会。为什么选择Apache呢？因为我们是专注于大数据场景的MQ，而Apache是基于大数据这个生态最为出名的社区，而我们也同样也受惠于这个生态，所以理所当然就想回馈社区，将项目捐献给Apache。前段时间TubeMQ已经成为了Apache的孵化项目。五、TubeMQ的后续发展探讨2020年上半年我们在开源的协同推广下，内部接入的业务数据将会越来越多，日均接入量相信很快就会过40万亿。我们的机器也将会由以前的TS60升级成BX2，它将会带来什么样的变化呢？以往的机器是CPU 99%，磁盘IO是30~40%，根据最新的测试数据，在BX2上面变为CPU 30~40%，磁盘IO 99%。由此可见，我们需要把它磁盘的IO尽可能地降下来，或者选择其他更合适的机器，这是需要去研究的。另外，因为我们已经开源了，后续如何培养社区也是一个比较关键的点。目前来看，我们会基于协作的机制将它开源，无论是公司内还是公司外的同学，一起贡献来把这个项目做大，我们会在自己擅长的领域把这个东西继续夯实，大家可以根据自己的需要去使用我们的项目。同时大家在使用的过程中如果能发现有些不完善的地方，也希望能通过社区贡献出来，大家一起努力把这个项目做好。其实我们不仅仅只有MQ，我们同样在做的还有汇聚层和采集层，在此之上还有管理层。我们的希望是把MQ这一块做稳定以后，再将整体开源出来。我们会允许这一套系统接纳不同的MQ，根据MQ不同的特点提供给外部业务使用，但对外部业务又是无感知的。六、Q&AQ：张老师，你刚才做了TubeMQ和Kafka的对比，还介绍了TubeMQ内部的存储结构，但是我发现它的内部存储结构和Kafka的存储没有差别，你们只多了一个备份缓存，我不知道为什么你们只是一个备份问题就可以把Kafka甩这么远？A：Kafka是基于Partition的结构，一个Partition就会有一个文件块，而TubeMQ是基于Topic的，Partition已经是一个逻辑的概念。第二个不同是我们的内存是主备模式，你刚才已经提到了，为什么多了一个内存块就会快一些？写内存更快基本上是共识，然后把一块盘写满，写满了的切为备块异步去刷到文件，然后换块内存继续写，这样主备切换的话读写冲突就少了很多，整体就会更快一些。我们为什么改为这样的存储结构呢？像Kafka，1000×10的时候就已经变成了随机读写，跑起来数据指标不是很好，而且也不稳定。RocketMQ是所有数据存储在一个文件，每一个Partition又构造了一个文件，这样子就带来一个问题：数据文件会有写入瓶颈，遇到流量增长时整个系统指标就上不来了。JMQ是按Topic定义数据文件，但每个Partition定义新的文件，它比RocketMQ更宽泛一点，它数据不会集中到一个文件，它是按照Topic来的，解决了RokcetMQ的一些问题。TubeMQ是怎样呢？TubeMQ是一个Topic一个数据文件，不同的Topic有不同的文件，我们没有Partition。我们都是按Topic来定义存储单元的，一个数据文件 + 一个索引文件。大家可以去分析一下，它们是各有特点，不同的场景下的表现特征是不一样的，包括你的硬件场景，其实还是有很大差异的。

导语 | 近日，云+社区技术沙龙“腾讯开源技术”圆满落幕。本次沙龙邀请了多位腾讯技术专家围绕腾讯开源与各位开发者进行探讨，深度揭秘了腾讯开源项目TencentOS tiny、TubeMQ、Kona JDK、TARS以及MedicalNet。本文是对张国成老师演讲的整理。本文要点：Message Queue 的原理和特点；TubeMQ相关实现原理及使用介绍；TubeMQ后续的发展和探讨。一、Message Queue 简介对于Message Queue(以下简称MQ)，Wiki百科上的定义指：不同进程之间或者相同进程不同线程之间的一种通讯方式，它是一种通讯方式。那我们为什么要采用MQ呢？这是由MQ的特点来决定的。第一是因为它可以整合多个不同系统共同协作；第二是它可以解耦，进行数据传递和处理；第三是它可以做峰值的缓冲处理，我们平常接触到的像Kafka、RocketMQ、Pulsar等基本上也都有这样的特点。那作为大数据场景下的MQ又有什么特点呢？从我个人的理解来说，就是高吞吐低延时，系统尽可能地稳定，成本尽可能地低，协议也不需要特别地复杂，特别是水平扩展能力要尽可能的高。因为像海量数据基本上都是到百亿、千亿、万亿，比方说我们自己的生产环境可能一个月、一年的时间就会翻一番，如果没有横向的扩展能力，系统就很容易出现各种问题。二、TubeMQ实现原理及使用介绍1.TubeMQ的特点那么，腾讯自研的TubeMQ又有什么样的特点呢？TubeMQ属于万亿级分布式消息中间件，专注于海量数据下的数据传输和存储，在性能、可靠性，和成本方面具有独特的优势。针对大数据场景的应用，我们给出了一个测试方案（详细方案在腾讯TubeMQ开源docs目录里可以查询得到）。首先我们要做一个实际应用场景定义，在这个定义之下，再进行系统数据的收集整理。结果是：我们的吞吐量达到了14万的TPS，消息时延可以做到5ms以内。可能有些人会感到好奇，因为有很多研究报告都分析过，同样的分布式发布订阅消息系统，例如Kafka这种，都能达到上百万的TPS。相比来说，我们的数据会不会显得太差？其实这里是有一个前提的，那就是：我们是在1000个Topic中，并为每个Topic配置10个Partition的场景下达到的性能指标。对于Kafka，它或许可以达到百万级的TPS，或许时延可以降到很低，但在我们的大数据的场景下，就远远到不了这个量级。我们的系统在线上已经稳定运行了7年的时间，系统的架构采用的是瘦客户端、偏服务器管控模型。这样的技术特点就决定了它相应的应用场景。比如实时的广告推荐、海量的数据上报、指标&监控、流处理、以及IOT场景下的数据上报等。部分数据有损是可以允许的，因为大不了重复再报一下也可以解决，但是考虑到数据的量级实在太大，所以只能牺牲部分的可靠性来换取高性能的数据接入。2.TubeMQ的系统架构 TubeMQ系统架构是怎样的？如下图所示，它通过一个SDK与外部交互，当然直接通过我们定义的TCP协议实现对接也是可以的。值得注意的是：TubeMQ是纯Java编写的，它拥有Master HA的协调节点，采用弱zk来管理Offset。在消息存储模式上也同样进行了改进，调整了数据可靠性的方案，采用的是磁盘RAID10多副本+快速消费，而非多Broker节点多副本的方案，而且启用了元数据自管理模式等。3.TubeMQ的研发历程如下图所示，自有数据记录以来，我们一共经历了四个阶段，从引入到改进再到开始自研，再到现在的自我创新。从2013年6月最开始的200亿，到2019年11月的35万亿，预计2020年能达到40万亿。总的来说，这也是我们当初要选择自研的重要的原因。因为当数据量还不大的时候，比如在10亿或者是10亿量级以下，用什么样MQ其实都是可以的。但是一旦达到百亿、百亿到千亿、千亿到万亿甚至是万亿以上数据量的时候，越来越多的制约因素就会相继出现，包括系统的稳定性、性能以及机器成本和运维成本等问题。我们现网TubeMQ拥有1500台的机器，而1500台机器就只需要用到一个左右的人力来运维就可以了（非全职的两个人）。对于我们的Broker存储服务器，能够做到上线持续运行4个月不重启，这些都是我们在原基础之上所做的改进和创新带来的。4.TubeMQ与其他MQ的横向比较下图的这个表格是TubeMQ与其他MQ横向比较的数据情况，我们项目是开源的，大家如果感兴趣的话可以直接去验证一下。总的来说，TubeMQ比较适合需要高性能、低成本又容许极端情况下有数据损失的场景，是经得起实践考验的。

下单功能时序图下单基本流程下单接口2 校验订单这是下单流程重要的第一步按如下流程执行代码3 生成预订单当校验订单成功后，我们就可以开始生成预订单了流程图注意设置订单 id 时，可能涉及分库分表，所以要注意确保 id 的唯一性哦，使用 雪花算法4 扣减库存牵涉到库存表这里经常需要注意超卖问题

1.为什么使用redis大量的数据请求可能造成数据库的宕机(如秒杀，热点数据，)，redis是使用内存存贮，每秒的运行次数可以达到10w+,可以减少数据库的压力。2.Redis有哪些数据结构？String、Hash、List、Set、SortedSet。3.如果有大量的key需要设置同一时间过期，一般需要注意什么？要设置随机的过期时间，否则造成雪崩。4.Redis分布式锁怎么实现？使用set语法，加入jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime); 也可以用setnx然后在使用expire设置过期时间，但是如果在setnx后服务挂掉了，就会造成死锁。5.假如Redis里面有1亿个key，其中有10w个key是以某个固定的已知的前缀开头的，如何将它们全部找出来？使用keys指令。6.如果这个redis正在给线上的业务提供服务，那使用keys指令会有什么问题？keys指令会导致线程阻塞一段时间，线上服务会停顿，直到指令执行完毕，服务才能恢复。7.Redis怎么做异步队列？使用rpush生产消息，lpop消费消息。当lpop没有消息的时候，要适当sleep一会再重试。8.可不可以不用sleep呢？list还有个指令叫blpop，在没有消息的时候，它会阻塞住直到消息到来。9.能不能生产一次消费多次呢？使用pub/sub主题订阅者模式，可以实现 1:N 的消息队列。10.pub/sub有什么缺点？在消费者下线的情况下，生产的消息会丢失，得使用专业的消息队列如RocketMQ等。11.Redis如何实现延时队列？使用zset score放入要延时的时间 ，然后使用rangebyscore轮询获取。12.能不能生产一次消费多次呢？使用pub/sub。redisTemplate.convertAndSend(“testkafka”, data); 放入队列。13.缓存雪崩，缓存击穿，缓存穿透？请参考上文。14.怎么保持系统高可用？Redis 高可用，主从+哨兵，Redis cluster，避免全盘崩溃。本地 ehcache 缓存 + Hystrix 限流+降级，避免** MySQL** 被打死。Redis 持久化 RDB+AOF，一旦重启，自动从磁盘上加载数据，快速恢复缓存数据。15.Redis是怎么持久化的？服务主从数据怎么交互的？请参考上文。16.是否使用过Redis集群，集群的高可用怎么保证，集群的原理是什么？Redis Cluster:主要为扩展分片 当内存不足使用Cluster扩展。Redis Sentinal：高可用，当主机死掉后选举从机为主机。17.Redis的同步机制？首先进行RDB全局持久化存放于内存，然后将RDB文件发送给从机，同时，在接下来使用AOF逐步同步。18.Pipeline有什么好处，为什么要用pipeline？redis管道，可以将多次请求放于pipeline，然后一起读取出来，这样可以增加redis吞吐量。19.缓存有哪些类型？本地缓存就是在进程的内存中进行缓存，比如我们的 JVM 堆中，可以用 LRUMap 来实现，也可以使用 Ehcache 这样的工具来实现。本地缓存是内存访问，没有远程交互开销，性能最好，但是受限于单机容量，一般缓存较小且无法扩展。分布式缓存一般都具有良好的水平扩展能力，对较大数据量的场景也能应付自如。缺点就是需要进行远程请求，性能不如本地缓存。为了平衡这种情况，实际业务中一般采用多级缓存，本地缓存只保存访问频率最高的部分热点数据，其他的热点数据放在分布式缓存中。20.淘汰策略？不管是本地缓存还是分布式缓存，为了保证较高性能，都是使用内存来保存数据，由于成本和内存限制，当存储的数据超过缓存容量时，需要对缓存的数据进行剔除。一般的剔除策略有 FIFO 淘汰最早数据、LRU 剔除最近最少使用、和 LFU 剔除最近使用频率最低的数据几种策略。noeviction:返回错误当内存限制达到并且客户端尝试执行会让更多内存被使用的命令（大部分的写入指令，但DEL和几个例外）allkeys-lru: 尝试回收最少使用的键（LRU），使得新添加的数据有空间存放。volatile-lru: 尝试回收最少使用的键（LRU），但仅限于在过期集合的键,使得新添加的数据有空间存放。allkeys-random: 回收随机的键使得新添加的数据有空间存放。volatile-random: 回收随机的键使得新添加的数据有空间存放，但仅限于在过期集合的键。volatile-ttl: 回收在过期集合的键，并且优先回收存活时间（TTL）较短的键,使得新添加的数据有空间存放。如果没有键满足回收的前提条件的话，策略volatile-lru, volatile-random以及volatile-ttl就和noeviction 差不多了。21.怎么解决redis瓶颈？使用Redis cluster，主从同步读写分离，类似Mysql的主从同步，Redis cluster 支撑 N 个 Redis master node，每个master node都可以挂载多个 slave node。22.主从同步，能说一下主从之间的数据怎么同步的么？你启动一台slave 的时候，他会发送一个psync命令给master ，如果是这个slave第一次连接到master，他会触发一个全量复制。master就会启动一个线程，生成RDB快照，还会把新的写请求都缓存在内存中，RDB文件生成后，master会将这个RDB发送给slave的，slave拿到之后做的第一件事情就是写进本地的磁盘，然后加载进内存，然后master会把内存里面缓存的那些新命名都发给slave。23.Redis的过期策略？是有定期删除+惰性删除两种。定期好理解，默认100ms就随机抽一些设置了过期时间的key，去检查是否过期，过期了就删了。惰性策略：请求时判断是否过去。24.为啥不扫描全部设置了过期时间的key呢？假如Redis里面所有的key都有过期时间，都扫描一遍？那太恐怖了，而且我们线上基本上也都是会设置一定的过期时间的。全扫描跟你去查数据库不带where条件不走索引全表扫描一样，100ms一次，Redis累都累死了。25.如果一直没随机到很多key，里面不就存在大量的无效key了？好问题，惰性删除，见名知意，惰性嘛，我不主动删，我懒，我等你来查询了我看看你过期没，过期就删了还不给你返回，没过期该怎么样就怎么样。26.数据传输的时候断网了或者服务器挂了怎么办啊？传输过程中有什么网络问题啥的，会自动重连的，并且连接之后会把缺少的数据补上的。27.如果你多个系统同时操作（并发）Redis带来的数据问题？如果第一个请求网络延迟了 可能覆盖之后的请求，造成数据不一致28.这样应该怎么办 ？1 使用分布式锁2 输入时加入时间戳 通过时间戳判断29.数据一致性问题怎么解决？在不要求效率的情况下，可以放入队列中。 但是会引起阻塞。30.你了解最经典的KV、DB读写模式么？读缓存 ，缓存没有读db,保存到缓存更新 先更新数据库在删除缓存31.为什么是删除缓存，而不是更新缓存？缓存来源不一定是一个操作，而且如果多次操作db,同时多次操作db同时多次更新缓存会造成很多冷数据。32.Redis 的线程模型了解么?Redis 内部使用文件事件处理器 file event handler，这个文件事件处理器是单线程的，所以 Redis 才叫做单线程的模型。它采用 IO 多路复用机制同时监听多个 Socket，根据 Socket 上的事件来选择对应的事件处理器进行处理。文件事件处理器的结构包含 4 个部分：多个 Socket IO 多路复用程序 文件事件分派器 事件处理器（连接应答处理器、命令请求处理器、命令回复处理器） 多个 Socket 可能会并发产生不同的操作，每个操作对应不同的文件事件，但是 IO 多路复用程序会监听多个 Socket，会将 Socket 产生的事件放入队列中排队，事件分派器每次从队列中取出一个事件，把该事件交给对应的事件处理器进行处理。33.缓存的问题1.数据不一致第二个问题是数据不一致的问题，可以说只要使用缓存，就要考虑如何面对这个问题。缓存不一致产生的原因一般是主动更新失败，例如更新 DB 后，更新 Redis 因为网络原因请求超时；或者是异步更新失败导致。解决的办法是，如果服务对耗时不是特别敏感可以增加重试；如果服务对耗时敏感可以通过异步补偿任务来处理失败的更新，或者短期的数据不一致不会影响业务，那么只要下次更新时可以成功，能保证最终一致性就可以。2.部署比如 Redis 是单线程处理请求，应尽量避免耗时较高的单个请求任务，防止相互影响；Redis 服务应避免和其他 CPU 密集型的进程部署在同一机器；或者禁用 Swap 内存交换，防止 Redis 的缓存数据交换到硬盘上，影响性能。

package org.apache.rocketmq.spring.core; import org.springframework.messaging.Message; public interface RocketMQLocalTransactionListener { // 用来执行本地事务 RocketMQLocalTransactionState executeLocalTransaction(final Message msg, final Object arg); // 本地事务的检查接口,即MQServer没有收到二次确认消息时调用的方法 // 去检查本地事务到底有没有成功 RocketMQLocalTransactionState checkLocalTransaction(final Message msg); } 具体应用参见Spring Cloud Alibaba实战

1 Producer一个应用尽可能用一个Topic,消息子类型用tags来标识，tags可以由应用自由设置只有发送消息设置了tags,消费方在订阅消息时，才可以利用tags在broker做消息过滤message.setTags("TagA"); 如有可靠性需要，消息发送成功或者失败，要打印消息日志(sendresult和key信 息)如果相同性质的消息量大，使用批量消息，可以提升性能建议消息大小不超过512KBsend(msg)会阻塞，如果有性能要求，可以使用异步的方式: send(msg, callback)如果在一个JVM中，有多个生产者进行大数据处理，建议: 少数生产者使用异步发送方式(3~5个就够了)通过setInstanceName方法，给每个生产者设置一个实例名send消息方法，只要不抛异常，就代表发送成功 , 但是发送成功会有多个状态， 在sendStatus类里定义● SEND_ OK : 消息发送成功● FLUSH_ DISK_ TIMEOUT: 消息发送成功， 但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失● FLUSH SLAVE_ TIMEOUT: 消息发送成功，但是服务器同步到Slave时超时,消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失● SLAVE_ NOT_ AVAILABLE: 消息发送成功， 但是此时slave不可用， 消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失● 如果状态是FLUSH_ DISK_ TIMEOUT或FLUSH SLAVE_ _TIMEOUT,并且Broker正好关闭此时，可以丢弃这条消息，或者重发。但建议最好重发，由消费端去重● Producer向Broker发送请求会等待响应，但如果达到最大等待时间，未得到响应，则客户端将抛出RemotingTimeoutException● 默认等待时间是3秒，如果使用send(msg, timeout),则可以自己设定超时时间,但超时时间不能设置太小，应为Borker需要一些时间来刷新磁盘或与从属设备同步● 如果该值超过syncFlushTimeout,则该值可能影响不大，因为Broker可能会在超时之前返回FLUSH_ SLAVE_ TIMEOUT或FLUSH_ SLAVE_ TIMEOUT的响应8.对于消息不可丢失应用，务必要有消息重发机制Producer的send方法本身支持内部重试:● 至多重试3次● 如果发送失败，则轮转到下一-个Broker● 这个方法的总耗时时间不超过sendMsgTimeout设置的值，默认10s所以，如果本身向broker发送消息产生超时异常,就不会再做重试以上策略仍然不能保证消息一定发送成功，为保证消息一定成功，建议将消息存储到db,由后台线程定时重试，保证消息一定到达Broker2 Consumer每个消息在业务层面的唯一标识码，要设置到keys字段，方便将来定位消息丢失问题服务器会为每个消息创建索引(哈希索引)，应用可以通过topic, key来查询这条消息内容,以及消息被谁消费由于是哈希索引，请务必保证key尽可能唯一,这样可以避免潜在的哈希冲突String orderld =“1250689524981"; message.setKeys(orderld);console客户端使GUImvn clean package -Dmaven.test.skip=true--server.port=8081 --rocketmq.config. namesrvAddr=192.168.1.17:9876http://localhost:8081/#/2.1 消费者组和订阅不同的消费群体可以独立地消费同样的主题，并且每个消费者都有自己的消费偏移量(offsets) 。确保同一组中的每个消费者订阅相同的主题2.2 消息监听器(MessageListener)2.2.1 顺序 (Orderly)消费者将锁定每个MessageQueue,以确保每个消息被一个按顺序使用。这将导致性能损失如果关心消息的顺序时，它就很有用了。不建议抛出异常，可以返回ConsumeOrderlyStatus. SUSPEND_ CURRENT_ QUEUE_ A_ MOMENT代替2.2.2 消费状态(Consume Status)对于MessageListenerConcurrently,可以返回RECONSUME_ LATER告诉消费者，当前不能消费它并且希望以后重新消费。然后可以继续使用其他消息对于MessageListenerOrderly, 如果关心顺序，就不能跳过消息，可以返回SUSPEND_ CURRENT_ QUEUE_ A_ MOMENT来告诉消费者等待片刻。阻塞(Blocking)不建议阻塞Listener,因为它会阻塞线程池，最终可能会停止消费程序线程数DefaultMQPushConsumer消费者使用一个ThreadPoolExecutor来处理内部的消费，因此可以通过设置更改它从何处开始消费● 当建立一个新的Consumer Group时，需要决定是否需要消费Broker中已经存在的历史消息。● CONSUME_ FROM LAST_ OFFSET将忽略历史消息，并消费此后生成的任何内容。● CONSUME_ FROM_ FIRST_ OFFSET将消耗Broker中存在的所有消息。还可以使用CONSUME_ FROM_ TIMESTAMP 来消费在指定的时间戳之后生成的消息。重复(幂等性)RocketMQ无法避免消息重复，如果业务对重复消费非常敏感，务必在业务层面做去重:● 通过记录消息唯一键进行去重● 使用业务层面的状态机制去重3 最佳实践之 NameServer在Apache RocketMQ中，NameServer用于协调分布式系统的每个组件，主要通过管理主题路由信息来实现协调。管理由两部分组成:Brokers定期更新保存在每个名称服务器中的元数据名称服务器是为客户端提供最新的路由信息服务的，包括生产者、消费者和命令行客户端。因此，在启动brokers和clients之前，我们需要告诉他们如何通过给他们提供的一个名称服务器地址列表来访问名称服务器。在Apache RocketMQ中，可以用四种方式完成。3.1 编程方式对于brokers,我们可以在broker的配置文件中指定namesrvAddr=name-server-ip1:port;name-server-ip2:port对于生产者和消费者，我们可以给他们提供姓名服务器地址列表如下:DefaultMQProducer producer = new DefaultMQProducer(" please_ rename_ unique_ group name"); producer.setNamesrvAddr("name-server1-ip:port;name-server2-ip:port"); DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(" please_ rename_ unique_ _group_ name"); consumer.setNamesrvAddr(" name-server1-ip:port;name-server2-ip:port");如果从shell中使用管理命令行，也可以这样指定:sh mqadmin command-name -n name-server-ip1:port;name-server-ip2:port -X OTHER-OPTION一个简单的例子，在NameServer节点上查询集群信息:sh mqadmin -n localhost:9876 clusterList如果将管理工具集成到自己的项目中，可以这样DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(" please_ rename_ _unique_ group_ _name"); defaultMQAdminExt.setNamesrvAddr("name-server1-ip:port;name-server2-ip:port");3.2 Java参数NameServer的地址列表也可以通过java参数rocketmq.namesrv.addr在启动之前指定3.3 环境变量可以设置NAMESRV_ ADDR环境变量。如果设置了，Broker和clients将检 查并使用其值3.4 HTTP端点(HTTP Endpoint)如果没有使用前面提到的方法指定NameServer地址列表，Apache RocketMQ将每2分钟发送一次HTTP请求，以获取和更新NameServer地址列表，初始延迟10秒。默认情况下，访问的HTTP地址是:http://jmenv.tbsite.net:8080/rocketmq/nsaddr通过Java参数rocketmq.namesrv.domain,可以修改jmenv.tbsite.net通过Java参数rocketmq.namesrv.domain.subgroup,可以修改nsaddr3.5 优先级编程方式> Java参数>环境变量> HTTP方式4 JVM与Linux内核配置4.1 JVM配置推荐使用JDK 1.8版本，使用服务器编译器和8g堆。设置相同的Xms和Xmx值，以防止JVM动态调整堆大小以获得更好的性能。简单的JVM配置如下所示:-server -Xms8g -Xmx8g -Xmn4g如果不关心Broker的启动时间，可以预先触摸Java堆，以确保在JVM初始化期间分配页是更好的选择。-XX:+AlwaysPreTouch禁用偏置锁定可能会减少JVM暂停:-XX: UseBiasedL ocking对于垃圾回收，建议使用G1收集器:-XX:+UseG1GC -XX:G1HeapRegionSize= 16m -XX:G lReservePercent=25 -XX:InitiatingHeapOccupancyPercent=30这些GC选项看起来有点激进,但事实证明它在生产环境中具有良好的性能。-XX:MaxGCPauseMillis不要设置太小的值，否则JVM将使用一个小的新生代，这将导致非常频繁的新生代GC。推荐使用滚动GC日志文件:-XX:+UseGCLogFileRotation -Xx:NumberOfGCLogFiles=5 -XX:GCLogFileSize=30m如果写入GC文件会增加代理的延迟，请将重定向GC日志文件考虑在内存文件系统中:-Xloggc:/dev/shm/mq_ gc. _%p.log4.2 Linux内核配置在bin目录中，有一个os.sh脚本列出了许多内核参数，只需要稍微的修改，就可以用于生产环境。以下参数需要注意，详细信息请参考https://www.kernel.org/doc/Documentation/sysctl/vm.txt

1 批量消息1.1 为什么使用批量消息在很多调优的时候，比如数据库批量处理，有些请求进行合并发送等都是类似批量的实现RocketMQ批量发送也是为了追求性能，特别在消息数量特别大的时候，批量效果就非常明显1.2 使用批量消息的限制同一批次的消息应该具有相同主题、相同的消息配置不支持延迟消息建议一个批量消息大小最好不要超过1MB2 事务消息2.1 什么是事务消息RocketMQ的事务消息，是指Producer端消息发送事件和本地事务事件，同时成功/失败。2.2 事务消息设计2.3 事务消息的使用约束事务消息不支持定时和批量为了避免一个消息被多次检查，导致半数队列消息堆积RocketMQ限制了单个消息的默认检查次数为15次通过修改broker配置文件中的transactionCheckMax参数进行调整特定的时间段之后才检查事务通过broker配置文件参数transactionTimeout或用户配置CHECK_ IMMUNITY_ TIME_ IN_ SECONDS调整时间一个事务消息可能被检查或消费多次提交过的消息重新放到用户目标主题可能会失败事务消息的生产者ID不能与其他类型消息的生产者ID共享2.4 事务消息的状态TransactionStatus.CommitTransaction提交事务，允许消费者消费这个消息TransactionStatus. RollbackTransaction回滚事务，消息将会被删除或不再允许消费Transaction Status.Unknown中间状态，MQ需要重新检查来确定状态

1 RocketMQ的发布订阅1.1 基本概念发布订阅模式又叫观察者模式，它定义对象间的一种一对多的依赖关系，当一个对象的状态发生改变时，所有依赖它的对象都将得到通知。RocketMQ的消息订阅分为两种模式Push模式 (MQPushConsumer) : Broker主动向消费者推送Pull模 式(MQPullConsumer) :消费者在需要消息时，主动到Broker拉取但是，在RocketMQ中，具体实现时Push和Pull模式都是采用消费端主动从Broker拉取消息]pull很复杂,建议使用push2 RocketMQ 订阅模式实现原理2.1 Push(推模式)红框内流程已经帮助我们实现了,所以使用起来很简单!2.2 Pull (拉模式)Pull方式里，取消息的过程需要用户自己写。首先通过打算消费的Topic拿到MessageQueue的集合，遍历MessageQueue集合。然后针对每个MessageQueue批量取消息，一次取完后，记录该队列下- -次要取的开始offset，直到取完了，再换另一个MessageQueue。3 使用订阅模式官方教程: http://rocketmq.apache.org/docs/broadcast example/4 定时消息4.1 基本概念定时消息是指消息发到Broker后，不能立刻被Consumer消费，要到特定的时间点或者等待特定的时间后才能被消费。如果要支持任意的时间精度，在Broker层面，必须要做消息排序，如果再涉及到持久化，那么消息排序要不可避免的产生巨大性能开销。RocketMQ支持定时消息，但是不支持任意时间精度，支持特定的level,例如定时5s，10s， 1m等。4.2 延迟级别在Pro端Broker定时消息发送逻辑源码实现import java.util.HashMap; import java.util.Iterator; import java.util.Map; import java.util.Map.Entry; import java.util.Timer; import java.util.TimerTask; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import org.apache.rocketmq.common.ConfigManager; import org.apache.rocketmq.common.TopicFilterType; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; import org.apache.rocketmq.common.message.MessageAccessor; import org.apache.rocketmq.common.message.MessageConst; import org.apache.rocketmq.common.message.MessageDecoder; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.running.RunningStats; import org.apache.rocketmq.store.ConsumeQueue; import org.apache.rocketmq.store.ConsumeQueueExt; import org.apache.rocketmq.store.DefaultMessageStore; import org.apache.rocketmq.store.MessageExtBrokerInner; import org.apache.rocketmq.store.PutMessageResult; import org.apache.rocketmq.store.PutMessageStatus; import org.apache.rocketmq.store.SelectMappedBufferResult; import org.apache.rocketmq.store.config.StorePathConfigHelper; public class ScheduleMessageService extends ConfigManager { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); public static final String SCHEDULE_TOPIC = "SCHEDULE_TOPIC_XXXX"; private static final long FIRST_DELAY_TIME = 1000L; private static final long DELAY_FOR_A_WHILE = 100L; private static final long DELAY_FOR_A_PERIOD = 10000L; private final ConcurrentMap<Integer /* level */, Long/* delay timeMillis */> delayLevelTable = new ConcurrentHashMap<Integer, Long>(32); private final ConcurrentMap<Integer /* level */, Long/* offset */> offsetTable = new ConcurrentHashMap<Integer, Long>(32); private final Timer timer = new Timer("ScheduleMessageTimerThread", true); private final DefaultMessageStore defaultMessageStore; private int maxDelayLevel; public ScheduleMessageService(final DefaultMessageStore defaultMessageStore) { this.defaultMessageStore = defaultMessageStore; } public static int queueId2DelayLevel(final int queueId) { return queueId + 1; } public static int delayLevel2QueueId(final int delayLevel) { return delayLevel - 1; } public void buildRunningStats(HashMap<String, String> stats) { Iterator<Entry<Integer, Long>> it = this.offsetTable.entrySet().iterator(); while (it.hasNext()) { Entry<Integer, Long> next = it.next(); int queueId = delayLevel2QueueId(next.getKey()); long delayOffset = next.getValue(); long maxOffset = this.defaultMessageStore.getMaxOffsetInQueue(SCHEDULE_TOPIC, queueId); String value = String.format("%d,%d", delayOffset, maxOffset); String key = String.format("%s_%d", RunningStats.scheduleMessageOffset.name(), next.getKey()); stats.put(key, value); } } private void updateOffset(int delayLevel, long offset) { this.offsetTable.put(delayLevel, offset); } public long computeDeliverTimestamp(final int delayLevel, final long storeTimestamp) { Long time = this.delayLevelTable.get(delayLevel); if (time != null) { return time + storeTimestamp; } return storeTimestamp + 1000; } public void start() { for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) { Integer level = entry.getKey(); Long timeDelay = entry.getValue(); Long offset = this.offsetTable.get(level); if (null == offset) { offset = 0L; } if (timeDelay != null) { this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME); } } this.timer.scheduleAtFixedRate(new TimerTask() { @Override public void run() { try { ScheduleMessageService.this.persist(); } catch (Throwable e) { log.error("scheduleAtFixedRate flush exception", e); } } }, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval()); } public void shutdown() { this.timer.cancel(); } public int getMaxDelayLevel() { return maxDelayLevel; } public String encode() { return this.encode(false); } public boolean load() { boolean result = super.load(); result = result && this.parseDelayLevel(); return result; } @Override public String configFilePath() { return StorePathConfigHelper.getDelayOffsetStorePath(this.defaultMessageStore.getMessageStoreConfig() .getStorePathRootDir()); } @Override public void decode(String jsonString) { if (jsonString != null) { DelayOffsetSerializeWrapper delayOffsetSerializeWrapper = DelayOffsetSerializeWrapper.fromJson(jsonString, DelayOffsetSerializeWrapper.class); if (delayOffsetSerializeWrapper != null) { this.offsetTable.putAll(delayOffsetSerializeWrapper.getOffsetTable()); } } } public String encode(final boolean prettyFormat) { DelayOffsetSerializeWrapper delayOffsetSerializeWrapper = new DelayOffsetSerializeWrapper(); delayOffsetSerializeWrapper.setOffsetTable(this.offsetTable); return delayOffsetSerializeWrapper.toJson(prettyFormat); } public boolean parseDelayLevel() { HashMap<String, Long> timeUnitTable = new HashMap<String, Long>(); timeUnitTable.put("s", 1000L); timeUnitTable.put("m", 1000L * 60); timeUnitTable.put("h", 1000L * 60 * 60); timeUnitTable.put("d", 1000L * 60 * 60 * 24); String levelString = this.defaultMessageStore.getMessageStoreConfig().getMessageDelayLevel(); try { String[] levelArray = levelString.split(" "); for (int i = 0; i < levelArray.length; i++) { String value = levelArray[i]; String ch = value.substring(value.length() - 1); Long tu = timeUnitTable.get(ch); int level = i + 1; if (level > this.maxDelayLevel) { this.maxDelayLevel = level; } long num = Long.parseLong(value.substring(0, value.length() - 1)); long delayTimeMillis = tu * num; this.delayLevelTable.put(level, delayTimeMillis); } } catch (Exception e) { log.error("parseDelayLevel exception", e); log.info("levelString String = {}", levelString); return false; } return true; } class DeliverDelayedMessageTimerTask extends TimerTask { private final int delayLevel; private final long offset; public DeliverDelayedMessageTimerTask(int delayLevel, long offset) { this.delayLevel = delayLevel; this.offset = offset; } @Override public void run() { try { this.executeOnTimeup(); } catch (Exception e) { // XXX: warn and notify me log.error("ScheduleMessageService, executeOnTimeup exception", e); ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask( this.delayLevel, this.offset), DELAY_FOR_A_PERIOD); } } /** * 纠正可投递时间 * 因为发送级别对应的发送间隔可以调整，如果超过当前间隔，则修正成当前配置，避免后面的消息无法发送。 * @param now 当前时间 * @param deliverTimestamp 投递时间 * @return 纠正结果 */ private long correctDeliverTimestamp(final long now, final long deliverTimestamp) { long result = deliverTimestamp; long maxTimestamp = now + ScheduleMessageService.this.delayLevelTable.get(this.delayLevel); if (deliverTimestamp > maxTimestamp) { result = now; } return result; } public void executeOnTimeup() { // 获取消费队列 ConsumeQueue cq = ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(SCHEDULE_TOPIC, delayLevel2QueueId(delayLevel)); long failScheduleOffset = offset; if (cq != null) { SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset); if (bufferCQ != null) { try { long nextOffset = offset; int i = 0; ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit(); for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) { long offsetPy = bufferCQ.getByteBuffer().getLong(); int sizePy = bufferCQ.getByteBuffer().getInt(); long tagsCode = bufferCQ.getByteBuffer().getLong(); if (cq.isExtAddr(tagsCode)) { if (cq.getExt(tagsCode, cqExtUnit)) { tagsCode = cqExtUnit.getTagsCode(); } else { //can't find ext content.So re compute tags code. log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}", tagsCode, offsetPy, sizePy); long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy); tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime); } } long now = System.currentTimeMillis(); long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode); nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE); long countdown = deliverTimestamp - now; // 消息是否达到投递时间 if (countdown <= 0) { MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset( offsetPy, sizePy); if (msgExt != null) { try { // 达到投递时间，发送消息 MessageExtBrokerInner msgInner = this.messageTimeup(msgExt); PutMessageResult putMessageResult = ScheduleMessageService.this.defaultMessageStore .putMessage(msgInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { continue; // 消息发送成功，继续下一轮循环 } else { // 消息发送失败，安排下次定时任务，10000ms后执行 // XXX: warn and notify me log.error( "ScheduleMessageService, a message time up, but reput it failed, topic: {} msgId {}", msgExt.getTopic(), msgExt.getMsgId()); ScheduleMessageService.this.timer.schedule( new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset), DELAY_FOR_A_PERIOD); // 更新进度 ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset); return; } } catch (Exception e) { /* * XXX: warn and notify me */ log.error( "ScheduleMessageService, messageTimeup execute error, drop it. msgExt=" + msgExt + ", nextOffset=" + nextOffset + ",offsetPy=" + offsetPy + ",sizePy=" + sizePy, e); } } } else { // 消息未达到投递时间，安排下次定时任务，N ms后执行 ScheduleMessageService.this.timer.schedule( new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset), countdown); // 更新进度 ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset); return; } } // end of for nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE); ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask( this.delayLevel, nextOffset), DELAY_FOR_A_WHILE); ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset); return; } finally { bufferCQ.release(); } } // end of if (bufferCQ != null) else { // 不存在新消息，安排下次定时任务，100ms后执行 long cqMinOffset = cq.getMinOffsetInQueue(); if (offset < cqMinOffset) { failScheduleOffset = cqMinOffset; log.error("schedule CQ offset invalid. offset=" + offset + ", cqMinOffset=" + cqMinOffset + ", queueId=" + cq.getQueueId()); } } } // end of if (cq != null) // 获取不到消费队列，安排下次定时任务，100ms后执行 ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel, failScheduleOffset), DELAY_FOR_A_WHILE); } /** * 设置消息内容 * @param msgExt * @return */ private MessageExtBrokerInner messageTimeup(MessageExt msgExt) { MessageExtBrokerInner msgInner = new MessageExtBrokerInner(); msgInner.setBody(msgExt.getBody()); msgInner.setFlag(msgExt.getFlag()); MessageAccessor.setProperties(msgInner, msgExt.getProperties()); TopicFilterType topicFilterType = MessageExt.parseTopicFilterType(msgInner.getSysFlag()); long tagsCodeValue = MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags()); msgInner.setTagsCode(tagsCodeValue); msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties())); msgInner.setSysFlag(msgExt.getSysFlag()); msgInner.setBornTimestamp(msgExt.getBornTimestamp()); msgInner.setBornHost(msgExt.getBornHost()); msgInner.setStoreHost(msgExt.getStoreHost()); msgInner.setReconsumeTimes(msgExt.getReconsumeTimes()); msgInner.setWaitStoreMsgOK(false); MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_DELAY_TIME_LEVEL); msgInner.setTopic(msgInner.getProperty(MessageConst.PROPERTY_REAL_TOPIC)); String queueIdStr = msgInner.getProperty(MessageConst.PROPERTY_REAL_QUEUE_ID); int queueId = Integer.parseInt(queueIdStr); msgInner.setQueueId(queueId); return msgInner; } } }参考http://rocketmq.apache.org/docs/broadcast-example/