RocketMQ - 生产者原理

时间:2023-02-16 10:08:45

https://rocketmq.apache.org/

Apache RocketMQ是一款开源的、分布式的消息投递与流数据平台。出生自阿里巴巴,在阿里巴巴内部经历了3个版本后,作为Apache
*开源项目之一直到现在。在GitHub上有10000+star、5000+fork、170+contributors(在GitHub上提交代码并被采纳的开发者)

RocketMQ的前世

和大部分组件产生的原因类似,阿里巴巴内部为了适应淘宝 B2C 的更快、更复杂的业务,2001年启动了“五彩石项目”,阿里巴巴的第一代消息队列服务Notify就是在这个背景下产生的。
2010 年,阿里巴巴内部的 Apache ActiveMQ 仍然作为核心技术被广泛用于各个业务线,而顺序消息、海量消息堆积、完全自主控制消息队列服务,也是阿里巴巴同时期急需的。在这种背景下,2011年,MetaQ 诞生

RocketMQ 云化

2011年,LinkedIn将Kafka开源。2012年,阿里巴巴参考Kafka的设计,基于对MetaQ的理解和实际使用,研发了一套通用消息队列引擎,也就是 RocketMQ。自此才有了第一代真正的RocketMQ,2016年阿里云上线云RocketMQ消息队列服务。

Apache RocketMQ——金融级消息队列,一个拥有亚毫秒级延迟、万亿级消息容量保证、高消息容错设计的中间件,在阿里巴巴、VIPKID、微众银行、民生银行、蚂蚁金服、滴滴等国内知名互联网公司的实践中,有着完美的表现。
随着RocketMQ 5.0的发布,借助OpenMessaging提供跨平台、多语言的能力,将会打通 Prometheus、ELK 等上游组件,通过消息、Streaming 等形式将数据扭转到 Flink、Elasticsearch、Hbase、Spark等下游组件。届时整个生态体系将会更加完美、便捷。

RocketMQ支持3种消息:普通消息(并发消息)、顺序消息、事务消息
RocketMQ支持3种发送方式:同步发送、异步发送、单向发送。

生产者概述

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

生产者组: 一个逻辑概念,在使用生产者实例的时候需要指定一个组名。一个生产者组可以生产多个Topic的消息。
生产者实例: 一个生产者组部署了多个进程,每个进程都可以称为一个生产者实例。
Topic: 主题名字,一个Topic由若干Queue组成。

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

消息结构和消息类型

消息类的核心字段

public class Message implements Serializable {
	private static final long serialVersionUID = 8445773977080406428L;
	// 主题名字,可以通过RocketMQ Console创建
	private String topic;
	// 目前没用
	private int flag;
	// 消息扩展信息,Tag、keys、延迟级别都保存在这里
	private Map<String, String> properties;
	// 消息体,字节数组。需要注意生产者使用什么编码,消费者也必须使用相同编码解码,否则会产生乱码
	private bytel[] body;
	// 设置消息的 key,多个 key 可 以 用MessageConst.KEY_SEPARATOR(空格)分隔或者直接用另一个重载方法。如果 Broker 中 messageIndexEnable=true 则会根据 key创建消息的Hash索引,帮助用户进行快速查询。
	public void setKeys(String keys) {}	
	public void setKeys(Collection<String> keys){}
	// 消息过滤的标记,用户可以订阅某个Topic的某些 Tag,这样Broker只会把订阅了topic-tag的消息发送给消费者。
	public void setTags(String tags) {}
	// 设置延迟级别,延迟多久消费者可以消费
	public void setDelayTimeLevel(int level) { )
	public void setTopic(String topic) { }
	// 如果还有其他扩展信息,可以存放在这里。内部是一个Map,重复调用会覆盖旧值。
	public void putUserProperty(final String name, final String value) {...}
}

普通消息

普通消息也称为并发消息,和传统的队列相比,并发消息没有顺序,但是生产消费都是并行进行的,单机性能可达十万级别的TPS

分区有序消息

与Kafka中的分区类似,把一个Topic消息分为多个分区“保存”和消费,在一个分区内的消息就是传统的队列,遵循FIFO(先进先出)原则。

全局有序消息

如果把一个 Topic 的分区数设置为 1,那么该Topic 中的消息就是单分区,所有消息都遵循FIFO(先进先出)的原则。

延迟消息

