中间件知识点-消息中间件（Rabbitmq）一

消息中间件介绍

MQ的作用(优点)主要有以下三个方面：
a.异步
b.解耦
c.削峰

MQ的作用(缺点)主要有以下三个方面：
a.系统可用性降低
b.系统复杂度提高
c.存在消息一致性问题需要解决
备注：
引入MQ后系统的复杂度会大大提高。
以前服务之间可以进行同步的服务调用，引入MQ后，会变为异步调用，数据的链路就会变得更复杂。并且还会带来其他一些问
题。比如：如何保证消费不会丢失？不会被重复调用？怎么保证消息的顺序性等问题。(消息一致性问题)
A系统处理完业务，通过MQ发送消息给B、C系统进行后续的业务处理。如果B系统处理成功，C系统处理失败怎么办？这就需要考虑如何保证消息数据处理的一致性。

几大MQ产品特点比较（简单介绍）
rabbitmq可靠性：即不会丢数据，每个节点都存着全量数据（即数据一致性，rabbitmq仲裁队列quorum基于raft协议保证数据一致性，保证多半节点确认后才将消息写到队列中）。

kafka高可用：即使用了分区副本，把数据分散，保证高可用，不会因为一台机器宕机就无法使用。
kafka吞吐量大：使用了分区后可以在多台机器都存储数据，能存储的数据比较多
kafka性能非常好：使用了直接内存，零拷贝。

rocketmq借鉴了kafka、rabbitmq。

rabbitmq适合企业内部使用。一般企业使用rabbitmq就够了，一般企业5个节点就已经很多了。

在这里插入图片描述上面说的RabbitMq吞吐量比较低、消息积累会影响性能这些，其实在RabbitMq设计了Stream队列之后，这些都已经解决了。所以说RabbitMq的缺点只是对其过去版本的总结，现有的版本已经提升了很多了。因为加入了Stream队列这些，所以Stream队列借助了Kafka、RocketMq的思想。

另外上面说的RabbitMq吞吐量比较低、消息积累会影响性能这些是因为Classic和Quorum队列在收到消息时,如果是持久化消息,则会将消息储存在内存中,同时也会写入磁盘，所以消息量大的时候会内存会不断变小，吞吐量和性能就会变差。而RocketMQ或者kafka是会存在本地的，需要的时候再读取到内存，没那么耗性能。
1.kafka、rocketmq吞吐量高，是因为会将数据分散存储。那他会把收到的消息数据存到内存？
不会，他们是记录到日志中的。
但是在操作系统层面，当应用程序写入一个文件时，文件内容并不会直接写入到硬件当中，而是会先写入到操作系统中的一个缓存PageCache中。但可以认为会把收到的消息数据存到内存。
2.Stream队列不会将消息存在内存？不会。
3.Stream类似Kafka、RocketMq,以append-only log方式将消息记录到日志文件，然后消费的时候也是通过offset方式进行消费。
4.Quorum、Classic队列没有offset，不能快速找到数据，才需要将全部数据存放到内存。

这3个主要中间件一直在更新，一直在发展，越来越像。

activemq现在很少使用了，主要是用在很老的项目。

rocketmq适合业务比较多（topic比较多）的场景。

rabbitmq里面的routingkey和rocketmq里面的tag的区别？
routingkey在用在从交换机转换到哪个队列的，用在服务端。而tag用于消息过滤、消息路由、业务分类。

mq一般使用推模式。

rocketmq和kafka在互联网公司用的比较多，rabbitmq在中小公司用的比较多。
kafka一般用在大数据场景。

Rabbitmq

Rabbitmq单机搭建
略。

