当前位置:   article > 正文

RocketMQ的生产者原理和最佳实践_rocketmq 生产者组

rocketmq 生产者组

对于消息队列,生产者通常是入门第一个接触的对象,用于生产消息给消费者消费。本章通过介绍生产者实现类的属性、方法,引出生产者的启动过程、高可靠的实现方式等,主要讲解内容如下:

● RocketMQ支持3种消息:普通消息(并发消息)、顺序消息、事务消息。

● RocketMQ支持3种发送方式:同步发送、异步发送、单向发送。

● RocketMQ生产者最佳实践和总结。

2.1 生产者原理

通过第1章的讲解,相信读者对RocketMQ有了一个基本的认识,本节将对RocketMQ中的生产者做基本介绍。

2.1.1 生产者概述

发送消息的一方被称为生产者,它在整个RocketMQ的生产和消费体系中扮演的角色如图2-1所示。

生产者组:一个逻辑概念,在使用生产者实例的时候需要指定一个组名。一个生产者组可以生产多个Topic的消息。

生产者实例:一个生产者组部署了多个进程,每个进程都可以称为一个生产者实例。

Topic:主题名字,一个Topic由若干Queue组成。

RocketMQ 客 户 端 中 的 生 产 者 有 两 个 独 立 实现 类 :
org.apache.rocketmq.client.producer.DefaultMQProducer和org.apache.rocketmq.client.producer.TransactionMQProducer 。 前者用于生产普通消息、顺序消息、单向消息、批量消息、延迟消息,后者主要用于生产事务消息。

2.1.2 消息结构和消息类型

消息类的核心字段定义如下:

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)消息优雅地实现分布式事务。

2.1.3 生产者高可用

通常,我们希望不管Broker、Namesrv出现什么情况,发送消息都不要出现未知状态或者消息丢失。在消息发送的过程中,客户端、Broker、Namesrv 都有可能发生服务器损坏、掉电等各种故障。当这些故障发生时,RocketMQ是怎么处理的呢?

1.客户端保证

第一种保证机制:重试机制。RocketMQ 支持同步、异步发送,不管哪种方式都可以在配置失败后重试,如果单个 Broker 发生故障,重试会选择其他 Broker 保证消息正常发送。配置项 retryTimesWhenSendFailed表示同步重试次数,默认为 2次,加上正常发送 1次,总共3次机会。

同步发送的重试代码可以参考
org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(),每次发送失败后,除非发送被打断否则都会执行重试代码。同步发送重试代码如下:

重试是在通信层异步发送完成的,当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主从复制分为两种:同步复制和异步复制。同步复制是指消息发送到MasterBroker后,同步到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主从异步复制,这样效率极高。

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/盐析白兔/article/detail/603908
推荐阅读
相关标签
  

闽ICP备14008679号