RabbitMQ简介
1. 概述
软件组件的解耦是软件设计中最重要的部分之一。实现此目的的一种方法是使用消息传递系统，它提供组件（服务）之间的异步通信方式。在本文中，我们将介绍其中一个系统：RabbitMQ。
RabbitMQ 是一个实现高级消息队列协议 （AMQP） 的消息代理。它为主要编程语言提供客户端库。
除了用于解耦软件组件外，RabbitMQ 还可用于：
- 执行后台操作
- 执行异步操作
2. 消息传递模型
首先，让我们快速、高层次地了解一下消息传递的工作原理。
简单地说，有两种类型的应用程序与消息传递系统交互：生产者和消费者。生产者是向代理发送（发布）消息的人，以及从代理接收消息的消费者。通常，这些程序（软件组件）在不同的机器上运行，RabbitMQ充当它们之间的通信中间件。
在本文中，我们将讨论一个简单的示例，其中包含两个将使用 RabbitMQ 进行通信的服务。其中一个服务将向 RabbitMQ 发布消息，另一个服务将使用。
3. 设置
首先，让我们在这里使用官方设置指南运行 RabbitMQ。
我们自然会使用 Java 客户端与 RabbitMQ 服务器进行交互;此客户端的Maven 依赖项为：
代码语言：javascript代码运行次数：0运行复制<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>4.0.0</version>
</dependency>Copy使用官方指南运行 RabbitMQ 代理后，我们需要使用 java 客户端连接到它：
代码语言：javascript代码运行次数：0运行复制ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
Copy我们使用ConnectionFactory来设置与服务器的连接，它还负责协议（AMQP）和身份验证。这里我们连接到本地主机上的服务器，我们可以使用setHost函数修改主机名。
如果 RabbitMQ 服务器不使用默认端口，我们可以使用setPort来设置端口;RabbitMQ 的默认端口是15672：
代码语言：javascript代码运行次数：0运行复制factory.setPort(15678);Copy我们可以设置用户名和密码：
代码语言：javascript代码运行次数：0运行复制factory.setUsername("user1");
factory.setPassword("MyPassword");Copy此外，我们将使用此连接来发布和使用消息。
4. 生产者
考虑一个简单的方案，其中 Web 应用程序允许用户向网站添加新产品。每当添加新产品时，我们都需要向客户发送电子邮件。
首先，让我们定义一个队列：
代码语言：javascript代码运行次数：0运行复制channel.queueDeclare("products_queue", false, false, false, null);Copy每次用户添加新产品时，我们都会向队列发布一条消息：
代码语言：javascript代码运行次数：0运行复制String message = "product details";
channel.basicPublish("", "products_queue", null, message.getBytes());Copy最后，我们关闭通道和连接：
代码语言：javascript代码运行次数：0运行复制channel.close();
connection.close();Copy此消息将由负责向客户发送电子邮件的另一个服务使用。
5. 消费者
让我们看看我们可以在消费者端实现什么;我们将声明相同的队列：
代码语言：javascript代码运行次数：0运行复制channel.queueDeclare("products_queue", false, false, false, null);Copy以下是我们如何定义将异步处理来自队列的消息的使用者：
代码语言：javascript代码运行次数：0运行复制DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(
String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body) throws IOException {
String message = new String(body, "UTF-8");
// process the message
}
};
channel.basicConsume("products_queue", true, consumer);
发布评论