Rabbitmq集群(多节点的集群有两种方式模式)
a.默认的普通集群模式
不支持高可用：master宕机需要手动重启。
消息也不可靠：消息只存在一个节点中。
集群的各个节点之间只会有相同的元数据（比如队列、交换机。）。另外消息不会进行冗余，只存在一个节点中。消费时，如果消费的不是存有数据的节点， RabbitMQ会临时在节点之间进行数据传输，将消息从存有数据的节点传输到消费的节点。这种集群模式的消息可靠性不是很高。并且，这种集群模式也不支持高可用，即当某一个节点服务挂了后，需要手动重启服务，才能保证这一部分消息能正常消费。
所以这种集群模式只适合一些对消息安全性不是很高的场景。而在使用这种模式时，消费者应该尽量的连接上每一个节点，减少消息在集群中的传输。
b.镜像模式（会主动同步数据，而不是被动）
高可用：会自动选举新master节点。
消息可靠：消息会存在各节点。
但吞吐量不高：消息存在各节点，这样会消耗大量带宽。
这种模式是在普通集群模式基础上的一种增强方案，这也就是RabbitMQ的官方HA高可用方案。需要在搭建了普通集群之后再补充搭建。其本质区别在于，这种模式会在镜像节点中间主动进行消息同步，而不是在客户端拉取消息时临时同步。
并且在集群内部有一个算法会选举产生master和slave，当一个master挂了后，也会自动选出一个来。从而给整个集群提供高可用能力。
这种模式的消息可靠性更高，因为每个节点上都存着全量的消息。而他的弊端也是明显的，集群内部的网络带宽会被这种同步通讯大量的消耗（每个节点都存在全量数据，需要进行大量数据同步，不像kafka将数据分散存储），进而降低整个集群的性能。这种模式下，队列数量最好不要过多。

Exchanges:消息发送到RabbitMQ中后，会首先进入一个交换机，然后由交换机负责将数据转发到不同的队列中。即交换机的作用是可以对发送到RabbitMq Server的消息进行转发，转发到不同的队列中。
RabbitMQ中有多种不同类型的交换机来支持不同的路由策略。从Web管理界面就能看到，在每个虚拟主机中，RabbitMQ都会默认创建几个不同类型的交换机来。
交换机多用来与生产者打交道。生产者发送的消息通过Exchange交换机分配到各个不同的Queue队列上，而对于消息消费者来说，通常只需要关注自己感兴趣的队列就可以了。
在这里插入图片描述
队列消息删除机制：包括手动确认、自动确认、定时删除（TTL）等方式。
可以通过给队列设置TTL属性，来定时删除消息。
// 声明一个队列
String queueName = “testQueue”;
Map<String, Object> arguments = new HashMap<>();
// 设置消息的TTL为30000毫秒（30秒）
arguments.put(“x-message-ttl”, 30000);
channel.queueDeclare(queueName, true, false, false, arguments);

stream队列中的消息被消费了，是不会被删除的，而是通过类似kafka offset去定位从哪里去消费（只有Stream队列类型才会有offset概念，Classic经典队列没有），因为还需要发到其它消费组去推送消息。当过了一段时间后会统一删除队列的消息。

rabbitmq和kafka一样，同一个消费组的只有一个消费者才会消费消费消息。不同消费者组会重复消费一个消息。

rabbitmq没有类似kafka topic的概念，只有queue、exchanges的概念，只是在exchanges有topic的概念。而kafka、rocketmq有topic、queue的概念。
exchanges topic的概念？(Topic Exchange 类型支持通配符匹配路由键,然后将消息发送到符合特定规则的Queue 中)
RabbitMQ 和 Kafka 在消息队列架构上有一些不同之处。在 RabbitMQ 中，消息传递主要围绕着 Exchange 和 Queue 来进行，而 Exchange 中的类型包括 Direct、Fanout、Topic 和 Headers。
a.Exchange：Exchange 接收来自生产者的消息，并根据特定的规则将消息路由到一个或多个与之绑定的 Queue 中。Exchange 的类型决定了消息的路由方式。
b.Queue：Queue 是消息的容器，消费者从中接收消息。消息在进入 Queue 之前通过 Exchange 进行路由。
在 RabbitMQ 中，并没有像 Kafka 中的 Topic 概念那样直接映射到 Exchange。相对应的是，在 RabbitMQ 中的 Exchange 类型中，Topic Exchange 类型支持通配符匹配路由键，类似于 Kafka 中的 Topic，但并不是一一对应的概念。
因此，在 RabbitMQ 中，你可以使用 Topic Exchange 来实现类似于 Kafka Topic 的功能，通过设置特定的 Routing Key 和 Binding Key，将消息发送到符合特定规则的 Queue 中，实现灵活的消息路由和订阅功能。

Direct Exchange：将消息路由到绑定键（Binding Key）与消息的路由键（Routing Key）完全匹配的队列中。
Fanout Exchange：将消息广播到所有绑定的队列中。
Topic Exchange：根据消息的路由键和绑定键的模式进行匹配，将消息路由到一个或多个符合匹配规则的队列中。
Headers Exchange：使用消息的头部信息（headers）而不是路由键来决定消息的路由方式。