消息发送后,消费者要在一定时间后,或者指定某个时间点才可以消费。在没有延迟消息时,基本的做法是基于定时计划任务调度,定时发送消息。在 RocketMQ中只需要在发送消息时设置延迟级别即可实现

事务消息

主要涉及分布式事务,即需要保证在多个操作同时成功或者同时失败时,消费者才能消费消息。RocketMQ通过发送Half消息、处理本地事务、提交(Commit)消息或者回滚(Rollback)消息优雅地实现分布式事务。

生产者高可用

客户端保证

第一种保证机制:重试机制

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,源代码在 MQFaultStrategy.selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName)方法中

public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
    if (this.sendLatencyFaultEnable) {
        try {
            //第一步:获取一个在延迟上可以接受,并且和上次发送相同的Broker。首先获取一个自增序号 index,通过取模获取Queue的位置下标 Pos。如果 Pos对应的 Broker的延迟时间是可以接受的,并且是第一次发送,或者和上次发送的Broker相同,则将Queue返回。
            int index = tpInfo.getSendWhichQueue().incrementAndGet();
            for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                if (pos < 0)
                    pos = 0;
                MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                if (latencyFaultTolerance.isAvailable(mq.getBrokerName()))
                    return mq;
            }
            //第二步:如果第一步没有选中一个Broker,则选择一个延迟较低的Broker。
            final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
            int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
            if (writeQueueNums > 0) {
                final MessageQueue mq = tpInfo.selectOneMessageQueue();
                if (notBestBroker != null) {
                    mq.setBrokerName(notBestBroker);
                    mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums);
                }
                return mq;
            } else {
                latencyFaultTolerance.remove(notBestBroker);
            }
        } catch (Exception e) {
            log.error("Error occurred when selecting message queue", e);
        }

        return tpInfo.selectOneMessageQueue();
    }
    //第三步:如果第一、二步都没有选中一个Broker,则随机选择一个Broker
    return tpInfo.selectOneMessageQueue(lastBrokerName);
}

tpInfo.selectOneMessageQueue(lastBrokerName) 该方法的功能就是随机选择一个Broker

public MessageQueue selectOneMessageQueue(final String lastBrokerName)
{
	//第一步 如果没有上次使用的Broker作为参考,那么随机选择一个Broker。
	if (lastBrokerName == null) {
		return selectOneMessageQueue ();
	} else {//第二步 如果存在上次使用的Broker,就选择非上次使用的 Broker,目的是均匀地分散Broker的压力
		int index = this.sendwhichQueue.getAndIncrement();
		for (int i = 0;i < this.messageQueueList.size(); i++){
			int pos = Math.abs(index++)  this.messageQueueList.size();
			if (pos <0)
				pos =0;
			MessageQueue mg = this.messageQueueList.get(pos);
			if (!mq.getBrokerName().equals(lastBrokerName)) {
				return mq;
			}
		}
		//第三步 如果第一、二步都没有选中一个Broker,则采用兜底方案——随机选择一个Broker
		return selectOneMessageQueue();
	}
}

Broker 端保证

数据同步方式保证:在后面 Broker章节中会讲到 Broker主从复制分为两种:同步复制和异步复制。同步复制是指消息发送到MasterBroker后,同步到Slave Broker才算发送成功;异步复制是指消息发送到Master Broker,即为发送成功。在生产环境中,建议至少部署2个Master和2个Slave,下面分为几种情况详细描述。

  • 1个Slave掉电。Broker同步复制时,生产第一次发送失败,重试到另一组Broker后成功;Broker异步复制时,生产正常不受影响。
  • 2个Slave掉电。Broker同步复制时,生产失败;Broker异步复制时,生产正常不受影响。
  • 1个Master掉电。Broker 同步复制时,生产第一次失败,重试到另一组 Broker后成功;Broker异步复制时的做法与同步复制相同。
  • 2个Master掉电。全部生产失败。
  • 同一组Master和Slave掉电。Broker同步复制时,生产第一次发送失败,重试到另一组Broker后成功;Broker异步复制时,生产正常不受影响。
  • 2组机器都掉电:全部生产失败。

综上所述,想要做到绝对的高可靠,将 Broker 配置的主从同步进行复制即可,只要生产者收到消息保存成功的反馈,消息就肯定不会丢失。一般适用于金融领域的特殊场景。绝大部分场景都可以配置Broker主从异步复制,这样效率极高