RabbitMQ入门

参考

https://blog.csdn.net/weixin_45985053/article/details/126531765
https://blog.csdn.net/weixin_44183847/article/details/129389380

事务消息参考：
事务最终一致性
https://blog.csdn.net/m0_49093968/article/details/131854674
https://blog.csdn.net/Little_fxc/article/details/124045665

优势

应用解耦:提高系统的容错性和可维护性，消息的生产者发送消息，消费者消费消息，生产者发送消息后就去做其他事情；
异步提速:用户体验和系统吞吐量，生产者发送消息后就返回前端通知用户，后续的消费消息是异步的；
削峰填谷:使用mq，限制消费消息的速度，高峰期消息都被挤压在MQ中（削峰）不会压垮数据库，高峰过后，还是按照原来的速度消费消息（填谷），将待消费的消息消费完。可以调高系统的稳定性；

劣势：

1.系统的可用性降低，引入外部依赖，系统的稳定性降低了，如何保证高可用！
2.系统的复杂度提高了，以前是同步远程调用，现在是异步调用，如何保证消息不被重复消费，怎样处理消息丢失，保证消息传递的顺序性；
3.一致性问题，A系统处理业务，MQ转发给B、C、D而其中D出现异常，产生了数据的一致性问题；
什么时候用MQ
① 生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空，这才让明 明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。② 容许短暂的不一致性.③ 确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本、
#2、AMQP 和 JMS
MQ是消息通信的模型；实现MQ的大致有两种主流方式：AMQP、JMS。
AMQP:AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。JMS:JMS即Java消息服务（JavaMessage Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

AMQP 与 JMS 区别

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式,JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模式；而AMQP的消息模式更加丰富。

NON_PERSISTENT模式和 PERSISTENT模式

对于非持久的消息，JMS provider不会将它存到文件/数据库等稳定的存储介质中。也就是说非持久消息驻留在内存中，如果jms provider宕机，那么内存中的非持久消息会丢失。A JMS provider must diliver a NON_PERSISTENT message at-most-once 。对于持久消息，消息提供者会使用存储-转发机制，先将消息存储到稳定介质中，等消息发送成功后再删除。如果jms provider挂掉了，那么这些未送达的消息不会丢失；jms provider恢复正常后，会重新读取这些消息，并传送给对应的消费者。 A JMS provider must diliver a PERSISTENT message once -and-only-once

MQ产品介绍

ActiveMQ:基于JMS,java语言实现，万级数据吞吐量，处理速度ms级，主从架构，成熟度高
ZeroMQ：基于C语言开发
RabbitMQ: erlang语言实现，万级数据吞吐量，处理速度us级，主从架构，
RocketMQ: 基于AMQP协议，erlang语言开发，稳定性好,java语言实现，十万级数据吞吐量，处理速度ms级，分布式架构，功能强大，扩展性强
kafka: scala语言实现，十万级数据吞吐量，处理速度ms级，分布式架构，功能较少，应用于大数据较多

RocketMQ
RocketMQ是阿里开源的一款非常优秀中间件产品，脱胎于阿里的另一款队列技术MetaQ，后捐赠给Apache基金会 作为一款孵化技术，仅仅经历了一年多的时间就成为Apache基金会的顶级项目。并且它现在已经在阿里内部被广泛 的应用，并且经受住了多次双十一的这种极致场景的压力（2017年的双十一，RocketMQ流转的消息量达到了万亿 级，峰值TPS达到5600万）

原理

producer向命名服务器集群发消息获取broker的ip，然后向broker集群发消息，consumer向命名服务器集群发消息获取ip,与broker建立长连接，
消息发送
One-To-One（基础发送与基础接收）
One-To-Many（负载均衡模式与广播模式）

BROADCASTING("BROADCASTING"),
CLUSTERING("CLUSTERING");
consumer.setMessageModel(MessageModel.BROADCASTING);
1
2
3

Many-To-Many
单生产者多消费者消息发送（OneToMany）
消费者（广播模式,集群模式）
多生产者多消费者消息发送（ManyToMany）
多生产者产生的消息可以被同一个消费者消费，也可以被多个消费者消费

消息类别

1.同步消息
即时性较强，重要的消息，且必须有回执的消息，例如短信，通知（转账成功）；

producer.send(msg);
1

2.异步消息
即时性较弱，但需要有回执的消息，例如订单中的某些信息

producer.send(msg, new SendCallback() {});
1

3.单向消息
不需要有回执的消息，例如日志类消息

producer.sendOneway(msg);
1

4.延时消息
消息发送时并不直接发送到消息服务器，而是根据设定的等待时间到达，起到延时到达的缓冲作用

Message msg = new Message("topic3",("延时消息：hello rocketmq "+i).getBytes("UTF-8"));
//设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
// private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
msg.setDelayTimeLevel(3);
SendResult result = producer.send(msg);
1
2
3
4
5

