下载掌阅APP，畅读海量书库
立即打开
通过第1章的讲解，相信读者对RocketMQ有了一个基本的认识，本节将对RocketMQ中的生产者做基本介绍。
发送消息的一方被称为生产者，它在整个RocketMQ的生产和消费体系中扮演的角色如图2-1所示。
图2-1
生产者组： 一个逻辑概念，在使用生产者实例的时候需要指定一个组名。一个生产者组可以生产多个Topic的消息。
生产者实例： 一个生产者组部署了多个进程，每个进程都可以称为一个生产者实例。
Topic： 主题名字，一个Topic由若干Queue组成。
RocketMQ 客户端中的生产者有两个独立实现类：org.apache.rocketmq.client.producer.DefaultMQProducer和 org.apache.rocketmq.client.producer.TransactionMQProducer。前者用于生产普通消息、顺序消息、单向消息、批量消息、延迟消息，后者主要用于生产事务消息。
消息类的核心字段定义如下：
Topic： 主题名字，可以通过RocketMQ Console创建。
Flag： 目前没用。
Properties ：消息扩展信息，Tag、keys、延迟级别都保存在这里。
Body ：消息体，字节数组。需要注意生产者使用什么编码，消费者也必须使用相同编码解码，否则会产生乱码。
setKeys（）： 设置消息的key，多个key可以用MessageConst.KEY_SEPARATOR（空格）分隔或者直接用另一个重载方法。如果 Broker 中 messageIndexEnable=true 则会根据 key创建消息的Hash索引，帮助用户进行快速查询。
setTags（）： 消息过滤的标记，用户可以订阅某个Topic的某些Tag，这样Broker只会把订阅了topic-tag的消息发送给消费者。
setDelayTimeLevel（） ：设置延迟级别，延迟多久消费者可以消费。
putUserProperty（）： 如果还有其他扩展信息，可以存放在这里。内部是一个Map，重复调用会覆盖旧值。
RocketMQ支持普通消息、分区有序消息、全局有序消息、延迟消息和事务消息。
普通消息： 普通消息也称为并发消息，和传统的队列相比，并发消息没有顺序，但是生产消费都是并行进行的，单机性能可达十万级别的TPS。
分区有序消息 ：与Kafka中的分区类似，把一个Topic消息分为多个分区“保存”和消费，在一个分区内的消息就是传统的队列，遵循FIFO（先进先出）原则。
全局有序消息： 如果把一个 Topic 的分区数设置为 1，那么该 Topic 中的消息就是单分区，所有消息都遵循FIFO（先进先出）的原则。
延迟消息： 消息发送后，消费者要在一定时间后，或者指定某个时间点才可以消费。在没有延迟消息时，基本的做法是基于定时计划任务调度，定时发送消息。在 RocketMQ中只需要在发送消息时设置延迟级别即可实现。
事务消息： 主要涉及分布式事务，即需要保证在多个操作同时成功或者同时失败时，消费者才能消费消息。RocketMQ通过发送Half消息、处理本地事务、提交（Commit）消息或者回滚（Rollback）消息优雅地实现分布式事务。
通常，我们希望不管Broker、Namesrv出现什么情况，发送消息都不要出现未知状态或者消息丢失。在消息发送的过程中，客户端、Broker、Namesrv 都有可能发生服务器损坏、掉电等各种故障。当这些故障发生时，RocketMQ是怎么处理的呢？
1.客户端保证
第一种保证机制 ：重试机制。RocketMQ 支持同步、异步发送，不管哪种方式都可以在配置失败后重试，如果单个 Broker 发生故障，重试会选择其他 Broker 保证消息正常发送。
配置项 retryTimesWhenSendFailed表示同步重试次数，默认为 2次，加上正常发送 1次，总共3次机会。
同步发送的重试代码可以参考 org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl（），每次发送失败后，除非发送被打断否则都会执行重试代码。同步发送重试代码如下：
异步发送重试代码可以参考 org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessageAsync（），具体代码如下：
重试是在通信层异步发送完成的，当operationComplete（）方法返回的response值为null时，会重新执行重试代码。返回值 response 为 null 通常是因为客户端收到 TCP请求解包失败，或者没有找到匹配的request。
生产者配置项 retryTimesWhenSendAsyncFailed 表示异步重试的次数，默认为 2 次，加上正常发送的1次，总共有3次发送机会。
第二种保证机制 ：客户端容错。RocketMQ Client会维护一个“Broker-发送延迟”关系，根据这个关系选择一个发送延迟级别较低的 Broker 来发送消息，这样能最大限度地利用 Broker 的能力，剔除已经宕机、不可用或者发送延迟级别较高的 Broker，尽量保证消息的正常发送。
这种机制主要体现在发送消息时如何选择 Queue，源代码在 sendDefaultImpl（）方法调用的selectOneMessageQueue（）方法中，我们分两段来讲。
第一段代码如下：
sendLatencyFaultEnable ：发送延迟容错开关，默认为关闭，如果开关打开了，会触发发送延迟容错机制来选择发送Queue。
发送Queue时如何选择呢？
第一步：获取一个在延迟上可以接受，并且和上次发送相同的Broker。首先获取一个自增序号 index，通过取模获取Queue的位置下标 Pos。如果 Pos对应的 Broker的延迟时间是可以接受的，并且是第一次发送，或者和上次发送的Broker相同，则将Queue返回。
第二步：如果第一步没有选中一个Broker，则选择一个延迟较低的Broker。
第三步：如果第一、二步都没有选中一个Broker，则随机选择一个Broker。
第二段代码主要包括一个随机选择方法 tpInfo.selectOneMessageQueue （lastBrokerName），该方法的功能就是随机选择一个Broker，具体实现代码如下：
上面这段代码标注了三个步骤，分别解释如下：
第一步：如果没有上次使用的Broker作为参考，那么随机选择一个Broker。
第二步：如果存在上次使用的Broker，就选择非上次使用的Broker，目的是均匀地分散Broker的压力。
第三步：如果第一、二步都没有选中一个Broker，则采用兜底方案——随机选择一个Broker。
在执行如上两段代码时，需要 Broker 和发送延迟的数据作为判断的依据，这些数据是怎么来的呢？
客户端在发送消息后，会调用 updateFaultItem（）方法来更新当前接收消息的 Broker的延迟情况，这些主要逻辑都在 MQFaultStrategy类中实现，延迟策略有一个标准接口LatencyFaultTolerance，如果读者想要自己实现一种延迟策略，可以通过这个接口来实现。
2.Broker端保证
数据同步方式保证：在后面 Broker章节中会讲到 Broker主从复制分为两种：同步复制和异步复制。同步复制是指消息发送到Master Broker后，同步到Slave Broker才算发送成功；异步复制是指消息发送到Master Broker，即为发送成功。在生产环境中，建议至少部署2个Master和2个Slave，下面分为几种情况详细描述。
（1）1个Slave掉电。Broker同步复制时，生产第一次发送失败，重试到另一组Broker后成功；Broker异步复制时，生产正常不受影响。
（2）2个 Slave掉电。Broker同步复制时，生产失败；Broker异步复制时，生产正常不受影响。
（3）1 个 Master 掉电。Broker 同步复制时，生产第一次失败，重试到另一组 Broker后成功；Broker异步复制时的做法与同步复制相同。
（4）2个Master掉电。全部生产失败。
（5）同一组Master和Slave掉电。Broker同步复制时，生产第一次发送失败，重试到另一组Broker后成功；Broker异步复制时，生产正常不受影响。
（6）2组机器都掉电：全部生产失败。
综上所述，想要做到绝对的高可靠，将 Broker 配置的主从同步进行复制即可，只要生产者收到消息保存成功的反馈，消息就肯定不会丢失。一般适用于金融领域的特殊场景。绝大部分场景都可以配置Broker主从异步复制，这样效率极高。
DefaultMQProducer是RocketMQ中默认的生产者实现，DefaultMQProducer的类之间的继承关系如图2-2所示，可以看到这个生产者在实现时包含生产者的操作和配置属性，这是典型的类对象设计。下面我们将介绍类对象的一些核心属性和方法。
以下是一些核心属性：
namesrvAddr： 继承自 ClientConfig，表示 RocketMQ 集群的 Namesrv 地址，如果是多个则用分号分开。比如：127.0.0.1：9876；127.0.0.2：9876。
clientIP： 使用的客户端程序所在机器的 IP地址。支持 IPv4和 IPv6，IPv4 排除了本地的环回地址（127.0.xxx.xxx）和私有内网地址（192.168.xxx.xxx）。这里需要注意的是，如果 Client 运行在 Docker 容器中，获取的 IP 地址是容器所在的 IP 地址，而非宿主机的IP地址。
图2-2
instanceName： 实例名，每个实例都需要取唯一的名字，因为有时我们会在同一个机器上部署多个程序进程，如果名字有重复就会导致启动失败。
vipChannelEnabled： 这是一个 boolean 值，表示是否开启 VIP 通道。VIP 通道和非VIP通道的区别是：在通信过程中使用的端口号不同。
clientCallbackExecutorThreads： 客户端回调线程数。该参数表示 Netty 通信层回调线程的个数，默认值Runtime.getRuntime（）.availableProcessors（）表示当前CPU的有效个数。
pollNameServerInterval： 获取 Topic 路由信息的间隔时长，单位为 ms，默认为30 000ms。
heartbeatBrokerInterval： 与Broker心跳间隔的时长，单位为ms，默认为30 000ms。
defaultMQProducerImpl： 默认生产者的实现类，其中封装了Broker的各种API（启动及关闭生产者的接口）。如果你想自己实现一个生产者，可以添加一个新的实现，保持DefaultMQProducer对外接口不变，用户完全没有感知。
producerGroup： 生产者组名，这是一个必须传递的参数。RocketMQ-way表示同一个生产者组中的生产者实例行为需要一致。
sendMsgTimeout： 发送超时时间，单位为ms。
compressMsgBodyOverHowmuch： 消息体的容量上限，超过该上限时消息体会通过ZIP进行压缩，该值默认为4MB。该参数在Client中是如何生效的呢？具体实现代码如下：
retryTimesWhenSendFailed： 同步发送失败后重试的次数。默认为2次，也就是说，一共有3次发送机会。
retryTimesWhenSendAsyncFailed： 异步发送失败后重试的次数。默认为 2次。异步重试是有条件的重试，并不是每次发送失败后都重试。源代码可以查看org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessageAsync（）方法。每次发送失败抛出异常后，通过执行onExceptionImpl（）方法来决定什么场景进行重试。
以下是一些核心方法：
start（）： 这是启动整个生产者实例的入口，主要负责校验生产者的配置参数是否正确，并启动通信通道、各种定时计划任务、Pull服务、Rebalance服务、注册生产者到Broker等操作。
shutdown（）： 关闭本地已注册的生产者，关闭已注册到Broker的客户端。
fetchPublishMessageQueues（Topic）： 获取一个Topic有哪些Queue。在发送消息、Pull消息时都需要调用。
send（Message msg）： 同步发送普通消息。
send（Message msg，long timeout）： 同步发送普通消息（超时设置）。
send（Message msg，SendCallback sendCallback）： 异步发送普通消息。
send（Message msg，SendCallback sendCallback，long timeout）： 异步发送普通消息（超时设置）。
sendOneway（Message msg）： 发送单向消息。只负责发送消息，不管发送结果。
send（Message msg，MessageQueue mq）： 同步向指定队列发送消息。
send（Message msg，MessageQueue mq，long timeout）： 同步向指定队列发送消息（超时设置）。
同步向指定队列发送消息时，如果只有一个发送线程，在发送到某个指定队列中时，这个指定队列中的消息是有顺序的，那么就按照发送时间排序；如果某个Topic的队列都是这种情况，那么我们称该Topic的全部消息是分区有序的。
send（Message msg，MessageQueue mq，SendCallback sendCallback）： 异步发送消息到指定队列。
send（Message msg，MessageQueue mq，SendCallback sendCallback，long timeout）： 异步发送消息到指定队列（超时设置）。
send（Message msg，MessageQueueSelector selector，Object arg，SendCallback sendCallback）： 自定义消息发送到指定队列。通过实现MessageQueueSelector接口来选择将消息发送到哪个队列。
send（Collection＜Message＞msgs）： 批量发送消息。
下面介绍两个核心管理接口：
createTopic（String key，String newTopic，int queueNum）： 创建Topic。
viewMessage（String offsetMsgId）： 根据消息id查询消息内容。
生产者启动的流程比消费者启动的流程更加简单，一般用户使用 DefaultMQProducer的构造函数构造一个生产者实例，并设置各种参数。比如Namesrv地址、生产者组名等，调用start（）方法启动生产者实例，start（）方法调用了生产者默认实现类的start（）方法启动，这里我们主要讲实现类的start（）方法内部是怎么实现的，其流程如图2-3所示。
图2-3
第一步：通过 switch-case 判断当前生产者的服务状态，创建时默认状态是CREATE_JUST。设置默认启动状态为启动失败。
第二步：执行checkConfig（）方法。校验生产者实例设置的各种参数。比如生产者组名是否为空、是否满足命名规则、长度是否满足等。
第三步：执行changeInstanceNameToPID（）方法。校验instance name，如果是默认名字则将其修改为进程id。
第四步：执行getAndCreateMQClientInstance（）方法。根据生产者组名获取或者初始化一个MQClientInstance。初始化代码如下：
由此可见，MQClientInstance实例与clientId是一一对应的，而clientId是由clientIP、instanceName 及 unitName 构成的。一般来讲，为了减少客户端的使用资源，如果将所有的 instanceName和 unitName设置为同样的值，就会只创建一个 MQClientInstance实例，具体实现代码如下：
MQClientInstance 实例的功能是管理本实例中全部生产者与消费者的生产和消费行为。下面我们来看一下 org.apache.rocketmq.client.impl.factory.MQClientInstance 类的核心属性，具体代码（篇幅原因，删去了初始化代码）如下：
下面给大家解读这段代码：
producerTable： 当前client实例的全部生产者的内部实例。
consumerTable ：当前client实例的全部消费者的内部实例。
adminExtTable： 当前client实例的全部管理实例。
mQClientAPIImpl： 其实每个client也是一个Netty Server，也会支持Broker访问，这里实现了全部client支持的接口。
mQAdminImpl： 管理接口的本地实现类。
topicRouteTable： 当前生产者、消费者中全部Topic的本地缓存路由信息。
scheduledExecutorService： 本地定时任务，比如定期获取当前 Namesrv 地址、定期同步Namesrv信息、定期更新Topic路由信息、定期发送心跳信息给Broker、定期清理已下线的Broker、定期持久化消费位点、定期调整消费线程数（这部分源代码被官方删除了）。
clientRemotingProcessor： 请求的处理器，从处理方法processRequest（）中我们可以知道目前支持哪些功能接口。
pullMessageService： Pull服务。
这里为什么会启动用于消费的Pull服务呢？这是一个兼容写法。通过查看源代码运行过程，读者就会发现Pull服务是由一个状态变量方法this.isStopped（）控制的，这个stopped状态变量默认是False，而pullRequestQueue也是空的，所以这里只是启动了pullMessageService，并没有真正地执行Pull操作，相关代码如下：
rebalanceService ：重新平衡服务。定期执行重新平衡方法 this.mqClientFactory.doRebalance（）。这里的 mqClientFactory 就是 MQClientInstance 实例，通过依次调用MQClientInstance中保存的消费者实例的doRebalance（）方法，来感知订阅关系的变化、集群变化等，以达到重新平衡。
consumerStatsManager ：消费监控。比如拉取RT（Response Time，响应时间）、拉取TPS（Transactions Per Second，每秒处理消息数）、消费RT等都可以统计。
MQClientInstance中还有一些核心方法如下：
下面对这些方法逐一进行讲解：
updateTopicRouteInfoFromNameServer： 从多个Namesrv中获取最新Topic路由信息，更新本地缓存。
cleanOfflineBroker： 清理已经下线的Broker。
checkClientInBroker： 检查Client是否在Broker中有效。
sendHeartbeatToAllBrokerWithLock： 发送客户端的心跳信息给所有的Broker。
registerConsumer： 在本地注册一个消费者。
unregisterConsumer： 取消本地注册的消费者。
registerProducer： 在本地注册一个生产者。
unregisterProducer： 取消本地注册的生产者。
registerAdminExt： 注册一个管理实例。
rebalanceImmediately ：立即执行一次 Rebalance。该操作是通过 RocketMQ 的一个CountDownLatch2锁来实现的。
doRebalance ：对于所有已经注册的消费者实例，执行一次Rebalance。
findBrokerAddressInAdmin ：在本地缓存中查找Master或者Slave Broker信息。
findBrokerAddressInSubscribe： 在本地缓存中查找Slave Broker信息。
findBrokerAddressInPublish ：在本地缓存中查找Master Broker地址。
findConsumerIdList： 查找消费者id列表。
findBrokerAddrByTopic ：通过Topic名字查找Broker地址。
resetOffset： 重置消费位点。
getConsumerStatus： 获取一个订阅关系中每个队列的消费进度。
getTopicRouteTable： 获取本地缓存Topic路由。
consumeMessageDirectly： 直接将消息发送给指定的消费者消费，和正常投递不同的是，指定了已经订阅的消费者组中的一个，而不是全部已经订阅的消费者。一般适用于在消费消息后，某一个消费者组想再消费一次的场景。
consumerRunningInfo： 获取消费者的消费统计信息。包含消费RT、消费TPS等。
RocketMQ客户端的消息发送通常分为以下3层：
业务层 ：通常指直接调用RocketMQ Client发送API的业务代码。
消息处理层 ：指RocketMQ Client获取业务发送的消息对象后，一系列的参数检查、消息发送准备、参数包装等操作。
通信层 ：指RocketMQ基于Netty封装的一个RPC通信服务，RocketMQ的各个组件之间的通信全部使用该通信层。
总体上讲，消息发送流程首先是 RocketMQ 客户端接收业务层消息，然后通过DefaultMQProducerImpl发送一个RPC请求给Broker，再由Broker处理请求并保存消息。下面以DefaultMQProducer.send（Message msg）接口为例讲解发送流程，如图2-4所示。
消息发送流程具体分为3步：
第一步：调用defaultMQProducerImpl.send（）方法发送消息。
第二步：通过设置的发送超时时间，调用defaultMQProducerImpl.send（）方法发送消息。设置的超时时间可以通过sendMsgTimeout进行变更，其默认值为3s。
第三步：执行defaultMQProducerImpl.sendDefaultImpl（）方法。这是一个公共发送方法，我们先看看入参：
图2-4
communicationMode：通信模式，同步、异步还是单向。
sendCallback：对于异步模式，需要设置发送完成后的回调。
该方法是发送消息的核心方法，执行过程分为5步：
第一步，两个检查：生产者状态、消息及消息内容。没有运行的生产者不能发送消息。消息检查主要检查消息是否为空，消息的Topic的名字是否为空或者是否符合规范；消息体大小是否符合要求，最大值为4MB，可以通过maxMessageSize进行设置。
第二步，执行tryToFindTopicPublishInfo（）方法：获取Topic路由信息，如果不存在则发出异常提醒用户。如果本地缓存没有路由信息，就通过Namesrv获取路由信息，更新到本地，再返回。具体实现代码如下：
第三步，计算消息发送的重试次数，同步重试和异步重试的执行方式是不同的。
第四步，执行队列选择方法selectOneMessageQueue（）。根据队列对象中保存的上次发送消息的Broker的名字和Topic路由，选择（轮询）一个Queue将消息发送到Broker。我们可以通过 sendLatencyFaultEnable 来设置是否总是发送到延迟级别较低的 Broker，默认值为False。
第五步，执行sendKernelImpl（）方法。该方法是发送消息的核心方法，主要用于准备通信层的入参（比如Broker地址、请求体等），将请求传递给通信层，内部实现是基于Netty的，在封装为通信层request对象RemotingCommand前，会设置RequestCode表示当前请求是发送单个消息还是批量消息。具体实现代码如下：
Netty 本身是一个异步的网络通信框架，怎么实现同步的调用呢？我们可以通过org.apache.rocketmq.remoting.netty.NettyRemotingAbstract.invokeSyncImpl（）方法来实现同步的调用，具体实现代码如下：
在每次发送同步请求后，程序会执行 waitResponse（）方法，直到 Netty接收 Broker的返回结果，相关代码如下：
然后，通过putResponse（）方法释放锁，让请求线程同步返回。
异步发送时有很多request，每个response返回后怎么与request进行对应呢？这里面有一个关键参数——opaque，RocketMQ每次发送同步请求前都会为一个request分配一个opaque，这是一个原子自增的id，一个response会以opaque作为key保存在responseTable中，这样用opaque就将request和response连接起来了。
无论请求发送成功与否，都执行 updateFaultItem（）方法，这就是在第三步中讲的总是发送到延迟级别较低的Broker的逻辑。