标签：集群部署 发送消息 pac 应用 数据源 sock 查询 故障 更改
Apache RocketMQ是一个分布式消息传递和流平台，具有低延迟，高性能和可靠性，万亿级容量和灵活的可扩展性。 它由四部分组成：NamerServer，Broker，Produer和Customer。 它们中的每一个都可以水平扩展而没有单一的故障点。 如下面的截图所示。
NameServers提供轻量级服务发现和路由。 每个NameServer记录完整的路由信息，提供相应的读写服务，并支持快速存储扩展。
NameServer是一个几乎无状态的节点，可集群部署，节点之间无任何信息同步。
将NameServer地址列表提供给客户端有四种方法：
Brokers通过提供轻量级的TOPIC和QUEUE机制来处理消息存储。 它们支持Push和Pull模型，包含容错机制（2个副本或3个副本），并提供强大的峰值填充和按原始时间顺序累积数千亿条消息的能力。 此外，Brokers还提供灾难恢复，丰富的指标统计和警报机制，这些都是传统的消息传递系统所缺少的。
Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slaver，但是一个Slaver只能对应一个Master，Master与Slaver的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slaver。Master可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有的NameServer。
如下图所示，Broker有几个重要的子模块组成：
Customers也支持Push和Pull两种模式的分布式部署。 它还支持群集消费和消息广播。 它提供实时消息订阅机制，可以满足大多数消费者的需求。
Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slaver建立长连接，且定时向Master、Slaver发送心跳。Consumer即可从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。
1.一个应用尽可能用一个 Topic，消息子类型用 tags 来标识，tags 可以由应用自由设置。只有发送消息设置了tags，消费方在订阅消息时，才可以利用 tags 在 broker 做消息过滤。
2.每个消息在业务层面的唯一标识码，要设置到 keys 字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引（哈希索引），应用可以通过 topic，key 来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。
3.消息发送成功或者失败，要打印消息日志，务必要打印 sendresult 和 key 字段。
4.send 消息方法，只要不抛异常，就代表发送成功。但是发送成功会有多个状态：
5.对于消息不可丢失应用，务必要有消息重发机制
Producer 的 send 方法本身支持内部重试，重试逻辑如下：
如果调用 send 同步方法发送失败，则尝试将消息存储到 db，由后台线程定时重试，保证消息一定到达 Broker。
一个 RPC 调用，通常是这样一个过程
所以一个 RPC 的耗时时间是上述三个步骤的总和，而某些场景要求耗时非常短，但是对可靠性要求并不高，例如日志收集类应用，此类应用可以采用 oneway 形式调用，oneway 形式只发送请求不等待应答，而发送请求在客户端实现层面仅仅是一个 os 系统调用的开销，即将数据写入客户端的 socket 缓冲区，此过程耗时通常在微秒级。RocketMQ不止可以直接推送消息，在消费端注册监听器进行监听，还可以由消费端决定自己去拉取数据。
RocketMQ无法做到消息重复，所以如果业务对消息重复非常敏感，务必要在业务层面去重。将消息的唯一键，可以是MsgId，也可以是消息内容中的唯一标识字段，例如订单ID，消费之前判断是否在DB或Tair（全局KV存储）中存在，如果不存在则插入，并消费，否则跳过。（实践过程要考虑原子性问题，判断是否存在可以尝试插入，如果报主键冲突，则插入失败，直接跳过） msgid一定是全局唯一的标识符，但是可能会存在同样的消息有两个不同的msgid的情况（有多种原因），这种情况可能会使业务上重复，建议最好使用消息体中的唯一标识字段去重。
如果业务流程支持批量方式消费，则可以很大程度上的提高吞吐量，可以通过设置Consumer的consumerMessageBatchMaxSize参数，默认是1，即一次消费一条。
发生消息堆积时，如果消费速度一直跟不上发送速度，可以选择丢弃不重要的消息。例如当前offset和maxOffset差值过大时（可能时因为消息系统堆积），直接把当前消息消费成功，可以快速使消息的消费和发出达到平衡。
根据实际业务需要，尽可能的优化代码，减少DB访问数量，进而减少RT，提高消息的消费速度。
RocketMQ通过轮询所有队列的方式来确定消息被发送到哪一个队列（负载均衡策略）。比如下面的示例中，订单号相同的消息会被先后发送到同一个队列中：
// RocketMQ通过MessageQueueSelector中实现的算法来确定消息发送到哪一个队列上
// RocketMQ默认提供了两种MessageQueueSelector实现：随机/Hash
// 当然你可以根据业务实现自己的MessageQueueSelector来决定消息按照何种策略发送到消息队列中
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);
RocketMQ除了支持普通消息，顺序消息，另外还支持事务消息。
A（存在DB操作）、B（存在DB操作）两方需要保证分布式事务一致性，通过引入中间层MQ，A和MQ保持事务一致性（异常情况下通过MQ反查A接口实现check），B和MQ保证事务一致（通过重试），从而达到最终事务一致性。
上面以DB为例，其实此处可以是任何业务或者数据源。
TransactionCheckListener 是在消息的commit或者rollback消息丢失的情况下才会回调（上图中灰色部分）。这种消息丢失只存在于断网或者rocketmq集群挂了的情况下。当rocketmq集群挂了，如果采用异步刷盘，存在1s内数据丢失风险，异步刷盘场景下保障事务没有意义。所以如果要核心业务用Rocketmq解决分布式事务问题，建议选择同步刷盘模式。
当需要保证多方（超过2方）的分布式一致性，上面的两方事务一致性（通过Rocketmq的事务性消息解决）已经无法支持。这个时候需要引入TCC模式思想。
以上图交易系统为例：
1）交易系统创建订单（往DB插入一条记录），同时发送订单创建消息。通过RocketMq事务性消息保证一致性
2）接着执行完成订单所需的同步核心RPC服务（非核心的系统通过监听MQ消息自行处理，处理结果不会影响交易状态）。执行成功更改订单状态，同时发送MQ消息。
3）交易系统接受自己发送的订单创建消息，通过定时调度系统创建延时回滚任务（或者使用RocketMq的重试功能，设置第二次发送时间为定时任务的延迟创建时间。在非消息堵塞的情况下，消息第一次到达延迟为1ms左右，这时可能RPC还未执行完，订单状态还未设置为完成，第二次消费时间可以指定）。延迟任务先通过查询订单状态判断订单是否完成，完成则不创建回滚任务，否则创建。 PS：多个RPC可以创建一个回滚任务，通过一个消费组接受一次消息就可以；也可以通过创建多个消费组，一个消息消费多次，每次消费创建一个RPC的回滚任务。 回滚任务失败，通过MQ的重发来重试。
以上是交易系统和其他系统之间保持最终一致性的解决方案。
标签：集群部署 发送消息 pac 应用 数据源 sock 查询 故障 更改
原文地址：https://www.cnblogs.com/helloxc/p/10549842.html
