RabbitMQ 教程 - "Hello World!"

简介

信息

先决条件

本教程假设 RabbitMQ 已安装并在 localhost 上的标准端口 (5672) 上运行。如果您使用不同的主机、端口或凭据，则需要调整连接设置。

哪里寻求帮助

如果您在学习本教程时遇到困难，可以通过 GitHub Discussions 或 RabbitMQ 社区 Discord 联系我们。

RabbitMQ 是一个消息代理：它接收并转发消息。你可以把它想象成一个邮局：当你把想要寄出的邮件放进邮箱时，你可以确信邮递员最终会将邮件送到你的收件人手中。在这个比喻中，RabbitMQ 就是邮箱、邮局和邮递员。

RabbitMQ 和邮局的主要区别在于，它不处理纸张，而是接收、存储和转发数据的二进制数据块——消息。

RabbitMQ 和消息传递通常会使用一些行话。

生产的意思仅仅是发送。发送消息的程序被称为生产者
队列是 RabbitMQ 中邮箱的名称。虽然消息在 RabbitMQ 和你的应用程序之间流动，但它们只能存储在队列中。队列的大小仅受主机内存和磁盘空间的限制，它本质上是一个大的消息缓冲区。

许多生产者 可以发送消息到同一个队列，许多消费者 可以尝试从同一个队列接收数据。

这就是我们表示队列的方式
消费的含义与接收类似。消费者 是一个主要等待接收消息的程序。

请注意，生产者、消费者和代理不一定需要驻留在同一台主机上；事实上，在大多数应用程序中它们都不会。一个应用程序也可以同时是生产者和消费者。

"Hello World"

(使用 AMQP 1.0 Java 客户端)

在本教程的这一部分，我们将编写两个 Java 程序；一个发送单条消息的生产者，以及一个接收消息并将其打印出来的消费者。我们将略过 Java API 的一些细节，专注于这个非常简单的事情以便入门。这就是消息传递领域的“Hello World”。

在下面的图表中，“P”是我们的生产者，“C”是我们的消费者。中间的框是一个队列——RabbitMQ 代表消费者存储的消息缓冲区。

AMQP 1.0 Java 客户端库

RabbitMQ 支持多种协议。本教程使用与 AMQP 0-9-1 相同的端口（默认 5672）通过 AMQP 1.0 进行通信。它需要 RabbitMQ 4.0 或更高版本。

请使用 RabbitMQ AMQP 1.0 Java 客户端 (com.rabbitmq.client:amqp-client)，不要使用传统的 AMQP 0-9-1 客户端 (com.rabbitmq:amqp-client)。请参阅 AMQP 1.0 客户端库和客户端参考文档。

将依赖项添加到您的构建配置中，例如使用 Maven
<dependency>
  <groupId>com.rabbitmq.client</groupId>
  <artifactId>amqp-client</artifactId>
  <version>1.0.0</version>
</dependency>
本教程系列的可运行源代码与 RabbitMQ 教程存储库（java-amqp 目录）中的其他端口一起提供。

现在我们已经在类路径中拥有了客户端，可以编写一些代码了。

发送

我们将消息发布者（发送方）命名为 Send，将消息消费者（接收方）命名为 Recv。发布者将连接到 RabbitMQ，发送一条消息，然后退出。

在 Send.java 中，我们需要导入一些类

import com.rabbitmq.client.amqp.Connection;
import com.rabbitmq.client.amqp.Environment;
import com.rabbitmq.client.amqp.Publisher;
import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

设置类并命名队列

public class Send {
  private static final String QUEUE_NAME = "hello";
  public static void main(String[] argv) throws Exception {
      ...
  }
}

然后创建一个环境 (environment) 和一个 连接 (connection)。环境保存共享设置；每个连接都指向代理。这里的 URI 指向本地机器上的代理及其默认虚拟主机（%2f 即 /）。

Environment environment = new AmqpEnvironmentBuilder()
  .connectionSettings()
  .uri("amqp://guest:guest@localhost:5672/%2f")
  .environmentBuilder()
  .build();
Connection connection = environment.connectionBuilder().build();

该连接抽象了套接字连接，并负责协议协商和身份验证。要连接到不同的主机，请更改 URI 中的主机（和凭据）。

connection.management().queue(QUEUE_NAME).quorum().queue().declare();

RabbitMQ 仍然使用 AMQ 0.9.1 模型（队列、交换机、绑定）进行拓扑。在发布之前先声明一个 仲裁队列 (quorum queue)。声明 API 使用流式链式调用；对于仲裁队列，它必须以 .quorum().queue().declare() 结尾。声明队列是幂等的：只有在队列不存在时才会创建它。

要发送消息，请创建一个寻址到该队列的 发布者，构建消息，并调用 publish。代理会异步报告结果；请等待闭锁 (latch)，以防程序在反馈到达之前退出。发布成功后的状态为 Publisher.Status.ACCEPTED