rabbitmq Exchanges四种类型：Direct、Fanout、Topic 和 Headers的java代码使用案例:
1.Direct Exchange

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

public class DirectExchangeExample {
    private static final String EXCHANGE_NAME = "direct_exchange";
    private static final String QUEUE_NAME = "direct_queue";
    private static final String ROUTING_KEY = "direct_routing_key";

    public static void main(String[] args) throws Exception {
        // 创建连接和通道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 声明一个 Direct Exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");

            // 声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            // 将队列绑定到 Exchange，并指定 Routing Key
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

            // 发送消息到 Exchange，并指定 Routing Key
            String message = "Hello, Direct Exchange!";
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, message.getBytes("UTF-8"));
            System.out.println("Sent message: '" + message + "'");
        }
    }
}

解释：
在这个例子中，我们创建了一个 Direct Exchange 名称为 direct_exchange。
创建了一个队列 direct_queue，并将它绑定到 direct_exchange，绑定时使用了路由键 direct_routing_key。
发送了一条消息到 Exchange，并指定了与队列绑定时相同的路由键 direct_routing_key。

2 Fanout Exchange

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

public class FanoutExchangeExample {
    private static final String EXCHANGE_NAME = "fanout_exchange";
    private static final String QUEUE_NAME1 = "fanout_queue1";
    private static final String QUEUE_NAME2 = "fanout_queue2";

    public static void main(String[] args) throws Exception {
        // 创建连接和通道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 声明一个 Fanout Exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

            // 声明两个队列
            channel.queueDeclare(QUEUE_NAME1, false, false, false, null);
            channel.queueDeclare(QUEUE_NAME2, false, false, false, null);

            // 将队列1绑定到 Exchange
            channel.queueBind(QUEUE_NAME1, EXCHANGE_NAME, "");

            // 将队列2绑定到 Exchange
            channel.queueBind(QUEUE_NAME2, EXCHANGE_NAME, "");

            // 发送消息到 Exchange
            String message = "Hello, Fanout Exchange!";
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
            System.out.println("Sent message: '" + message + "'");
        }
    }
}

解释：
在这个例子中，我们创建了一个 Fanout Exchange 名称为 fanout_exchange。
创建了两个队列 fanout_queue1 和 fanout_queue2，并将它们都绑定到 fanout_exchange。
发送了一条消息到 Exchange，消息会被广播到所有与 Exchange 绑定的队列上。

3.Topic Exchange

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

public class TopicExchangeExample {
    private static final String EXCHANGE_NAME = "topic_exchange";
    private static final String QUEUE_NAME1 = "topic_queue1";
    private static final String QUEUE_NAME2 = "topic_queue2";
    private static final String ROUTING_KEY1 = "topic.key1";
    private static final String ROUTING_KEY2 = "topic.key2";

    public static void main(String[] args) throws Exception {
        // 创建连接和通道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 声明一个 Topic Exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");

            // 声明两个队列
            channel.queueDeclare(QUEUE_NAME1, false, false, false, null);
            channel.queueDeclare(QUEUE_NAME2, false, false, false, null);

            // 将队列1绑定到 Exchange，使用通配符路由键 "*.key1"
            channel.queueBind(QUEUE_NAME1, EXCHANGE_NAME, "*.key1");

            // 将队列2绑定到 Exchange，使用通配符路由键 "topic.*"
            channel.queueBind(QUEUE_NAME2, EXCHANGE_NAME, "topic.*");

            // 发送消息到 Exchange，使用路由键 "topic.key1"
            String message1 = "Hello, Topic Exchange - Key1!";
            channel.basicPublish(EXCHANGE_NAME, "topic.key1", null, message1.getBytes("UTF-8"));
            System.out.println("Sent message 1: '" + message1 + "'");

            // 发送消息到 Exchange，使用路由键 "topic.key2"
            String message2 = "Hello, Topic Exchange - Key2!";
            channel.basicPublish(EXCHANGE_NAME, "topic.key2", null, message2.getBytes("UTF-8"));
            System.out.println("Sent message 2: '" + message2 + "'");
        }
    }
}

解释：
在这个例子中，我们创建了一个 Topic Exchange 名称为 topic_exchange。
创建了两个队列 topic_queue1 和 topic_queue2，并分别使用不同的通配符路由键将它们绑定到 topic_exchange。
发送了两条消息到 Exchange，分别使用了符合队列绑定模式的路由键。

