RocketMQ源码学习
Apache RocketMQ是一个分布式消息传递和流媒体平台，具有低延迟，高性能和可靠性，万亿级容量和灵活的可伸缩性。 它具有多种功能：
- 发布/订阅消息传递模型
- 财务级交易消息
- 各种跨语言客户端，例如Java，C / C ++，Python，Go
- 可插拔的传输协议，例如TCP，SSL，AIO
- 内置的消息跟踪功能，还支持开放式跟踪
- 多功能的大数据和流生态系统集成
- 按时间或偏移量追溯消息
- 可靠的FIFO和严格的有序消息传递在同一队列中
- 高效的推拉消费模型
- 单个队列中的百万级消息累积容量
- 多种消息传递协议，例如JMS和OpenMessaging
- 灵活的分布式横向扩展部署架构
- 快如闪电的批量消息交换系统
- 各种消息过滤器机制，例如SQL和Tag
- 用于隔离测试和云隔离群集的Docker映像
- 功能丰富的管理仪表板，用于配置，指标和监视
- 认证与授权
RocketMQ源码主要分为以下几个package：
rocketmq-broker：mq的核心，它能接收producer和consumer的请求，并调用store层服务对消息进行处理。HA服务的基本单元，支持同步双写，异步双写等模式。
rocketmq-client：mq客户端实现，目前官方仅仅开源了java版本的mq客户端，c++，go客户端有社区开源贡献。
rocketmq-common：一些模块间通用的功能类，比如一些配置文件、常量。
rocketmq-filter：消息过滤服务，相当于在broker和consumer中间加入了一个filter代理。
rocketmq-namesrv：命名服务，更新和路由发现 broker服务。
rocketmq-remoting：基于netty的底层通信实现，所有服务间的交互都基于此模块。
rocketmq-store：存储层实现，同时包括了索引服务，高可用HA服务实现。
rocketmq-tools：mq集群管理工具，提供了消息查询等功能。
RocketMQ主要的功能集中在rocketmq-broker、rocketmq-remoting、rocketmq-store 三个模块中
Client生产者发送消息
1.org.apache.rocketmq.client.producer.DefaultMQProducer.send(org.apache.rocketmq.common.message.Message)
2.defaultMQProducerImpl.send(msg);
3.DefaultMQProducerImpl.sendKernelImpl()
4.DefaultMQProducerImpl.mQClientFactory.getMQClientAPIImpl().sendMessage()
5.MQClientAPIImpl.sendMessageSync()
6.RemotingClient.remotingClient.invokeSync(addr, request, timeoutMillis)
7.NettyRemotingAbstract->channel.writeAndFlush(request).addListener() 写入通道，等待Netty的Selector轮询出来
Broker(发送消息到达Broker)后续流程
1.消息到达Broker()
-NettyRemotingClient.NettyClientHandler.channelRead0
-processMessageReceived() processRequestCommand(ctx, cmd)
-NettyRequestProcessor【接口】.processRequest(ctx, cmd);
-SendMessageProcessor【具体实现类】.processRequest()
3.SendMessageProcessor.brokerController.getMessageStore().putMessage(msgInner);【非事务消息存储】
4.刷盘
consumer消息获取流程
DefaultMQPushConsumer:系统控制读取操作 DefaultMQPullConsumer:读取操作大部分功能自己控制
nameserver中5个变量RouteInfoManager
HashMap<String/* topic /, List> topicQueueTable;
HashMap<String/ brokerName /, BrokerData> brokerAddrTable;
HashMap<String/ clusterName /, Set<String/ brokerName />> clusterAddrTable;
HashMap<String/ brokerAddr /, BrokerLiveInfo> brokerLiveTable;
HashMap<String/ brokerAddr /, List/ Filter Server */> filterServerTable;