5.批量消息
批量发送消息能显著提高传递小消息的性能.
批量消息的限制条件：
1这些批量消息应该有相同的topic
2相同的waitStoreMsgOK
3不能是延时消息
4消息内容总长度不超过4M
​ 消息内容总长度包含如下：
​ topic（字符串字节数）
​ body （字节数组长度）
​ 消息追加的属性（key与value对应字符串字节数）
​ 日志（固定20字节）

SendResult result = producer.send(msgList);
1

消息过滤

1.按照tag过滤信息;

//发送消息
Message msg = new Message("topic6","tag2",("消息过滤按照tag：hello rocketmq 2").getBytes("UTF-8"));
//接收消息，除了制定topic，还可以指定接收的tag,*代表任意tag
consumer.subscribe("topic6","tag1 || tag2");
1
2
3
4

2.语法过滤（属性过滤/SQL过滤）
基本语法
数值比较，比如：>，>=，<，<=，BETWEEN，=；
字符比较，比如：=，<>，IN；
IS NULL 或者 IS NOT NULL；
逻辑符号 AND，OR，NOT；
常量支持类型为：
数值，比如：123，3.1415；
字符，比如：‘abc’，必须用单引号包裹起来；
NULL，特殊的常量
布尔值，TRUE 或 FALSE

//生产者，为消息添加属性
msg.putUserProperty("vip","1");
msg.putUserProperty("age","20");
// 消费者，使用消息选择器来过滤对应的属性，语法格式为类SQL语法
consumer.subscribe("topic7", MessageSelector.bySql("age >= 18"));
consumer.subscribe("topic6", MessageSelector.bySql("name = 'litiedan'"));
1
2
3
4
5
6

SQL过滤需要依赖服务器的功能支持，在broker.conf配置文件中添加对应的功能项，并开启对应功能（enablePropertyFilter=true）

springboot整合

    org.apache.rocketmq
    rocketmq-spring-boot-starter
    2.0.3

1
2
3
4
5

rocketmq.name-server=localhost:9876
rocketmq.producer.group=demo_producer
1
2