4.Headers Exchange

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import java.util.HashMap;
import java.util.Map;

public class HeadersExchangeExample {
    private static final String EXCHANGE_NAME = "headers_exchange";
    private static final String QUEUE_NAME = "headers_queue";
    private static final Map<String, Object> HEADERS = new HashMap<>();

    public static void main(String[] args) throws Exception {
        // 创建连接和通道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 声明一个 Headers Exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "headers");

            // 声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            // 定义 headers 属性匹配规则
            HEADERS.put("header1", "value1");

            // 将队列绑定到 Exchange，指定 headers 匹配规则
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "", HEADERS);

            // 发送消息到 Exchange，并设置匹配的 headers 属性
            Map<String, Object> messageHeaders = new HashMap<>();
            messageHeaders.put("header1", "value1");
            String message = "Hello, Headers Exchange!";
            channel.basicPublish(EXCHANGE_NAME, "", new BasicProperties.Builder().headers(messageHeaders).build(), message.getBytes("UTF-8"));
            System.out.println("Sent message: '" + message + "'");
        }
    }
}

解释：
在这个例子中，我们创建了一个 Headers Exchange 名称为 headers_exchange。
创建了一个队列 headers_queue，并使用 headers 属性 header1=value1 将它绑定到 headers_exchange。
发送了一条消息到 Exchange，并设置了符合队列绑定 headers 属性的消息头。

rabbitmq实现顺序消费需要在单队列才能实现。

在一个worker(节点)中创建了VirtualHost、Queues后，集群中的其它节点也会有这个VirtualHosts、Queues。即集群中所有的节
点数据都是一样的，是会进行同步的。
消费者消费消息有推模式和拉模式

队列
Classic 经典队列(最为常用)
单机环境中（不是集群），拥有比较高的消息可靠性。因为可以持久化数据。
多节点下采用主从复制，可能会丢失数据。

Quorum 仲裁队列(针对Classic优化方式一)
（单机下通过持久化保证消息可靠性。集群下通过Raft协议保证分布式消息可靠性）
a.仲裁队列相比Classic经典队列，在分布式环境下对消息的可靠性保障更高。官方文档中表示，未来会使用Quorum仲裁队列代替传统Classic队列。
b.基于Raft一致性协议实现的一种新型的分布式消息队列，他实现了持久化，多备份的FIFO队列，主要就是针对RabbitMQ的镜像模式设计的。
c.仲裁队列的消息都是持久化的,同时队列的所有消息一直会保存在内存中。经典队列也一样，都会保存在内存中。
d.适合场景
Quorum队列更适合：需要队列长期存在，并且对容错、数据安全方面的要求比较高。对低延迟要求不高，且不需要临时消息的场景。例如电商系统的订单，引入MQ后，处理速度可以慢一点，但是订单不能丢失。
e.不适合场景
响应的Quorum队列不适合使用的场景：
1、一些临时使用的队列：比如transient临时队列，exclusive独占队列，或者经常会修改和删除的队列。（即Quorum不支持临时队列和独占队列）
2、对消息低延迟要求高：一致性算法会影响消息的延迟。
3、对数据安全性要求不高：Quorum队列需要消费者手动通知或者生产者手动确认。
4、队列消息积压严重：如果队列中的消息很大，或者积压的消息很多，就不要使用Quorum队列(因为Classic和Quorum队列在收到消息时,如果是持久化消息,则会将消息储存在内存中,同时也会写入磁盘，所以消息量大的时候会内存会不断变小，吞吐量和性能就会变差)。Quorum队列当前会将所有消息始终保存在内存中，直到达到内存使用极限。
所以，对于消息特别多且消息积压严重，我们一般不使用activemq/rabbitmq的经典队列和仲裁队列。

