【中间件】RabbitMQ

标签：队列中间件 RabbitMQ 交换机消息 import message

一、RabbitMQ概念

RabbitMQ是一个开源消息中间件，使用Erlang语言编写。它实现了高级消息队列协议（AMQP），提供可靠的消息传递机制。
RabbitMQ允许应用程序在分布式环境中进行异步通信，通过将消息发送到中间件而不是直接发送给接收方，应用程序可以通过解耦和异步处理来提高系统的可靠性和可扩展性。
RabbitMQ支持多种消息模式，包括点对点通信、发布/订阅模式和消息路由。它还提供了丰富的功能，如消息持久化、消息优先级、消息确认和消息队列监控等。
由于其高可用性、高性能和可靠性，RabbitMQ被广泛应用于各种分布式系统和微服务架构中。

二、RabbitMQ的架构

在这里插入图片描述

Producer（生产者）： 生产者负责发送消息到RabbitMQ的消息队列中。它可以是任何发送消息的应用程序或系统。
Message Queue（消息队列）： 消息队列是RabbitMQ中的主要组件，用于存储消息。它充当生产者和消费者之间的中间媒介。
Exchange（交换机）： 交换机是消息的分发中心，负责接收生产者发送的消息，并将消息路由到适当的消息队列。RabbitMQ支持多种类型的交换机，如直连交换机、主题交换机、扇形交换机和头交换机等。
Consumer（消费者）： 消费者从消息队列中接收消息，并进行消息的处理或消费。它可以是任何接收和处理消息的应用程序或系统。
Binding（绑定）： 绑定将交换机和消息队列关联起来，定义了消息如何从交换机路由到特定的队列。绑定是通过指定绑定键（binding key）来实现的。
Virtual Host（虚拟主机）： 虚拟主机是RabbitMQ提供的一种逻辑隔离机制，允许将消息队列和交换机等资源分组到独立的虚拟环境中，以避免不同应用程序之间的冲突。
Connection（连接）： 连接是RabbitMQ与生产者和消费者之间的通信链接。一个连接可以包含多个信道（channel），每个信道又可以进行消息的传输和交互。

三、RabbitMQ的使用场景

异步任务处理： RabbitMQ可以用于将任务发送到队列，然后后台的工作进程可以从队列中接收和处理任务。这在分布式系统中很有用，可以将工作负载分布到多个进程或服务器上。
示例： 一个电商网站接收到用户下单请求，可以将订单信息发布到RabbitMQ的任务队列中，后台的工作进程可以从队列中接收并处理订单。这样可以使网站的前台和后台解耦，提高并发处理能力。
消息传递： RabbitMQ可以用于在应用程序之间进行可靠的消息传递。发送方将消息发送到队列，接收方可以从队列中接收和处理消息。这种模式适用于解耦发送方和接收方，提高应用程序的可靠性和可扩展性。
示例： 一个微服务架构的应用中，不同的服务之间需要进行通信。例如，用户服务可以将用户注册消息发送到RabbitMQ的消息队列中，然后其他服务（如邮件服务、短信服务等）可以从队列中接收并处理这些消息，完成用户注册后续操作。
发布/订阅模式： RabbitMQ支持发布/订阅模式，其中发布者将消息发布到交换机，然后多个订阅者可以从交换机接收消息。这种模式适用于将消息广播给多个接收方，例如日志记录或事件通知。
示例： 一个新闻发布系统，新闻内容发布到RabbitMQ的交换机中，多个订阅者（如网站、移动应用等）可以从交换机中接收并展示新闻内容。这样可以实现消息的广播和消费者的灵活订阅。
负载均衡： RabbitMQ可以与负载均衡器结合使用，将消息传递给多个消费者，以实现负载均衡和高可用性。
示例： 一个Web应用中，多个后台处理进程需要处理用户请求。将用户请求发送到RabbitMQ的任务队列中，多个消费者（即后台处理进程）可以从队列中接收并处理请求，实现负载均衡和高可用性。
消息路由： RabbitMQ支持多种消息路由模式，例如直接路由、主题路由和头部路由。这些模式可以根据消息内容或其他条件将消息路由到不同的队列，以实现灵活的消息传递。
示例： 一个电商网站的订单系统中，根据订单的地区属性将订单消息路由给不同的处理队列。例如，国内订单发送到国内订单队列，国际订单发送到国际订单队列。这样可以根据具体需求灵活地将消息路由到不同的队列中。
消息持久化： RabbitMQ可以将消息持久化到磁盘，以防止消息丢失。这对于重要的消息和数据的可靠性很重要。
示例： 一个日志收集系统，将应用程序产生的日志消息发送到RabbitMQ的持久化队列中。即使RabbitMQ发生故障或重启，消息仍然可靠地存储在磁盘上，确保日志消息不会丢失。

总而言之，RabbitMQ在需要可靠、可扩展和高性能消息传递的分布式系统中非常有用。它可以用于各种不同的场景，从异步任务处理到消息传递和负载均衡。

四、RabbitMQ的重要机制