@RestController
public class DemoProducers {
    @Autowired
    private RocketMQTemplate template;
    @RequestMapping("/producer")
    public String producersMessage() {
        User user = new User("sharfine", "123456789");
        template.convertAndSend("demo-topic", user);
        // 单向消息 具有最大的吞吐量，但可能会丢失消息（需要中等可靠性的情况，例如日志收集）
        // 指producer向 broker发送消息,执行API时直接返回，不等待broker 服务器的响应
        // rocketMQTemplate.sendOneWay("topic9",user);
        // 延时消息 同步消息支持延迟消息
        // 2秒消息还没发出那么就会抛异常 等级2,5秒发出
        // 同步消息是指producer向 broker发送消息,执行API时同步等待,直到broker服务器返回发送结果
        // rocketMQTemplate.syncSend("topic9", MessageBuilder.withPayload("test delay").build(),2000,2);
        // 异步消息
        // 指producer向broker发送消息时异步执行，不会影响后面逻辑。而异步里面会调用一个回调方法，来处理消息发送成功或失败的逻辑
        // rocketMQTemplate.asyncSend("topic9", user, new SendCallback() {

        // return JSON.toJSONString(user);
    // }
// rocketMQTemplate.syncSendOrderly("topic9", )
}
@Service
// @RocketMQMessageListener(topic = "topic9",consumerGroup = "group1",messageModel = MessageModel.BROADCASTING, selectorExpression = "tag1",selectorExpression = "age>18",selectorType= SelectorType.SQL92)
@RocketMQMessageListener(topic = "demo-topic", consumerGroup = "demo_consumer")
public class DemoConsumers1 implements RocketMQListener<user> {
    @Override
    public void onMessage(user user) {
        System.out.println("Consumers1接收消息:" + demoEntity.toString());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

消息的特殊处理

消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为分区有序或者全局有序。
顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。（队列内有序，队列外无序）

//发送时要指定对应的消息队列选择器
            SendResult result = producer.send(msg, new MessageQueueSelector() {
                //设置当前消息发送时使用哪一个消息队列
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    //根据发送的信息不同，选择不同的消息队列
                    //根据id来选择一个消息队列的对象，并返回->id得到int值
                    long orderId = order.getOrderId();
                    long mqIndex = orderId % list.size();
                    return list.get((int) mqIndex);
                }
            }, null);
//使用单线程的模式从消息队列中取数据，一个线程绑定一个消息队列
		consumer.registerMessageListener(new MessageListenerOrderly() {
            //使用MessageListenerOrderly接口后，对消息队列的处理由一个消息队列多个线程服务，转化为一个消息队列一个线程服务
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                for (MessageExt msg : list) {
                    System.out.println(Thread.currentThread().getName()+"。消息：" + new String(msg.getBody())+"。queueId:"+msg.getQueueId());
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

事务消息

• 正常事务过程

• 事务补偿过程
  

• 事务消息状态
  提交状态：允许进入队列，此消息与非事务消息无区别
  回滚状态：不允许进入队列，此消息等同于未发送过
  中间状态：完成了half消息的发送，未对MQ进行二次状态确认
  注意：事务消息仅与生产者有关，与消费者无关

基本概念

主题（Topic）
Apache RocketMQ 中消息传输和存储的顶层容器，用于标识同一类业务逻辑的消息。主题通过TopicName来做唯一标识和区分。更多信息，请参见主题（Topic）。
消息类型（MessageType）
Apache RocketMQ 中按照消息传输特性的不同而定义的分类，用于类型管理和安全校验。 Apache RocketMQ 支持的消息类型有普通消息、顺序消息、事务消息和定时/延时消息。

消息队列（MessageQueue）
队列是 Apache RocketMQ 中消息存储和传输的实际容器，也是消息的最小存储单元。 Apache RocketMQ 的所有主题都是由多个队列组成，以此实现队列数量的水平拆分和队列内部的流式存储。队列通过QueueId来做唯一标识和区分。更多信息，请参见队列（MessageQueue）。
队列的主要作用如下：

存储顺序性
队列天然具备顺序性，即消息按照进入队列的顺序写入存储，同一队列间的消息天然存在顺序关系，队列头部为最早写入的消息，队列尾部为最新写入的消息。消息在队列中的位置和消息之间的顺序通过位点（Offset）进行标记管理。
流式操作语义
Apache RocketMQ 基于队列的存储模型可确保消息从任意位点读取任意数量的消息，以此实现类似聚合读取、回溯读取等特性，这些特性是RabbitMQ、ActiveMQ等非队列存储模型不具备的。
Apache RocketMQ 默认提供消息可靠存储机制，所有发送成功的消息都被持久化存储到队列中，配合生产者和消费者客户端的调用可实现至少投递一次的可靠性语义。
Apache RocketMQ 队列模型和Kafka的分区（Partition）模型类似。在 Apache RocketMQ 消息收发模型中，队列属于主题的一部分，虽然所有的消息资源以主题粒度管理，但实际的操作实现是面向队列。例如，生产者指定某个主题，向主题内发送消息，但实际消息发送到该主题下的某个队列中。
Apache RocketMQ 中通过修改队列数量，以此实现横向的水平扩容和缩容。
每个主题下会由一到多个队列来存储消息，每个主题对应的队列数与消息类型以及实例所处地域（Region）相关，队列数暂不支持修改。
队列的名称属性在 Apache RocketMQ 服务端的不同版本中有如下差异：

服务端3.x/4.x版本：队列名称由{主题名称}+{BrokerID}+{QueueID}三元组组成，和物理节点绑定。

服务端5.x版本：队列名称为一个集群分配的全局唯一的字符串组成，和物理节点解耦。

因此，在开发过程中，建议不要对队列名称做任何假设和绑定。如果您在代码中自定义拼接队列名称并和其他操作进行绑定，一旦服务端版本升级，可能会出现队列名称无法解析的兼容性问题。
按照实际业务消耗设置队列数

Apache RocketMQ 的队列数可在创建主题或变更主题时设置修改，队列数量的设置应遵循少用够用原则，避免随意增加队列数量。
主题内队列数过多可能对导致如下问题：
集群元数据膨胀
Apache RocketMQ 会以队列粒度采集指标和监控数据，队列过多容易造成管控元数据膨胀。
客户端压力过大
Apache RocketMQ 的消息读写都是针对队列进行操作，队列过多容易产生空轮询请求，增加系统负荷。
常见队列增加场景
需要增加队列实现物理节点负载均衡
Apache RocketMQ 每个主题的多个队列可以分布在不同的服务节点上，在集群水平扩容增加节点后，为了保证集群流量的负载均衡，建议在新的服务节点上新增队列，或将旧的队列迁移到新的服务节点上。
需要增加队列实现顺序消息性能扩展
在 Apache RocketMQ 服务端4.x版本中，顺序消息的顺序性在队列内生效的，因此顺序消息的并发度会在一定程度上受队列数量的影响，因此建议仅在系统性能瓶颈时再增加队列。

消息（Message）
消息是 Apache RocketMQ 中的最小数据传输单元。生产者将业务数据的负载和拓展属性包装成消息发送到服务端，服务端按照相关语义将消息投递到消费端进行消费。更多信息，请参见消息（Message）。

消息视图（MessageView）
消息视图是 Apache RocketMQ 面向开发视角提供的一种消息只读接口。通过消息视图可以读取消息内部的多个属性和负载信息，但是不能对消息本身做任何修改。

消息标签（MessageTag）
消息标签是Apache RocketMQ 提供的细粒度消息分类属性，可以在主题层级之下做消息类型的细分。消费者通过订阅特定的标签来实现细粒度过滤。更多信息，请参见消息过滤。

消息位点（MessageQueueOffset）
消息是按到达Apache RocketMQ 服务端的先后顺序存储在指定主题的多个队列中，每条消息在队列中都有一个唯一的Long类型坐标，这个坐标被定义为消息位点。更多信息，请参见消费进度管理。

消费位点（ConsumerOffset）
一条消息被某个消费者消费完成后不会立即从队列中删除，Apache RocketMQ 会基于每个消费者分组记录消费过的最新一条消息的位点，即消费位点。更多信息，请参见消费进度管理。

消息索引（MessageKey）
消息索引是Apache RocketMQ 提供的面向消息的索引属性。通过设置的消息索引可以快速查找到对应的消息内容。

生产者（Producer）
生产者是Apache RocketMQ 系统中用来构建并传输消息到服务端的运行实体。生产者通常被集成在业务系统中，将业务消息按照要求封装成消息并发送至服务端。更多信息，请参见生产者（Producer）。

事务检查器（TransactionChecker）
Apache RocketMQ 中生产者用来执行本地事务检查和异常事务恢复的监听器。事务检查器应该通过业务侧数据的状态来检查和判断事务消息的状态。更多信息，请参见事务消息。

事务状态（TransactionResolution）
Apache RocketMQ 中事务消息发送过程中，事务提交的状态标识，服务端通过事务状态控制事务消息是否应该提交和投递。事务状态包括事务提交、事务回滚和事务未决。更多信息，请参见事务消息。

消费者分组（ConsumerGroup）
消费者分组是Apache RocketMQ 系统中承载多个消费行为一致的消费者的负载均衡分组。和消费者不同，消费者分组并不是运行实体，而是一个逻辑资源。在 Apache RocketMQ 中，通过消费者分组内初始化多个消费者实现消费性能的水平扩展以及高可用容灾。更多信息，请参见消费者分组（ConsumerGroup）。

消费者（Consumer）
消费者是Apache RocketMQ 中用来接收并处理消息的运行实体。消费者通常被集成在业务系统中，从服务端获取消息，并将消息转化成业务可理解的信息，供业务逻辑处理。更多信息，请参见消费者（Consumer）。

消费结果（ConsumeResult）
Apache RocketMQ 中PushConsumer消费监听器处理消息完成后返回的处理结果，用来标识本次消息是否正确处理。消费结果包含消费成功和消费失败。

订阅关系（Subscription）
订阅关系是Apache RocketMQ 系统中消费者获取消息、处理消息的规则和状态配置。订阅关系由消费者分组动态注册到服务端系统，并在后续的消息传输中按照订阅关系定义的过滤规则进行消息匹配和消费进度维护。更多信息，请参见订阅关系（Subscription）。

消息过滤
消费者可以通过订阅指定消息标签（Tag）对消息进行过滤，确保最终只接收被过滤后的消息合集。过滤规则的计算和匹配在Apache RocketMQ 的服务端完成。更多信息，请参见消息过滤。

重置消费位点
以时间轴为坐标，在消息持久化存储的时间范围内，重新设置消费者分组对已订阅主题的消费进度，设置完成后消费者将接收设定时间点之后，由生产者发送到Apache RocketMQ 服务端的消息。更多信息，请参见重置消费位点。

消息轨迹
在一条消息从生产者发出到消费者接收并处理过程中，由各个相关节点的时间、地点等数据汇聚而成的完整链路信息。通过消息轨迹，您能清晰定位消息从生产者发出，经由Apache RocketMQ 服务端，投递给消费者的完整链路，方便定位排查问题。

消息堆积
生产者已经将消息发送到Apache RocketMQ 的服务端，但由于消费者的消费能力有限，未能在短时间内将所有消息正确消费掉，此时在服务端保存着未被消费的消息，该状态即消息堆积。

事务消息
事务消息是Apache RocketMQ 提供的一种高级消息类型，支持在分布式场景下保障消息生产和本地事务的最终一致性。

定时/延时消息
定时/延时消息是Apache RocketMQ 提供的一种高级消息类型，消息被发送至服务端后，在指定时间后才能被消费者消费。通过设置一定的定时时间可以实现分布式场景的延时调度触发效果。

顺序消息
顺序消息是Apache RocketMQ 提供的一种高级消息类型，支持消费者按照发送消息的先后顺序获取消息，从而实现业务场景中的顺序处理。

发布-订阅（Pub/Sub）是一种消息范式，消息的发送者（称为发布者、生产者、Producer）会将消息直接发送给特定的接收者（称为订阅者、消费者、Comsumer）。而RocketMQ的基础消息模型就是一个简单的Pub/Sub模型。

topic
表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行消息订阅的基本单位。

生产者 Producer
发布消息的角色。Producer通过 MQ 的负载均衡模块选择相应的 Broker 集群队列进行消息投递，投递的过程支持快速失败和重试。

消费者 Consumer
消息消费的角色。

支持以推（push），拉（pull）两种模式对消息进行消费。
同时也支持集群方式和广播方式的消费。
提供实时消息订阅机制，可以满足大多数用户的需求。
名字服务器 NameServer
NameServer是一个简单的 Topic 路由注册中心，支持 Topic、Broker 的动态注册与发现。

主要包括两个功能：
Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；
路由信息管理，每个NameServer将保存关于 Broker 集群的整个路由信息和用于客户端查询的队列信息。Producer和Consumer通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。
NameServer通常会有多个实例部署，各实例间相互不进行信息通讯。Broker是向每一台NameServer注册自己的路由信息，所以每一个NameServer实例上面都保存一份完整的路由信息。当某个NameServer因某种原因下线了，客户端仍然可以向其它NameServer获取路由信息
代理服务器 Broker
Broker主要负责消息的存储、投递和查询以及服务高可用保证。

NameServer几乎无状态节点，因此可集群部署，节点之间无任何信息同步。Broker部署相对复杂。

在 Master-Slave 架构中，Broker 分为 Master 与 Slave。一个Master可以对应多个Slave，但是一个Slave只能对应一个Master。Master 与 Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId 来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个;

代理服务器 Broker
Broker主要负责消息的存储、投递和查询以及服务高可用保证。

NameServer几乎无状态节点，因此可集群部署，节点之间无任何信息同步。Broker部署相对复杂。
topic消息传输和存储的顶层容器，用于标识同一类业务逻辑的消息，由master和slave提供服务；
在 Master-Slave 架构中，Broker 分为 Master 与 Slave。一个Master可以对应多个Slave，但是一个Slave只能对应一个Master。Master 与 Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId 来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。
小结：
每个 Broker 与 NameServer 集群中的所有节点建立长连接，定时注册 Topic 信息到所有 NameServer。

Producer 与 NameServer 集群中的其中一个节点建立长连接，定期从 NameServer 获取Topic路由信息，并向提供 Topic 服务的 Master 建立长连接，且定时向 Master 发送心跳。Producer 完全无状态。

Consumer 与 NameServer 集群中的其中一个节点建立长连接，定期从 NameServer 获取 Topic 路由信息，并向提供 Topic 服务的 Master、Slave 建立长连接，且定时向 Master、Slave发送心跳。Consumer 既可以从 Master 订阅消息，也可以从Slave订阅消息。

RocketMQ集群工作流程

1. 启动NameServer
   启动NameServer。NameServer启动后监听端口，等待Broker、Producer、Consumer连接，相当于一个路由控制中心。

2. 启动 Broker
   启动 Broker。与所有 NameServer 保持长连接，定时发送心跳包。心跳包中包含当前 Broker 信息以及存储所有 Topic 信息。注册成功后，NameServer 集群中就有 Topic跟Broker 的映射关系。

3. 创建 Topic
   创建 Topic 时需要指定该 Topic 要存储在哪些 Broker 上，也可以在发送消息时自动创建Topic。

4. 生产者发送消息
   生产者发送消息。启动时先跟 NameServer 集群中的其中一台建立长连接，并从 NameServer 中获取当前发送的 Topic存在于哪些 Broker 上，轮询从队列列表中选择一个队列，然后与队列所在的 Broker建立长连接从而向 Broker发消息。

5. 消费者接受消息
   消费者接受消息。跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，然后开始消费消息。

消息
topic，表示要发送的消息的主题。
body 表示消息的存储内容
properties 表示消息属性
Tag: 不管是 RocketMQ 的 Tag 过滤还是延迟消息等都会利用 Properties 消息属性机制，这些特殊信息使用了系统保留的属性Key，设置自定义属性时需要避免和系统属性Key冲突。
Keys: 服务器会根据 keys 创建哈希索引，设置后，可以在 Console 系统根据 Topic、Keys 来查询消息，由于是哈希索引，请尽可能保证 key 唯一，例如订单号，商品 Id 等。
transactionId 会在事务消息中使用。

Topic 与 Tag 都是业务上用来归类的标识，区别在于 Topic 是一级分类，而 Tag 可以理解为是二级分类。使用 Tag 可以实现对 Topic 中的消息进行过滤。
Topic：消息主题，通过 Topic 对不同的业务消息进行分类。
Tag：消息标签，用来进一步区分某个 Topic 下的消息分类，消息从生产者发出即带上的属性。
Message 可以设置的属性值包括：
字段名 默认值 必要性 说明
Topic null 必填 消息所属 topic 的名称
Body null 必填 消息体
Tags null 选填 消息标签，方便服务器过滤使用。目前只支持每个消息设置一个
Keys null 选填 代表这条消息的业务关键词
Flag 0 选填 完全由应用来设置，RocketMQ 不做干预
DelayTimeLevel 0 选填 消息延时级别，0 表示不延时，大于 0 会延时特定的时间才会被消费
WaitStoreMsgOK true 选填 表示消息是否在服务器落盘后才返回应答。
什么时候该用 Topic，什么时候该用 Tag？
可以从以下几个方面进行判断：

消息类型是否一致：如普通消息、事务消息、定时（延时）消息、顺序消息，不同的消息类型使用不同的 Topic，无法通过 Tag 进行区分。
业务是否相关联：没有直接关联的消息，如淘宝交易消息，京东物流消息使用不同的 Topic 进行区分；而同样是天猫交易消息，电器类订单、女装类订单、化妆品类订单的消息可以用 Tag 进行区分。
消息优先级是否一致：如同样是物流消息，盒马必须小时内送达，天猫超市 24 小时内送达，淘宝物流则相对会慢一些，不同优先级的消息用不同的 Topic 进行区分。
消息量级是否相当：有些业务消息虽然量小但是实时性要求高，如果跟某些万亿量级的消息使用同一个 Topic，则有可能会因为过长的等待时间而“饿死”，此时需要将不同量级的消息进行拆分，使用不同的 Topic。
总的来说，针对消息分类，您可以选择创建多个 Topic，或者在同一个 Topic 下创建多个 Tag。但通常情况下，不同的 Topic 之间的消息没有必然的联系，而 Tag 则用来区分同一个 Topic 下相互关联的消息，例如全集和子集的关系、流程先后的关系。
Keys
Apache RocketMQ 每个消息可以在业务层面的设置唯一标识码 keys 字段，方便将来定位消息丢失问题。 Broker 端会为每个消息创建索引（哈希索引），应用可以通过 topic、key 来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。

// 订单Id
String orderId = “20034568923546”;
message.setKeys(orderId);
队列
为了支持高并发和水平扩展，需要对 Topic 进行分区，在 RocketMQ 中这被称为队列，一个 Topic 可能有多个队列，并且可能分布在不同的 Broker 上。
一般来说一条消息，如果没有重复发送（比如因为服务端没有响应而进行重试），则只会存在在 Topic 的其中一个队列中，消息在队列中按照先进先出的原则存储，每条消息会有自己的位点，每个队列会统计当前消息的总条数，这个称为最大位点 MaxOffset；队列的起始位置对应的位置叫做起始位点 MinOffset。队列可以提升消息发送和消费的并发度。
生产者
生产者（Producer）就是消息的发送者，Apache RocketMQ 拥有丰富的消息类型，可以支持不同的应用场景，在不同的场景中，需要使用不同的消息进行发送。比如在电商交易中超时未支付关闭订单的场景，在订单创建时会发送一条延时消息。这条消息将会在 30 分钟以后投递给消费者，消费者收到此消息后需要判断对应的订单是否已完成支付。如支付未完成，则关闭订单。如已完成支付则忽略，此时就需要用到延迟消息；电商场景中，业务上要求同一订单的消息保持严格顺序，此时就要用到顺序消息。在日志处理场景中，可以接受的比较大的发送延迟，但对吞吐量的要求很高，希望每秒能处理百万条日志，此时可以使用批量消息。在银行扣款的场景中，要保持上游的扣款操作和下游的短信通知保持一致，此时就要使用事务消息，下一节将会介绍各种类型消息的发送。
:::note 需要注意的是，生产环境中不同消息类型需要使用不同的主题，不要在同一个主题内使用多种消息类型，这样可以避免运维过程中的风险和错误。

消息发送

同步发送是最常用的方式，是指消息发送方发出一条消息后，会在收到服务端同步响应之后才发下一条消息的通讯方式，可靠的同步传输被广泛应用于各种场景，如重要的通知消息、短消息通知等。
同步发送的整个代码流程如下：

首先会创建一个producer。普通消息可以创建 DefaultMQProducer，创建时需要填写生产组的名称，生产者组是指同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。
设置 NameServer 的地址。Apache RocketMQ很多方式设置NameServer地址(客户端配置中有介绍)，这里是在代码中调用producer的API setNamesrvAddr进行设置，如果有多个NameServer，中间以分号隔开，比如"127.0.0.2:9876;127.0.0.3:9876"。
第三步是构建消息。指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤。
最后调用send接口将消息发送出去。同步发送等待结果最后返回SendResult，SendResut包含实际发送状态还包括SEND_OK（发送成功）, FLUSH_DISK_TIMEOUT（刷盘超时）, FLUSH_SLAVE_TIMEOUT（同步到备超时）, SLAVE_NOT_AVAILABLE（备不可用），如果发送失败会抛出异常。
单向模式调用sendOneway，不会对返回结果有任何等待和处理。
producer.sendOneway(msg);

事务消息发送

事务消息介绍
在一些对数据一致性有强需求的场景，可以用 Apache RocketMQ 事务消息来解决，从而保证上下游数据的一致性。
以电商交易场景为例，用户支付订单这一核心操作的同时会涉及到下游物流发货、积分变更、购物车状态清空等多个子系统的变更。当前业务的处理分支包括：

主分支订单系统状态更新：由未支付变更为支付成功。
物流系统状态新增：新增待发货物流记录，创建订单物流记录。
积分系统状态变更：变更用户积分，更新用户积分表。
购物车系统状态变更：清空购物车，更新用户购物车记录。
使用普通消息和订单事务无法保证一致的原因，本质上是由于普通消息无法像单机数据库事务一样，具备提交、回滚和统一协调的能力。 而基于 RocketMQ 的分布式事务消息功能，在普通消息基础上，支持二阶段的提交能力。将二阶段提交和本地事务绑定，实现全局提交结果的一致性。
事务消息发送分为两个阶段。第一阶段会发送一个半事务消息，半事务消息是指暂不能投递的消息，生产者已经成功地将消息发送到了 Broker，但是Broker 未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，如果发送成功则执行本地事务，并根据本地事务执行成功与否，向 Broker 半事务消息状态（commit或者rollback），半事务消息只有 commit 状态才会真正向下游投递。如果由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，Broker 端会通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit或是Rollback）。这样最终保证了本地事务执行成功，下游就能收到消息，本地事务执行失败，下游就收不到消息。总而保证了上下游数据的一致性。

集群消费模式适用于每条消息只需要被处理一次的场景，也就是说整个消费组会Topic收到全量的消息，而消费组内的消费分担消费这些消息，因此可以通过扩缩消费者数量，来提升或降低消费能力，
广播消费模式适用于每条消息需要被消费组的每个消费者处理的场景，也就是说消费组内的每个消费者都会收到订阅Topic的全量消息，因此即使扩缩消费者数量也无法提升或降低消费能力，
负载均衡
集群模式下，同一个消费组内的消费者会分担收到的全量消息，这里的分配策略是怎样的？如果扩容消费者是否一定能提升消费能力？

Apache RocketMQ 提供了多种集群模式下的分配策略，包括平均分配策略、机房优先分配策略、一致性hash分配策略等，可以通过如下代码进行设置相应负载均衡策略;
consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragely());
默认的分配策略是平均分配，这也是最常见的策略。平均分配策略下消费组内的消费者会按照类似分页的策略均摊消费。
但也不是一味地增加消费者就能提升消费能力的，比如下图中Topic的总队列数小于消费者的数量时，消费者将分配不到队列，即使消费者再多也无法提升消费能力。

消费位点

在Apache RocketMQ中每个队列都会记录自己的最小位点、最大位点。针对于消费组，还有消费位点的概念，在集群模式下，消费位点是由客户端提给交服务端保存的，在广播模式下，消费位点是由客户端自己保存的。一般情况下消费位点正常更新，不会出现消息重复，但如果消费者发生崩溃或有新的消费者加入群组，就会触发重平衡，重平衡完成后，每个消费者可能会分配到新的队列，而不是之前处理的队列。为了能继续之前的工作，消费者需要读取每个队列最后一次的提交的消费位点，然后从消费位点处继续拉取消息。但在实际执行过程中，由于客户端提交给服务端的消费位点并不是实时的，所以重平衡就可能会导致消息少量重复。

推、拉和长轮询

MQ的消费模式可以大致分为两种，一种是推Push，一种是拉Pull。

Push是服务端主动推送消息给客户端，优点是及时性较好，但如果客户端没有做好流控，一旦服务端推送大量消息到客户端时，就会导致客户端消息堆积甚至崩溃。

Pull是客户端需要主动到服务端取数据，优点是客户端可以依据自己的消费能力进行消费，但拉取的频率也需要用户自己控制，拉取频繁容易造成服务端和客户端的压力，拉取间隔长又容易造成消费不及时。

Apache RocketMQ既提供了Push模式也提供了Pull模式。

如果同一个消费者多次订阅某个Topic下的Tag，以最后一次订阅为准。
//如下错误代码中，Consumer只能订阅到TagFilterTest下TagB的消息，而不能订阅TagA的消息。
consumer.subscribe(“TagFilterTest”, “TagA”);
consumer.subscribe(“TagFilterTest”, “TagB”);
自定义属性设置在消息中
Message msg = new Message(“topic”, “tagA”, “Hello MQ”.getBytes());
// 设置自定义属性A，属性值为1。
msg.putUserProperties(“a”, “1”);
消息消费端： 使用SQL语法设置过滤表达式，并根据自定义属性过滤消息。
consumer.subscribe(“SqlFilterTest”,
MessageSelector.bySql(“(TAGS is not null and TAGS in (‘TagA’, ‘TagB’))” +
“and (a is not null and a between 0 and 3)”));
消息重试
若Consumer消费某条消息失败，则RocketMQ会在重试间隔时间后，将消息重新投递给Consumer消费，若达到最大重试次数后消息还没有成功被消费，则消息将被投递至死信队列

消息重试只针对集群消费模式生效；广播消费模式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息

最大重试次数：消息消费失败后，可被重复投递的最大次数。
consumer.setMaxReconsumeTimes(10);
重试间隔：消息消费失败后再次被投递给Consumer消费的间隔时间，只在顺序消费中起作用。

consumer.setSuspendCurrentQueueTimeMillis(5000);
顺序消费和并发消费的重试机制并不相同，顺序消费消费失败后会先在客户端本地重试直到最大重试次数，这样可以避免消费失败的消息被跳过，消费下一条消息而打乱顺序消费的顺序，而并发消费消费失败后会将消费失败的消息重新投递回服务端，再等待服务端重新投递回来，在这期间会正常消费队列后面的消息。

死信队列
当一条消息初次消费失败，RocketMQ会自动进行消息重试，达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息。此时，该消息不会立刻被丢弃，而是将其发送到该消费者对应的特殊队列中，这类消息称为死信消息（Dead-Letter Message），存储死信消息的特殊队列称为死信队列（Dead-Letter Queue），死信队列是死信Topic下分区数唯一的单独队列。如果产生了死信消息，那对应的ConsumerGroup的死信Topic名称为%DLQ%ConsumerGroupName，死信队列的消息将不会再被消费。可以利用RocketMQ Admin工具或者RocketMQ Dashboard上查询到对应死信消息的信息。

批量发送消息

批量发送的消息必须具有相同的Topic
批量发送的消息必须具有相同的刷盘策略
批量发送的消息不能是延时消息与事务消息
批量发送大小
默认情况下，一批发送的消息总大小不能超过4MB字节。如果想超出该值，有两种解决方案：
方案一：将批量消息进行拆分，拆分为若干不大于4M的消息集合分多次批量发送
方案二：在Producer端与Broker端修改属性

• Producer端需要在发送之前设置Producer的maxMessageSize属性
  ** Broker端需要修改其加载的配置文件中的maxMessageSize属性

于指定Topic消息的过滤有两种过滤方式：Tag过滤与SQL过滤
consumer.subscribe(“TOPIC”, “TAGA || TAGB || TAGC”);

SQL过滤是一种通过特定表达式对事先埋入到消息中的用户属性进行筛选过滤的方式。通过SQL过滤，
可以实现对消息的复杂过滤。不过，只有使用PUSH模式的消费者才能使用SQL过滤。

消息发送重试机制

生产者在发送消息时，若采用同步或异步发送方式，发送失败会重试，但oneway消息发送方式发送失败是没有重试机制的只有普通消息具有发送重试机制，顺序消息是没有的消息重投机制可以保证消息尽可能发送成功、不丢失，但可能会造成消息重复。消息重复在RocketMQ中是无法避免的问题消息重复在一般情况下不会发生，当出现消息量大、网络抖动，消息重复就会成为大概率事件producer主动重发、consumer负载变化（发生Rebalance，不会导致消息重复，但可能出现重复消费）也会导致重复消息消息重复无法避免，但要避免消息的重复消费。避免消息重复消费的解决方案是，为消息添加唯一标识（例如消息key），使消费者对消息进行消费判断来避免重复消费
消息发送重试有三种策略可以选择：同步发送失败策略、异步发送失败策略、消息刷盘失败策略同步发送失败策略
对于普通消息，消息发送默认采用round-robin策略来选择所发送到的队列。如果发送失败，默认重试2次。但在重试时是不会选择上次发送失败的Broker，而是选择其它Broker。当然，若只有一个Broker其也只能发送到该Broker，但其会尽量发送到该Broker上的其它Queue。同时，Broker还具有失败隔离功能，使Producer尽量选择未发生过发送失败的Broker作为目标Broker。其可以保证其它消息尽量不发送到问题Broker，为了提升消息发送效率，降低消息发送耗时。
异步发送失败策略
异步发送失败重试时，异步重试不会选择其他broker，仅在同一个broker上做重试，所以该策略无法保证消息不丢。

消息刷盘失败策略

消息刷盘超时（Master或Slave）或slave不可用（slave在做数据同步时向master返回状态不是SEND_OK）时，默认是不会将消息尝试发送到其他Broker的。不过，对于重要消息可以通过在Broker的配置文件设置retryAnotherBrokerWhenNotStoreOK属性为true来开启。

顺序消息的消费重试

顺序消息没有发送失败重试机制，但具有消费失败重试机制;对于无序消息（普通消息、延时消息、事务消息），当Consumer消费消息失败时，可以通过设置返回状态达到消息重试的效果。不过需要注意，无序消息的重试只对集群消费方式生效，广播消费方式不提供失败重试特性。即对于广播消费，消费失败后，失败消息不再重试，继续消费后续消息

重试队列

对于需要重试消费的消息，并不是Consumer在等待了指定时长后再次去拉取原来的消息进行消费，而是将这些需要重试消费的消息放入到了一个特殊Topic的队列中，而后进行再次消费的。这个特殊的队列就是重试队列。当出现需要进行重试消费的消息时，Broker会为每个消费组都设置一个Topic名称为%RETRY%consumerGroup@consumerGroup 的重试队列。Broker对于重试消息的处理是通过延时消息实现的。 先将消息保存到SCHEDULE_TOPIC_XXXX延迟队列中，延迟时间到后，会将消息投递到%RETRY%consumerGroup@consumerGroup重试队列中。

什么是死信队列

当一条消息初次消费失败，消息队列会自动进行消费重试；达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。这个队列就是死信队列（Dead-Letter Queue，DLQ），而其中的消息则称为死信消息（Dead-Letter Message，DLM）。死信队列是用于处理无法被正常消费的消息的。
死信队列具有如下特征：
死信队列中的消息不会再被消费者正常消费，即DLQ对于消费者是不可见的死信存储有效期与正常消息相同，均为 3 天（commitlog文件的过期时间），3 天后会被自动删除死信队列就是一个特殊的Topic，名称%DLQ%consumerGroup@consumerGroup ，即每个消费者组都有一个死信队列如果⼀个消费者组未产生死信消息，则不会为其创建相应的死信队列
死信消息的处理
实际上，当⼀条消息进入死信队列，就意味着系统中某些地方出现了问题，从而导致消费者无法正常消费该消息，比如代码中原本就存在Bug。因此，对于死信消息，通常需要开发人员进行特殊处理。最关键的步骤是要排查可疑因素，解决代码中可能存在的Bug，然后再将原来的死信消息再次进行投递消费。