rabbimq中Quorum队列和Classic经典队列的区别总结：
1.可靠性方面：单节点时，经典队列和仲裁队列可靠性都很高。多节点集群下，经典队列仅采用主从复制，即一个主节点和多个从节点，主节点负责处理所有的读写操作，从节点则复制主节点的数据，数据可能会不一致。而仲裁队列采用了基于Raft一致性协议实现数据一致性，需要又集群中多半节点同意确认后，才会写入到队列中，可靠性更高。
单机下通过持久化保证消息可靠性。集群下通过Raft协议保证分布式消息可靠性
2.持久化：经典队列可以设置是否持久化，而仲裁队列都是持久化。同时队列的所有消息一直会保存在内存中。经典队列也一样，都会保存在内存中。所以对于消息特别多且消息积压严重，我们一般不使用activemq/rabbitmq的经典队列和仲裁队列。
3.仲裁队列大部分功能都是在经典队列基础上做减法，比如仲裁队列相比经典队列少了是否持久化、是否独占这些属性配置。
4.仲裁队列有毒消息机制，而经典队列没有。

Stream队列(针对Classic优化方式二，推荐)
Stream队列不会把所有的消息都放在内存中，需要的时候才从磁盘上去取（通过offset快速找到，而不像Quorum、Classic队列没有offset，不能快速找到数据，才需要将全部数据存放到内存）

–

RabbitMQ编程模型
（可以使用下面三种方式来使用RabbitMq）
从原生API、SpringBoot集成（使用最多）、SpringCloudStream集成。

RabbitMQ高级使用场景
1.Header路由
2.分组消费模式
3.死信队列
4.消费优先级与流量控制
5.远程数据分发插件
6.懒队列（懒加载） Lazy Queue
RabbitMQ从3.6.0版本开始，就引入了懒队列的概念。懒队列会尽可能早的将消息内容保存到硬盘当中，并且只有在用户请求到时，才临时从硬盘加载到RAM内存当中。
懒队列的设计目标是为了支持非常长的队列(数百万级别)。队列可能会因为一些原因变得非常长-也就是数据堆积。
懒队列适合消息量大且长期有堆积的队列，可以减少内存使用，加快消费速度。但是这是以大量消耗集群的网络及磁盘IO为代价的（即尽早将消息持久化，尽早多节点同步）。
7.消息分片存储插件
如何提高吞吐量？
上面的懒队列其实就是针对这个问题的一种解决方案。但是很显然，懒队列的方式属于治标不治本。真正要提升RabbitMQ单队列的吞吐量，还是要从数据也就是消息入手，只有将数据真正的分开存储才行。RabbitMQ提供的Sharding插件，就是一个可选的方案。他会真正将一个队列中的消息分散存储到不同的节点上，并提供多个节点的负载均衡策略实现对等的读与写功能。

RabbitMQ使用中的常见问题
一、如何使用RabbitMQ保证消息不丢失？
二、如何保证消息幂等
三、如何保证消息的顺序？
四、关于RabbitMQ的数据堆积问题
五、RabbitMQ的备份与恢复
六、RabbitMQ的性能监控
七、搭建HAProxy，实现高可用集群（在集群基础上增加负载均衡的能力，将客户端的请求能够尽量均匀的分配到集群中各个节点上）

总结:
基于MQ的事件驱动机制，给庞大的互联网应用带来了不一样的方向。MQ的异步、解耦、削峰三大功能特点在很多业务场景下都能带来极大的性能提升，在日常工作过程中，应该尝试总结这些设计的思想。虽然MQ的功能，说起来比较简单，但是随着MQ的应用逐渐深化，所需要解决的问题也更深入。对各种细化问题的挖掘程度，很大程度上决定了开发团队能不能真正Hold得住MQ产品。通常面向互联网的应用场景，更加注重MQ的吞吐量，需要将消息尽快的保存下来，再供后端慢慢消费。而针对企业内部的应用场景，更加注重MQ的数据安全性，在复杂多变的业务场景下，每一个消息都需要有更加严格的安全保障。而在当今互联网，Kafka（吞吐量大）是第一个场景(即面向互联网的应用场景)的不二代表，但是他会丢失消息的特性，让kafka的使用场景比较局限。RabbitMQ作为一个老牌产品，是第二个场景(即对企业内部的应用场景)最有力的代表。当然，随着互联网应用不段成熟，也不断有其他更全能的产品冒出来，比如阿里的RocketMQ(所以rocketmq是rabbitmq和kafka之后推出的)以及雅虎的Pulsar。但是不管未来MQ领域会是什么样子，RabbitMQ依然是目前企业级最为经典也最为重要的一个产品。他的功能最为全面，周边生态也非常成熟，并且RabbitMQ有庞大的Spring社区支持，本身也在吸收其他产品的各种优点，持续进化，所以未来RabbitMQ的重要性也会更加凸显。