RabbitMQ简介-Java快速进阶教程

jack.yang

发布于 2025-04-05 12:15:05

9400

代码可运行

运行总次数：0

代码可运行

1. 概述

软件组件的解耦是软件设计中最重要的部分之一。实现此目的的一种方法是使用消息传递系统，它提供组件（服务）之间的异步通信方式。在本文中，我们将介绍其中一个系统：RabbitMQ。

RabbitMQ 是一个实现高级消息队列协议（AMQP）的消息代理。它为主要编程语言提供客户端库。

除了用于解耦软件组件外，RabbitMQ 还可用于：

执行后台操作
执行异步操作

2. 消息传递模型

首先，让我们快速、高层次地了解一下消息传递的工作原理。

简单地说，有两种类型的应用程序与消息传递系统交互：生产者和消费者。生产者是向代理发送（发布）消息的人，以及从代理接收消息的消费者。通常，这些程序（软件组件）在不同的机器上运行，RabbitMQ充当它们之间的通信中间件。

在本文中，我们将讨论一个简单的示例，其中包含两个将使用 RabbitMQ 进行通信的服务。其中一个服务将向 RabbitMQ 发布消息，另一个服务将使用。

3. 设置

首先，让我们在这里使用官方设置指南运行 RabbitMQ。

我们自然会使用 Java 客户端与 RabbitMQ 服务器进行交互;此客户端的Maven 依赖项为：

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>4.0.0</version>
</dependency>Copy

使用官方指南运行 RabbitMQ 代理后，我们需要使用 java 客户端连接到它：

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
Copy

我们使用ConnectionFactory来设置与服务器的连接，它还负责协议（AMQP）和身份验证。这里我们连接到本地主机上的服务器，我们可以使用setHost函数修改主机名。

如果 RabbitMQ 服务器不使用默认端口，我们可以使用setPort来设置端口;RabbitMQ 的默认端口是15672：

factory.setPort(15678);Copy

我们可以设置用户名和密码：

factory.setUsername("user1");
factory.setPassword("MyPassword");Copy

此外，我们将使用此连接来发布和使用消息。

4. 生产者

考虑一个简单的方案，其中 Web 应用程序允许用户向网站添加新产品。每当添加新产品时，我们都需要向客户发送电子邮件。

首先，让我们定义一个队列：

channel.queueDeclare("products_queue", false, false, false, null);Copy

每次用户添加新产品时，我们都会向队列发布一条消息：

String message = "product details"; 
channel.basicPublish("", "products_queue", null, message.getBytes());Copy

最后，我们关闭通道和连接：

channel.close();
connection.close();Copy

此消息将由负责向客户发送电子邮件的另一个服务使用。

5. 消费者

让我们看看我们可以在消费者端实现什么;我们将声明相同的队列：

channel.queueDeclare("products_queue", false, false, false, null);Copy

以下是我们如何定义将异步处理来自队列的消息的使用者：

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);