try (Publisher publisher = connection.publisherBuilder().queue(QUEUE_NAME).build()) {
  String message = "Hello World!";
  CountDownLatch latch = new CountDownLatch(1);
  publisher.publish(
      publisher.message(message.getBytes(StandardCharsets.UTF_8)),
      context -> {
        if (context.status() == Publisher.Status.ACCEPTED) {
          System.out.println(" [x] Sent '" + message + "'");
        }
        latch.countDown();
      });
  if (!latch.await(5, TimeUnit.SECONDS)) {
    throw new IllegalStateException("Timed out waiting for publish outcome");
  }
}

这是完整的 Send.java 类.

发送不起作用！

如果这是您第一次使用 RabbitMQ，但没有看到“Sent”消息，您可能会感到困惑，不知道哪里出了问题。也许代理启动时磁盘空间不足（默认需要至少 50 MB 可用空间），因此拒绝接收消息。检查代理的日志文件，看看是否有资源警报已记录，并在必要时降低可用磁盘空间阈值。配置指南将展示如何设置 disk_free_limit。

接收

我们的发布者就到这里了。我们的消费者会监听来自 RabbitMQ 的消息，所以与只发布一条消息的发布者不同，我们将保持消费者的运行状态，监听并打印消息。

Recv.java 中的代码使用相同的环境和连接设置。打开一个连接，然后声明同一个队列，这样消费者就可以在发布者之前启动

import com.rabbitmq.client.amqp.Connection;
import com.rabbitmq.client.amqp.Consumer;
import com.rabbitmq.client.amqp.Environment;
import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CountDownLatch;

public class Recv {

  private static final String QUEUE_NAME = "hello";

  public static void main(String[] argv) throws Exception {
    Environment environment = new AmqpEnvironmentBuilder()
        .connectionSettings()
        .uri("amqp://guest:guest@localhost:5672/%2f")
        .environmentBuilder()
        .build();
    Connection connection = environment.connectionBuilder().build();

    connection.management().queue(QUEUE_NAME).quorum().queue().declare();
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

    Consumer consumer = connection.consumerBuilder()
        .queue(QUEUE_NAME)
        .messageHandler((context, message) -> {
          String text = new String(message.body(), StandardCharsets.UTF_8);
          System.out.println(" [x] Received '" + text + "'");
          context.accept();
        })
        .build();

    new CountDownLatch(1).await();
  }
}

请注意，我们也在这里声明了队列。因为我们可能在发布者之前启动消费者，所以要确保在尝试从队列消费消息之前该队列已经存在。

为什么不在消费者中对 Environment 和 Connection 使用 try-with-resources？关闭它们会在 try 块结束时立即停止进程。示例保持消费者运行；使用 Ctrl+C 停止 JVM（或者扩展示例以在关闭时释放资源）。

使用 AMQP 1.0 时，消费者 必须结算 (settle) 每一条消息（accept、discard 或 requeue）。在这里，我们在打印消息体后调用 context.accept()。

这是完整的 Recv.java 类.

总而言之

创建一个 pom.xml，其中包含客户端和 Exec Maven 插件，以便您可以按名称运行类

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="https://w3org.cn/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.rabbitmq.examples</groupId>
  <artifactId>amqp10-tutorials</artifactId>
  <version>1.0-SNAPSHOT</version>
  <properties>
    <maven.compiler.release>11</maven.compiler.release>
  </properties>
  <dependencies>
    <dependency>
      <groupId>com.rabbitmq.client</groupId>
      <artifactId>amqp-client</artifactId>
      <version>1.0.0</version>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.codehaus.mojo</groupId>
        <artifactId>exec-maven-plugin</artifactId>
        <version>3.1.0</version>
        <configuration>
          <mainClass>${exec.mainClass}</mainClass>
        </configuration>
      </plugin>
    </plugins>
  </build>
</project>

将 Send.java 和 Recv.java 放在 src/main/java/ 下，然后在其中一个终端中运行消费者

mvn -q compile exec:java -Dexec.mainClass=Recv

然后运行发布者

mvn -q compile exec:java -Dexec.mainClass=Send

消费者将打印从发布者通过 RabbitMQ 获取的消息。消费者将保持运行，等待消息（使用 Ctrl+C 停止它），因此您可以尝试从另一个终端运行发布者。

列出队列

您可能希望查看 RabbitMQ 有哪些队列以及其中有多少消息。您可以使用 rabbitmqctl 工具（作为特权用户）来完成此操作。
sudo rabbitmqctl list_queues
在 Windows 上，省略 sudo。
rabbitmqctl.bat list_queues

现在进入第 2 部分并构建一个简单的 工作队列。

简介​

先决条件​

哪里寻求帮助​

"Hello World"​

(使用 AMQP 1.0 Java 客户端)​

AMQP 1.0 Java 客户端库​

发送​

发送不起作用！​

接收​

总而言之​

列出队列​

简介

先决条件

哪里寻求帮助

"Hello World"

(使用 AMQP 1.0 Java 客户端)

AMQP 1.0 Java 客户端库

发送

发送不起作用！

接收

总而言之

列出队列