RocketMQ介绍
RocketMQ是一个纯java、分布式、队列模型的开源消息中间件,具有以下特点：
能够保证严格的消息顺序
提供丰富的消息拉取模式
高效的订阅者水平扩展能力
实时的消息订阅机制
亿级消息堆积能力
选用理由:
强调集群无单点，可扩展，任意一点高可用，水平可扩展
海量消息堆积能力，消息堆积后，写入低延迟
支持上万个队列
消息失败重试机制
消息可查询
开源社区活跃
成熟度（经过双十一考验）
RocketMQ的各部分角色介绍
角色名称功能
Producer:消息生产者，负责产生消息，一般由业务系统负责产生消息
Consumer:消息消费者，负责消费消息，一般是后台系统负责异步消费
Broker:消息中转角色，负责存储消息，转发消息，一般也称为Server
NameServer:管理中心，一般存储Broker的信息
RocketMQ这四个角色就相当于我们现实生活中的邮政系统，其中Producer、Consumer、 Broker、NameServer分别代表发信者、收信者、负责暂存和传输的邮局、以及协调各个地方邮局的管理机构。
启动RocketMQ之前先要启动NameServer，再启动Broker，这时候消息队列已经在开始工作了。如果想要发送消息，就用Producer；接受消息就用Consumer。如果程序中既要接收也要发送，可以启动多个Producer和Consumer。如果想要增加可靠性或者增大吞吐量，防止单点故障也可以在多台机器上部署多个NameServer和Broker，并且每个Broker也可以部署一个或者过个Slave。
大致了解了基本角色功能后，再介绍两个重要的名词概念Topic(主题)和Message Queue(消息队列)。当一个企业搭建好消息平台后会有多条业务线接入进来，同一个业务也会有不同类型的消息需要投递，如何保证这消息准确地进行，就需要给不同类型的消息加上Topic名称来进行区分。所以在发送消息和接受消息时，需要先创建Topic。有了Topic后，仍然还有性能问题需要考虑。当一个Topic下的消息投递量或者发送量过大怎么办，这就需要在一个Topic下设置一个或者多个Message Queue来提高并行处理速度。有了Message Queue后，消息就可以并行地向各个Message Queue进行分发，从而消费者也可以从多个队列中读取消息，满足性能要求。
RocketMQ单点安装
参照官网:Downloading the Apache RocketMQ Releases - Apache RocketMQ
RocketMQ多级集群部署以及安装
本次先讲如何利用两台物理机，搭建出双主双从无单点故障的高可用RoketMQ集群。假设这两台物理机的ip分别为192.168.218.51和192.168.218.52，端口号默认为9876。
1.启动多个NameServer和Broker
首先按照单点部署，在两台服务器上分别安装RocketMQ，服务地址分别为192.168.218.51:9876和192.168.218.52:9876，然后启动NameServer(nohup sh bin/mqnamesrv &)
启动Broker，每台机器都需启动一个Master角色和一个Slave角色，作为主备。修改的配置文件在安装目录下的conf/2m-2s-sync下。
192.168.218.51机器上的Master Broker配置文件(conf/2m-2s-sync/broker-a.properties)
brokerClusterName=ifind-rocketmq-cluster 所属集群名字，集群比较多可以分成多个Cluster，每个供一个业务使用
brokerName=broker-a broker名字，注意此处不同的配置文件填写的不一样，2选1
brokerId=1 0表示 Master，>0表示 Slave
namesrvAddr=192.168.216.57:9876;192.168.216.61:9876;192.168.216.58:9876;192.168.216.62:9876 nameServer地址，分号分割
defaultTopicQueueNums=4 在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
autoCreateTopicEnable=true 是否允许 Broker自动创建Topic，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true 是否允许 Broker自动创建订阅组，建议线下开启，线上关闭
listenPort=10911 Broker对外服务的监听端口
deleteWhen=04 删除文件时间点，默认凌晨 4点
fileReservedTime=120 文件保留时间，默认 48小时
mapedFileSizeCommitLog=1073741824 commitLog每个文件的大小默认1G
brokerRole=ASYNC_MASTER Broker的角色
flushDiskType=ASYNC_FLUSH 刷盘方式
修改192.168.218.51机器上的Master Broker配置文件(conf/2m-2s-sync/broker-a-s.properties)
修改192.168.218.52机器上的Master Broker配置文件(conf/2m-2s-sync/broker-b.properties)
修改192.168.218.52机器上的Master Broker配置文件(conf/2m-2s-sync/broker-b-s.properties)
几个配置参数的含义
参数名含义
brokerId有三种:SYNCMASTER ASYNCMASTER SLAVE,SYNC表示当Slave和Master消息同步完成 后，再返回发送成功的状态
flushDiskType表示刷盘策略，分为SYNCFLUSH和ASYNCFLUSH两种，代表同步刷盘和异步刷盘。同 步状态下，消息真正写入磁盘才返回成功状态；异步刷盘情况下，消息写入缓存后才返回成功状态
2.发送和接受消息的demo
procucer
public class SyncProducer {
public static void main(String[] args) throws Exception {
//实例化一个生产者
DefaultMQProducer producer = new DefaultMQProducer("please_input_group_name"); //生产组名称
producer.setNamesrvAddr("192.168.218.51"); //确定服务地址，集群时通过读取配置文件变量赋值
producer.start(); //生产者开始工作
//发送消息
for () {
//三个参数，第一个topic，第二个tag标识，第三个是消息内容
Message msg = new Message("test-topic","tag-a","msg");
SendResult sendResult = producer.send(msg); //生产者发送消息
}
producer.shutdown();
}
}
consumer
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_input_group_name"); //消费组名称
consumer.subscribe("topic","*"); //消费者订阅
consumer.registerMessageListener(new MessageListenerConcurrentliy) {
public ConsumeConcurrentStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
//处理的逻辑代码
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
};
consumer.start();
消息队列的协调者
NameServer的功能
NameServer是消息队列中的状态服务器，集群中的各个组件通过它了解全局的信息。同时，各个机器都要定期向NameServer上报自己的状态，如果超时不上报的话，其它组件会把这个机器从列表中移除。NameServer可以部署多个，本身是无状态的，也就是Broker、Topic等状态信息不会持久存储，是由各个角色上报存储到内存的。
集群状态的存储结构
private final HashMap> topicQueue topicQueueTable //这个map的key是Topic的名称，存储了所有topic的信息。Value存储着Broker的名称、读写Queue的数量以及同步标识等
private final HashMap //这个结构key是BrokerName，value存储着地址信息以及所属Cluster的名称
private final HashMap //这个结构的key是Broker的ip地址，value为Broker机器的实时状态，包括上次更新状态的时间戳
private final HashMap> //key为Cluster的名称，set中存储的是Broker的名称，就是集群的BrokerName的集合
private final HashMap> filterServerTable //key是Broker的地址，value是和这个Broker关联的多个过滤器的地址
以上五个变量的定义，可以清楚的看出各个组件的状态是如何进行存储的，而NameServer的作用便是维护这五个变量中存储的信息。
