下载文档原格式
RocketMQ事务消息实现分析这张图说明了事务消息的⼤致⽅案，分为两个逻辑：正常事务消息的发送及提交、事务消息的补偿流程事务消息发送及提交：发送消息（half消息）服务端响应消息写⼊结果根据发送结果执⾏本地事务（如果写⼊失败，此时half消息对业务不可见，本地逻辑不执⾏）根据本地事务状态执⾏Commit或者Rollback（Commit操作⽣成消息索引，消息对消费者可见）补偿流程：对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起⼀次“回查”Producer收到回查消息，检查回查消息对应的本地事务的状态根据本地事务状态，重新Commit或者Rollback补偿阶段⽤于解决消息Commit或者Rollback发⽣超时或者失败的情况。
以上RocketMQ事务消息的整体⽅案，对于了解Notify的同学应该是很熟悉的，下⾯是之前Notify相关的资料：整体⽅案是完全相同的，只是两者的Storage不同。
RocketMQ事务消息设计⼀阶段的消息如何对⽤户不可见事务消息相对普通消息最⼤的特点就是⼀阶段发送的消息对⽤户是不可见的。
如何做到写⼊了消息但是对⽤户不可见?——写⼊消息数据，但是不创建对应的消息的索引信息。
熟悉RocketMQ的同学应该都清楚，消息在服务端的存储结构如上，每条消息都会有对应的索引信息，Consumer通过索引读取消息。
那么实现⼀阶段写⼊的消息不被⽤户消费（需要在Commit后才能消费），只需要写⼊Storage Queue，但是不构建Index Queue即可。
RocketMQ中具体实现策略是：写⼊的如果事务消息，对消息的Topic和Queue等属性进⾏替换，同时将原来的Topic和Queue信息存储到消息的属性中。
RocketMQ原理-部署-⼊门（图解）⽂章很长，建议收藏起来，慢慢读！ 为⼩伙伴奉上以下珍贵的学习资源：疯狂创客圈经典升级：⾯试必备 + ⼤⼚必备 + 涨薪必备疯狂创客圈经典图书：⾯试必备 + ⼤⼚必备 +涨薪必备疯狂创客圈经典图书：⾯试必备 + ⼤⼚必备 + 涨薪必备疯狂创客圈资源宝库： Java 必备百度⽹盘资源⼤合集价值>1000元【】RocketMQ简介rocketMQ作为⼀款分布式的消息中间件,RocketMQ作为⼀款分布式的消息中间件（阿⾥的说法是不遵循任何规范的，所以不能完全⽤JMS的那⼀套东西来看它），经历了Metaq1.x、Metaq2.x的发展和淘宝双⼗⼀的洗礼，在功能和性能上远超ActiveMQ。
1.要知道RocketMQ原⽣就是⽀持分布式的，⽽ActiveMQ原⽣存在单点性。
2.RocketMQ可以保证严格的消息顺序，⽽ActiveMQ⽆法保证！3.RocketMQ提供亿级消息的堆积能⼒，这不是重点，重点是堆积了亿级的消息后，依然保持写⼊低延迟！4.丰富的消息拉取模式（Push or Pull） Push好理解，⽐如在消费者端设置Listener回调；⽽Pull，控制权在于应⽤，即应⽤需要主动的调⽤拉消息⽅法从Broker获取消息，这⾥⾯存在⼀个消费位置记录的问题（如果不记录，会导致消息重复消费）。
RocketMQ是什么RocketMQ是⼀个队列模型的消息中间件，具有⾼性能、⾼可靠、⾼实时、分布式特点。
Producer、Consumer、队列都可以分布式。
Producer 向⼀些队列轮流发送消息，队列集合称为 Topic，Consumer 如果做⼴播消费，则⼀个 consumer 实例消费这个 Topic 对应的所有队列，如果做集群消费，则多个 Consumer 实例平均消费这个 topic 对应的队列集合。
能够保证严格的消息顺序提供丰富的消息拉取模式⾼效的订阅者⽔平扩展能⼒实时的消息订阅机制亿级消息堆积能⼒较少的依赖选择RocketMQ的理由强调集群⽆单点，可扩展，任意⼀点⾼可⽤，⽔平可扩展⽅便集群配置，⽽且容易扩展（横向和纵向），通过slave的⽅式每⼀点都可以实现⾼可⽤⽀持上万个队列，顺序消息顺序消费是实现在同⼀队列的，如果⾼并发的情况就需要队列的⽀持，rocketmq可以满⾜上万个队列同时存在任性定制你的消息过滤rocketmq提供了两种类型的消息过滤，也可以说三种可以通过topic进⾏消息过滤、可以通过tag进⾏消息过滤、还可以通过filter的⽅式任意定制过滤消息的可靠性（⽆Buffer，持久化，容错，回溯消费）消息⽆buffer就不⽤担⼼buffer回满的情况，rocketmq的所有消息都是持久化的，⽣产者本⾝可以进⾏错误重试，发布者也会按照时间阶梯的⽅式进⾏消息重发，消息回溯说的是可以按照指定的时间进⾏消息的重新消费，既可以向前也可以向后（前提条件是要注意消息的擦除时间）海量消息堆积能⼒，消息堆积后，写⼊低延迟针对于provider需要配合部署⽅式，对于consumer，如果是集群⽅式⼀旦master返现消息堆积会向consumer下发⼀个重定向指令，此时consumer就可以从slave进⾏数据消费了分布式事务我个⼈感觉rocketmq对这⼀块说的不是很清晰，⽽且官⽅也说现在这块存在缺陷（会令系统pagecache过多），所以线上建议还是少⽤为好消息失败重试机制针对provider的重试，当消息发送到选定的broker时如果出现失败会⾃动选择其他的broker进⾏重发，默认重试三次，当然重试次数要在消息发送的超时时间范围内。
Python：Rocketmq消息队列使⽤rocketmq可以与kafka等⼀起使⽤，⽤于实时消息处理。
安装rocketmq：⽣产消息producer：from rocketmq.client import Producer, Messageimport jsonproducer = Producer('PID-test')producer.set_namesrv_addr('xxx.xxx.xxx.xxx:xxxxx') #rocketmq队列接⼝地址（服务器ip:port）producer.start()msg_body = {"id":"001","name":"test_mq","message":"abcdefg"}ss = json.dumps(msg_body).encode('utf-8')msg = Message('topic_name') #topic名称msg.set_keys('xxxxxx')msg.set_tags('xxxxxx')msg.set_body(ss) #message bodyretmq = producer.send_sync(msg)print(retmq.status, retmq.msg_id, retmq.offset)producer.shutdown()其中：设置ip:port的位置：producer.set_namesrv_addr('xxx.xxx.xxx.xxx:xxxxx')当只有单⼀服务器时，格式是上⾯这个；当有多个服务器地址（集群模式）时，可以使⽤：producer.set_namesrv_addr("xxx.xxx.xxx.xxx:xxxxx,xxx.xxx.xxx.xxx:xxxxx,xxx.xxx.xxx.xxx:xxxxx")不过以下这种⽅式本⼈测试不通过：producer.set_namesrv_addr(["xxx.xxx.xxx.xxx:xxxxx","xxx.xxx.xxx.xxx:xxxxx","xxx.xxx.xxx.xxx:xxxxx"])如果使⽤pandas数据，pandas数据可以直接转换some_df.to_json(orient='records').encode('utf-8')，然后放⼊body中发送。
python实现的消息队列RocketMQ客户端使⽤rocketmq-python 是⼀个基于 rocketmq-client-cpp 封装的 RocketMQ Python 客户端。
⼀、Producer#coding:utf-8import jsonfrom rocketmq.client import Producer, Messageproducer = Producer('PID-001') # 实例化Producer对象，指定group-id（可任意取名）producer.set_namesrv_addr('xxxxxx:xx') # rocketmq队列接⼝地址（服务器ip:port)producer.start() # 开启# 实例化消息对象，需要指定应⽤名:topic_namemsg = Message('your_topic_name') # 实例化消息对象时，传⼊topic名称,⼀个应⽤尽可能⽤⼀个Topic# 指定消息的keysmsg.set_keys('your_keys') # 业务层⾯的唯⼀标识码，⽅便将来定位消息丢失问题。
服务器会为每个消息创建索引（哈希索引），应⽤可以通过topic，key来查询这条消息内容，以及消息被谁消费。
# 指定消息tagsmsg.set_tags('your_tags') # 消息⼦类型⽤tags来标识，tags可以由应⽤⾃由设置。
#指定消息体（内容）msg_body = {'name':'laowang','age':28}body = json.dumps(msg_body).encode('utf-8')msg.set_body(body) # 传⼊消息体（json字节串）# 向队列发送消息ret = producer.send_sync(msg)print(f'status:{ret.status}') # 0表⽰OKprint(f'msg_id:{ret.msg_id}') # 消息id，同消费者获取到的消息idprint(f'offset:{ret.offset}') # 偏移量，默认从0开始，1，2。
消息队列实施方案1、背景异步解耦合、给前端系统提供最高效的反应2、常见消息队列对比2、1 ActiveMqActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规的JMS Provider实现优点：Java语言支持集群模式缺点：性能在消息中间件中处于下游2、2 RabbitmqRabbitmq是基于AMQP使用erlang语言实现的消息队列系统优点：1、完整的消息队列系统，支持多种消息队列模式，包括竞争消费；2、支持集群模式，扩展集群容量和性能比较方便，集成了集群的监控和管理；3、支持消息的持久化；缺点：1、需要学习比较复杂的接口和协议，比较耗费时间；2、性能不是特别理想大概在1wqps左右；3、使用Erlang语言，语言基础；2、3 KafkaKafka 是LinkedIn 开发的一个高性能、分布式的消息发布订阅系统。
优点：1、分布式集群可以透明的扩展，增加新的服务器进集群。
2、高性能。
单机写入TPS约在百万条/秒3、容错。
数据都会复制到几台服务器上。
缺点：1、复杂性。
Kafka需要zookeeper 集群的支持，Topic通常需要人工来创建，部署和维护较一般消息队列成本更高定位于日志传输、存在消息丢失的肯能、消息乱序3、消息发送错误无重试2、4 RocketMQRockerMq 是阿里公司中间件团队参考Kafka思想，用Java语言实现的消息传输系统优点：1、较高性能，单机写入TPS单实例约7万条/秒2、容错，多种集群模式、可以解决容错问题3、消息重试发送4、顺序消息可以严格执行缺点：1、消息重复、消费端需要做去重操作2、5 选用结论从项目业务与团队技术偏向考虑，我们应该需要一种数据安全性比较高，保证每个消息都会被执行；有容错机制、支持集群模式高可用；性能不错，可以在毫秒级处理消息；支持顺序消息的消息中间件，RockerMq 可以满足这些要求。
3、RockerMq简介3、1 RockerMq 产品介绍参考阿里公司提供的《RocketMQ 开发指南》，最新版针对v3.2.43、2 RockerMq集群3、2、1 部署方式Rockermq共有四种部署方式，分别是：1、单个Master一旦Broker 重启或者宕机时，会导致整个服务不可用2、多Master 模式一个集群无Slave，全是Master，例如2 个Master 戒者3 个Master优点：1、配置简单，2、容错，单个Master 宕机或重启维护对应用无影响，在磁盘配置为RAID10 时，即使机器宕机不可恢复情况下，由于RAID10 磁盘非常可靠，在同步刷盘时消息不会丢，异步刷盘丢失少量消息，3、性能最高。
rocketmq的使用RocketMQ是阿里巴巴推出的一种全新的大规模分布式消息系统，它能够支持海量的消息的高效传输和存储。
它具有高性能、高可靠性和高可用性的特点，可以帮助企业更好的实现分布式消息的发布和消费功能。
在本文中，我们将详细介绍RocketMQ的安装、使用、部署、优化以及实际应用示例，为大家提供一个详尽的介绍和实战指南。
一、RocketMQ是什么？RocketMQ是阿里巴巴推出的一种新型公有云消息中间件，它为企业提供最佳的消息服务解决方案，支持海量的消息的高效传输和存储。
RocketMQ有着高性能、高可靠性和高可用性的特点，能够支持用户从消息仓库中取出和发布消息。
RocketMQ支持多种编程语言，如Java、Go、PHP、.NET等，用户可以基于RocketMQ搭建分布式消息处理系统，实现快速、可靠的消息传递服务。
二、RocketMQ安装RocketMQ提供了丰富的安装方式，用户可以根据自身情况来选择安装方式，具体可以参阅官方文档。
我们主要介绍docker安装，docker安装更快更方便，用户可以快速的搭建一个RocketMQ集群。
1、准备docker环境首先要准备一台服务器，至少要有4G内存，32G的硬盘，可以在这台服务器上安装docker环境，并安装docker-compose。
2、安装RocketMQ在服务器上运行docker-compose安装脚本：$ docker-compose up -d等待程序完成安装，RocketMQ就安装完成了。
三、RocketMQ使用1、启动RocketMQ在终端（Windows系统可以使用PowerShell）中输入以下命令： $ docker exec -it [container_name] sh进入容器，启动RocketMQ：$ sh mqnamesrv &sh mqbroker -n localhost:9876 &2、发送消息使用RocketMQ提供的客户端程序可以很容易地发送消息，发送消息命令如下：$ ./mqadmin producer -t Queue -r 0hello3、消费消息消费消息的命令如下：$ ./mqadmin consumer -c consumerId=CID_Test四、RocketMQ部署1、环境准备部署RocketMQ时，需要准备多台服务器，每台服务器配备至少4G内存和32G硬盘。
rocketmq 代码实现的流程RocketMQ is an open-source, distributed messaging and streaming platform that is used for building scalable and reliable messaging systems. It offers low latency, high throughput, and fault-tolerant messaging capabilities, making it a popular choice for real-time data streaming applications, event-driven architectures, and message queuing systems. RocketMQ provides a variety of features, including publish-subscribe messaging, message broadcasting, message filtering, and message tracing. Its design is inspired by the principles of the Apache Kafka and the simplicity of the ActiveMQ.RocketMQ 的设计是受到了 Apache Kafka 的启发，结合了ActiveMQ的简单性。
RocketMQ是一个开源的、分布式的消息传递和流媒体平台，用于构建可扩展和可靠的消息系统。
它提供了低延迟、高吞吐量和容错的消息功能，因此成为实时数据流应用、事件驱动架构和消息队列系统的热门选择。
RocketMQ 提供了各种特性，包括发布-订阅消息、消息广播、消息过滤和消息跟踪。
One key aspect of RocketMQ's design is its highly scalable and fault-tolerant architecture. It is capable of handling large volumes ofmessages with low latency, making it suitable for high-throughput, real-time data processing tasks. RocketMQ achieves this scalability and fault tolerance through a distributed and decentralized design, where messages are stored and processed across multiple nodes in a cluster. This allows RocketMQ to handle failures gracefully and continue to operate reliably even in the face of network partitions or hardware failures.RocketMQ的设计的一个关键方面是它的高度可扩展和容错架构。
基于开源RocketMQ构建网易云音乐消息队列网易云音乐自2013年上线后，业务保持了高速增长。
云音乐除了提供好听的音乐外，还留下了我们在乐和人上的美好回忆。
本文整理自网易云音乐消息队列负责人林德智在近期 Apache Flink&RocketMQ Meetup 上海站的分享，通过该文，您将了解到：•网易云音乐消息队列改造背景•网易云音乐业务对消息队列要求•网易云音乐消息队列架构设计•网易云音乐消息队列部分高级特性介绍•网易云音乐消息队列落地使用情况•网易云音乐消息队列未公开规划背景网易云音乐从13年4月上线以来，业务和用户突飞猛进。
后台技术也从传统的Tomcat 集群到分布式微服务快速演进和迭代，在业务的不断催生下，诞生了云音乐的RPC，API 网关和链路跟踪等多种服务，消息队列也从 RabbitMQ 集群迁移到 Kafka集群。
对于消息队列，更多处于使用阶段，也在使用中出现很多问题。
因此我们期望提供一种完全可控，出现问题我们自己能排查，能跟踪，可以根据业务需求做定制改造的消息队列。
调研结果RabbitMQ 由于持久化场景下的吞吐量只有2.6万，不能满足我们业务吞吐量的需求，云音乐在 2017 年将消息队列从 RabbitMQ 迁移到Kafka 也是这个原因，因此不再考虑范围之内。
由于云音乐整体业务的 QPS 较高，因此，ActiveMQ 也不在考虑范围。
这里主要对比 RocketMQ 与 Kafka：Kafka 更偏向大数据，日志处理，缺少死信，消费失败自动重试，事物消息，定时消息，消息过滤，广播消息等特性，另外Kafka 没有同步刷盘。
云音乐的业务更偏向于多 T opic，死信可溯源，消费失败可收敛自动重试，高可用，自动Failover 等特性。
对于商城和社交业务来说，事物，顺序 Topic 使用会比较多。
Kafka 和RocketMQ 对比：/2016/03/24/rmq-vs-kafka经过 RabbitMQ，Kafka 和 RocketMQ（ ActiveMQ 性能较差，暂不考虑）的调研和分析后，我们发现 RocketMQ 比较适合云音乐的通用业务，但是开源 RocketMQ 也有一些缺陷，只有我们解决了这些缺陷才能在业务中大规模使用。
消息队列解决方案
《消息队列解决方案》
消息队列是一种在软件系统中可靠地传递消息的技术。
它可以帮助系统间或组件之间进行通信和协作，提高系统的可靠性和扩展性。
在现代分布式系统中，消息队列已经成为解决各种通信和协作问题的重要技术。
消息队列解决方案通常包括以下几个方面：
1. 消息发布和订阅：消息队列技术可以帮助系统中的组件进
行消息的发布和订阅，使得不同组件可以将消息发送给订阅了该消息的组件，从而实现组件间的松耦合通信。
2. 消息传递的可靠性：消息队列可以确保消息在传递过程中
不会丢失，并且可以提供一定程度的消息重试和消息持久化，从而保证消息传递的可靠性。
3. 消息队列的扩展性：消息队列通常可以快速地扩展到大规
模的消息处理，并且可以支持分布式部署和横向扩展，从而满足系统在大流量和高并发情况下的需求。
4. 消息处理的并发性：消息队列通常可以支持多个消息处理
者并发处理消息，从而可以提高系统的消息处理效率和吞吐量。
5. 消息队列的监控和管理：消息队列解决方案通常包括一套
完善的监控和管理工具，可以帮助系统管理员和开发人员监控
消息队列的运行状态，进行故障排查和性能优化。
总之，消息队列是现代分布式系统中不可或缺的重要技术，它可以帮助系统实现松耦合通信、提高消息传递的可靠性和效率，并且可以支持系统的可靠性和扩展性。
选择合适的消息队列解决方案对于系统的设计和实现来说非常重要。
消息队列实施方案1、背景异步解耦合、给前端系统提供最高效的反应2、常见消息队列对比2、1 ActiveMqActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规的JMS Provider实现优点：Java语言支持集群模式缺点：性能在消息中间件中处于下游2、2 RabbitmqRabbitmq是基于AMQP使用erlang语言实现的消息队列系统优点：1、完整的消息队列系统，支持多种消息队列模式，包括竞争消费；2、支持集群模式，扩展集群容量和性能比较方便，集成了集群的监控和管理；3、支持消息的持久化；缺点：1、需要学习比较复杂的接口和协议，比较耗费时间；2、性能不是特别理想大概在1wqps左右；3、使用Erlang语言，语言基础；2、3 KafkaKafka 是LinkedIn 开发的一个高性能、分布式的消息发布订阅系统。
优点：1、分布式集群可以透明的扩展，增加新的服务器进集群。
2、高性能。
单机写入TPS约在百万条/秒3、容错。
数据都会复制到几台服务器上。
缺点：1、复杂性。
Kafka需要zookeeper 集群的支持，T opic通常需要人工来创建，部署和维护较一般消息队列成本更高定位于日志传输、存在消息丢失的肯能、消息乱序3、消息发送错误无重试2、4 RocketMQRockerMq 是阿里公司中间件团队参考Kafka思想，用Java语言实现的消息传输系统优点：1、较高性能，单机写入TPS单实例约7万条/秒2、容错，多种集群模式、可以解决容错问题3、消息重试发送4、顺序消息可以严格执行缺点：1、消息重复、消费端需要做去重操作2、5 选用结论从项目业务与团队技术偏向考虑，我们应该需要一种数据安全性比较高，保证每个消息都会被执行；有容错机制、支持集群模式高可用；性能不错，可以在毫秒级处理消息；支持顺序消息的消息中间件，RockerMq 可以满足这些要求。
3、RockerMq简介3、1 RockerMq 产品介绍参考阿里公司提供的《RocketMQ 开发指南》，最新版针对v3.2.43、2 RockerMq集群3、2、1 部署方式Rockermq共有四种部署方式，分别是：1、单个Master一旦Broker 重启或者宕机时，会导致整个服务不可用2、多Master 模式一个集群无Slave，全是Master，例如2 个Master 戒者3 个Master优点：1、配置简单，2、容错，单个Master 宕机或重启维护对应用无影响，在磁盘配置为RAID10 时，即使机器宕机不可恢复情况下，由于RAID10 磁盘非常可靠，在同步刷盘时消息不会丢，异步刷盘丢失少量消息，3、性能最高。
3、多Master 多Slave 模式，异步复制每个Master 配置一个或多个Slave，有多对Master-Slave，HA（高可用集群）采用异步复制方式，主备有短暂消息延迟，毫秒级。
优点：1、即使磁盘损坏，消息丢失的非常少，消息实时性不会被影响，因为Master 宕机后，消费者仍然可以从Slave 消费，此过程对应用透明。
不需要人工干预。
性能同多Master 模式几乎一样。
缺点：1、Master宕机，磁盘损坏时，因为主备有短暂消息延迟，未复制到slave的消息会丢失。
2、目前master宕机后，备机不能自动切换为主机。
只有master可以接收消息，若所有master宕机，将不能接收消息4、多Master 多Slave 模式，同步双写每个Master 配置一个或多个Slave，有多对Master-Slave，HA 采用同步双写方式，主备都写成功，才返回成功。
优点：数据与服务都无单点，Master 宕机情冴下，消费者可以从slave消费、消息无延迟，服务可用性与数据可用性都非常高缺点：1、性能比异步复制模式略低，収送单个消息的RT（返回时间）会略高。
2、目前master宕机后，备机不能自动切换为主机。
只有master可以接收消息，若所有master宕机，将不能接收消息选用结论由于我们需要保证消息中间件的高可用性，消息不丢失、消息无延迟，所以我们选择“多Master多Slave模式，同步双写”模式。
并且选择同步刷盘。
3、2、2 多Master多Slave模式多master多slave模式网络结构图主要组件有：Name Server、Broker、Producer、Consumer1、Name Server是一个几乎无状态节点，可集群部署，节点之间无信息同步、记录Topic 路由信息。
2、Broker分为Master和Slave，一个Master 可以对应多个Slave，但是一个Slave只能对应一个Master。
3、Producer与Name Server 集群中的其中一个节点（随机选择）建立长连接，定期从Name Server获取Topic 路由信息，并向提供Topic服务的Master 建立长连接，定时向Master 发送心跳。
Producer只可以向Master发送消息。
Producer 完全无状态，可集群部署。
4、Consumer与Name Server 集群中的其中一个节点（随机选择）建立长连接，定期从Name Server获取Topic路由信息，并与提供T opic服务的Master、Slave建立长连接，并定时向Master、Slave収送心跳。
Consumer既可以从Master 订阅消息，也可以从Slave 订阅消息，订阅规则由Broker配置决定3、3 集群搭建linux 环境下部署rocketMq多master多slave模式、同步双写模式集群，暂定为2个master，2个slave3、3、1 安装条件4台linux服务器、分为master-a、slave-a ; master-b、slave-b服务器防火墙开启9876，10911lokkit -p 9876:tcp -p 10911:tcp服务器支持wget命令服务器安装jdk，不低于使用的rocketMq的支持版本3、3、2 安装步骤4台linux服务器、分为master-a、slave-a ; master-b、slave-b假设ip分别为：master-a =10.1.236.1slave–a =10.1.236.2master-b =10.1.236.3slave-b =10.1.236.43、3、2、1 master-a1 从github下载RocketMQ安装包或源码自编译安装wget https://github./alibaba/RocketMQ/releases/download/v3.2.6/alibaba-rocketmq-3.2.6.tar.gz 2 解压缩、并创建数据、日志目录tar –xvf alibaba-rocketmq-3.2.2.tar.gz3 配置环境变量:系统变量：Vi /etc/profile或者修改当前用户的环境变量例如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmqexport PATH=${PATH}:${ROCKETMQ_HOME}/binsource 命令是环境变量生效4 修改mq集群的master-a 配置修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties不是强制必须使用这个文件，使用者可以自行定义# brokerClusterName=DefaultClusterbrokerName=broker-a #归属master-slave组的名字brokerId=0 #0表示为master-slave组中为masternamesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876 #nameserv defaultTopicQueueNums=4autoCreateTopicEnable=trueautoCreateSubscriptionGroup=truelistenPort=10911 #Broker 对外服务的监听端口deleteWhen=04fileReservedTime=120mapedFileSizeCommitLog=1073741824mapedFileSizeConsumeQueue=50000000destroyMapedFileIntervalForcibly=120000redeleteHangedFileInterval=120000diskMaxUsedSpaceRatio=88storePathRootDir=/opt/RocketMQ/alibaba-rocketmq/data #数据目录storePathCommitLog=/opt/RocketMQ/alibaba-rocketmq/logs #日志目录maxMessageSize=65536flushCommitLogLeastPages=4flushConsumeQueueLeastPages=2flushCommitLogThoroughInterval=10000flushConsumeQueueThoroughInterval=60000checkTransactionMessageEnable=falsesendMessageThreadPoolNums=128pullMessageThreadPoolNums=128brokerRole=SYNC_MASTER #角色同步双写MasterflushDiskType=SYNC_FLUSH #同步刷盘brokerIP1=10.1.236.1 #本机IP地址，多网卡易出错，请手工指定其他配置请参考《RocketMQ 开发指南》，最新版针对v3.2.45 启动mq集群的master-a跳转到RocketMQ的bin目录下>cd $ROCKETMQ_HOME/bin>nohup sh mqnamesrv &>nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a.properties &3、3、2、2 slave-a1从github下载RocketMQ安装包或源码自编译安装wget https://github./alibaba/RocketMQ/releases/download/v3.2.2/alibaba-rocketmq-3.2.2.tar.gz 2 解压缩、并创建数据、日志目录tar –xvf alibaba-rocketmq-3.2.2.tar.gz3 配置环境变量例如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmqexport PATH=${PATH}:${ROCKETMQ_HOME}/binsource 命令是环境变量生效4修改mq集群的slave-a 配置修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties不是强制必须使用这个文件，使用者可以自行定义、只要保证配置文件的brokerName 正确即可# brokerClusterName=DefaultClusterbrokerName=broker-a #归属master-slave组的名字brokerId=1 #1表示在master-slave组中为slavenamesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876 defaultTopicQueueNums=4autoCreateTopicEnable=trueautoCreateSubscriptionGroup=truelistenPort=10911 #对外端口deleteWhen=04fileReservedTime=120mapedFileSizeCommitLog=1073741824mapedFileSizeConsumeQueue=50000000destroyMapedFileIntervalForcibly=120000redeleteHangedFileInterval=120000diskMaxUsedSpaceRatio=88storePathRootDir=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/data #数据存放storePathCommitLog=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/logs #日志存放maxMessageSize=65536flushCommitLogLeastPages=4flushConsumeQueueLeastPages=2flushCommitLogThoroughInterval=10000flushConsumeQueueThoroughInterval=60000checkTransactionMessageEnable=falsesendMessageThreadPoolNums=128pullMessageThreadPoolNums=128brokerRole=SLAVE #角色SlaveflushDiskType=SYNC_FLUSH # 同步刷盘brokerIP1=10.1.236.2 #本机ip，多网卡，建议自定义其他配置请参考《RocketMQ 开发指南》，最新版针对v3.2.45 启动mq集群的slave-a跳转到RocketMQ的bin目录下>cd $ROCKETMQ_HOME/bin>nohup sh mqnamesrv &>nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-a-s.properties &3、3、2、3 master-b1 从github下载RocketMQ安装包或源码自编译安装wget https://github./alibaba/RocketMQ/releases/download/v3.2.2/alibaba-rocketmq-3.2.2.tar.gz 2 解压缩、并创建数据、日志目录tar –xvf alibaba-rocketmq-3.2.2.tar.gz3 配置环境变量例如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmqexport PATH=${PATH}:${ROCKETMQ_HOME}/binsource 命令是环境变量生效4 修改mq集群的master-b配置修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties不是强制必须使用这个文件，使用者可以自行定义# brokerClusterName=DefaultClusterbrokerName=broker-b #归属master-slave组的名字brokerId=0 #0表示为master-slave组中为masternamesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876 #nameserv defaultTopicQueueNums=4autoCreateTopicEnable=trueautoCreateSubscriptionGroup=truelistenPort=10911 #Broker 对外服务的监听端口deleteWhen=04fileReservedTime=120mapedFileSizeCommitLog=1073741824mapedFileSizeConsumeQueue=50000000destroyMapedFileIntervalForcibly=120000redeleteHangedFileInterval=120000diskMaxUsedSpaceRatio=88storePathRootDir=/opt/RocketMQ/alibaba-rocketmq/data #数据目录storePathCommitLog=/opt/RocketMQ/alibaba-rocketmq/logs #日志目录maxMessageSize=65536flushCommitLogLeastPages=4flushConsumeQueueLeastPages=2flushCommitLogThoroughInterval=10000flushConsumeQueueThoroughInterval=60000checkTransactionMessageEnable=falsesendMessageThreadPoolNums=128pullMessageThreadPoolNums=128brokerRole=SYNC_MASTER #角色同步双写MasterflushDiskType=SYNC_FLUSH #同步刷盘brokerIP1=10.1.236.3 #本机IP地址，多网卡易出错，请手工指定其他配置请参考《RocketMQ 开发指南》，最新版针对v3.2.45 启动mq集群的master-b跳转到RocketMQ的bin目录下>cd $ROCKETMQ_HOME/bin>nohup sh mqnamesrv &>nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b.properties &3、3、2、4 slave-b1从github下载RocketMQ安装包或源码自编译安装wget https://github./alibaba/RocketMQ/releases/download/v3.2.2/alibaba-rocketmq-3.2.2.tar.gz2 解压缩、并创建数据、日志目录tar –xvf alibaba-rocketmq-3.2.2.tar.gz3 配置环境变量例如：export ROCKETMQ_HOME=/opt/RocketMQ/alibaba-rocketmqexport PATH=${PATH}:${ROCKETMQ_HOME}/binsource 命令是环境变量生效4修改mq集群的slave-a 配置修改文件$ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties不是强制必须使用这个文件，使用者可以自行定义、只要保证配置文件的brokerName 正确即可# brokerClusterName=DefaultClusterbrokerName=broker-b #归属master-slave组的名字brokerId=1 #1表示在master-slave组中为slavenamesrvAddr=10.1.236.1:9876;10.1.236.2:9876;10.1.236.3:9876;10.1.236.4:9876 defaultTopicQueueNums=4autoCreateTopicEnable=trueautoCreateSubscriptionGroup=truelistenPort=10911 #对外端口deleteWhen=04fileReservedTime=120mapedFileSizeCommitLog=1073741824mapedFileSizeConsumeQueue=50000000destroyMapedFileIntervalForcibly=120000redeleteHangedFileInterval=120000diskMaxUsedSpaceRatio=88storePathRootDir=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/data #数据存放storePathCommitLog=/aifs01/users/tstusr12/opt/RocketMQ/alibaba-rocketmq/logs #日志存放maxMessageSize=65536flushCommitLogLeastPages=4flushConsumeQueueLeastPages=2flushCommitLogThoroughInterval=10000flushConsumeQueueThoroughInterval=60000checkTransactionMessageEnable=falsesendMessageThreadPoolNums=128pullMessageThreadPoolNums=128brokerRole=SLAVE #角色SlaveflushDiskType=SYNC_FLUSH # 同步刷盘brokerIP1=10.1.236.4 #本机ip，多网卡，建议自定义其他配置请参考《RocketMQ 开发指南》，最新版针对v3.2.45 启动mq集群的slave-b跳转到RocketMQ的bin目录下>cd $ROCKETMQ_HOME/bin>nohup sh mqnamesrv &>nohup sh mqbroker -c $ROCKETMQ_HOME/conf/2m-2s-sync/broker-b-s.properties &4、MQ 消息服务接口实现4.1、流程图 消费端应用消息中间件MQ 服务器Http 请求发送消息消息发送至服务器监听消息并拉取消息通过backurl 将消息推送至消费端返回发送结果4.2、消息中间件接口规此服务接口以dubbo提供的restful协议对外提供发送消息服务，并通过backurl回调消费端把消息推送给消费者，使用此服务可以通过http post请求的方式，消费端要提供接受消息的http协议的post接口。