ACK机制：RabbitMQ支持消息的确认机制，即生产者发送消息后会收到一个确认信号，以确保消息已经成功发送到队列中。消费者在处理完消息后，需要发送ACK（确认）消息给RabbitMQ，告知消息已经被成功处理，此时RabbitMQ可以删除消息。如果消费者没有发送ACK，RabbitMQ会认为消息未被成功消费，并重新将消息发送给其他消费者。这种机制可以确保消息的可靠传递和处理。
TTL机制：RabbitMQ支持设置消息的生存时间，即消息的TTL（Time To Live），超过生存时间的消息将被丢弃。这个机制可以用来处理一些临时性的消息，避免消息的堆积。
延迟队列：RabbitMQ支持延迟队列，可以将消息发送到一个特定的队列，并设置一个延迟时间。当延迟时间到达后，消息才会被投递到目标队列，然后被消费者处理。这种机制可以用于解决各种延迟相关的问题，例如定时任务调度。
死信队列：RabbitMQ支持死信队列，即当消息无法被消费者处理时，可以将它们发送到另一个队列进行处理，以防止消息丢失。
优先级队列：RabbitMQ支持消息的优先级设置，可以为每个消息设置一个优先级。优先级高的消息会被优先投递到消费者，以确保重要消息能够尽快得到处理。
持久化：RabbitMQ支持消息的持久化，即使在服务器重启后也能保留消息。生产者可以将消息设置为持久化，而队列和交换机也可以设置为持久化。这样可以确保即使在系统故障或服务器崩溃时，消息也不会丢失。
集群和高可用性：RabbitMQ支持集群和高可用性配置，可以通过将多个节点组成集群来提高可靠性和性能。
镜像队列：RabbitMQ支持镜像队列，即将队列的副本分布在多个节点上，以提高可用性和容错性
限流机制：RabbitMQ允许在消费者和生产者之间设置限制，以控制消息传输的速率。

五、RabbitMQ的优点

可靠性：RabbitMQ通过持久化机制和确认机制来确保消息的可靠传输。它支持消息的持久化存储和传送，并确保生产者发送的消息被正确接收和保存。
灵活的消息模式：RabbitMQ支持多种消息模式，包括点对点模式、发布/订阅模式、请求/回复模式等。这使得开发人员可以根据具体需求选择适合的消息模式。
高性能：RabbitMQ使用基于Erlang语言开发的AMQP（Advanced Message Queuing Protocol）作为通信协议，具有高性能和低延迟的特点。它能够处理高并发的消息传输和处理需求。
可扩展性：RabbitMQ具有良好的扩展性，可以轻松地横向扩展、构建集群、分布式部署等。它支持镜像队列和分片机制，可以将队列的副本分布在多个节点上，提高可用性和容错性。
多语言支持：RabbitMQ提供了多种客户端库，支持多种编程语言，包括Java、Python、C#、Ruby、Node.js等。这使得开发人员可以使用自己熟悉的编程语言与RabbitMQ进行交互。
管理界面：RabbitMQ提供了一个直观的Web管理界面，可以用于监控和管理消息队列。通过管理界面，管理员可以查看消息的传输情况、队列的状态、交换机的绑定等。
可靠的社区支持：RabbitMQ是一个开源项目，拥有庞大的社区支持。开发者可以在社区中获得技术支持、问题解答、新功能开发等。

总的来说，RabbitMQ是一个功能丰富、可靠性高、性能优越、具有良好扩展性的消息队列中间件，适用于各种异步通信和解耦应用的场景。

六、RabbitMQ的Java使用示例

1、添加依赖

Maven:

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>版本号</version>
</dependency>

Gradle:

implementation 'com.rabbitmq:amqp-client:版本号'

2、发送消息（生产者）

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

public class Producer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        // 创建连接
        Connection connection = factory.newConnection();
        
        // 创建信道
        Channel channel = connection.createChannel();
        
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 发送消息
        String message = "Hello, RabbitMQ!";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
        System.out.println("Sent message: " + message);
        
        // 关闭连接和信道
        channel.close();
        connection.close();
    }
}

3、接收消息（消费者）

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

public class Consumer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        // 创建连接
        Connection connection = factory.newConnection();
        
        // 创建信道
        Channel channel = connection.createChannel();
        
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        // 创建消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, true, consumer);
        
        // 接收消息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Received message: " + message);
        }
    }
}

以上示例展示了生产者将消息发送到名为 “hello” 的队列中，消费者从该队列中接收消息并进行处理。

七、RabbitMQ的SpringBoot使用示例

1、添加依赖
在项目的Maven或Gradle配置文件中添加Spring Boot和RabbitMQ的依赖项。

Maven:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

Gradle:

implementation 'org.springframework.boot:spring-boot-starter-amqp'

2、配置RabbitMQ连接信息
application.yml：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

3、创建生产者
创建一个生产者类，用于发送消息。

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class Producer {
    private static final String QUEUE_NAME = "hello";

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private Queue queue;

    public void send(String message) {
        rabbitTemplate.convertAndSend(queue.getName(), message);
        System.out.println("Sent message: " + message);
    }
}

4、创建消费者
创建一个消费者类，用于接收消息。

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class Consumer {

    @RabbitListener(queues = "hello")
    public void receive(String message) {
        System.out.println("Received message: " + message);
    }
}

5、发送和接收消息
在需要发送和接收消息的地方，通过Autowired注入Producer对象，调用send方法发送消息。

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        ConfigurableApplicationContext context = SpringApplication.run(Application.class, args);
        Producer producer = context.getBean(Producer.class);
        producer.send("Hello, RabbitMQ!");
    }
}

标签：队列,中间件,RabbitMQ,交换机,消息,import,message
From： https://blog.csdn.net/qq_44845473/article/details/144331756