跳到主要内容

RabbitMQ Stream 教程 - “Hello World!”

简介​

信息

先决条件​

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

使用 Docker​

如果您没有安装 RabbitMQ，您可以在 Docker 容器中运行它

docker run -it --rm --name rabbitmq -p 5552:5552 -p 15672:15672 -p 5672:5672  \
    -e RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS='-rabbitmq_stream advertised_host localhost' \
    rabbitmq:4-management

等待服务器启动，然后启用 stream 和 stream management 插件

docker exec rabbitmq rabbitmq-plugins enable rabbitmq_stream rabbitmq_stream_management

在哪里获得帮助​

如果您在学习本教程时遇到问题，您可以通过邮件列表或 Discord 社区服务器联系我们。

RabbitMQ Streams 在 RabbitMQ 3.9 中引入。更多信息请访问此处。

“Hello World”​

（使用 Java Stream 客户端）​

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

Java stream 客户端库​

RabbitMQ 支持多种协议。本教程使用 RabbitMQ stream 协议，这是一种用于 RabbitMQ streams 的专用协议。在 许多不同的语言中都有许多 RabbitMQ 客户端，请参阅每种语言的 stream 客户端库。我们将使用 RabbitMQ 提供的 Java stream 客户端。

RabbitMQ Java 客户端 0.16.0 及更高版本通过 Maven 仓库分发。

本教程假设您在 Windows 上使用 powershell。在 MacOS 和 Linux 上，几乎任何 shell 都可以工作。

设置​

本教程需要 java 在 PATH 中。要验证它，请运行

java --help

本教程将使用 Maven 来管理依赖项和构建项目。没有必要安装 Maven，因为本教程使用 Maven Wrapper。本教程的可执行版本可以在 RabbitMQ tutorials 仓库中找到。

接下来，让我们验证 Maven 是否正常工作

./mvnw --version

接下来，创建一个 pom.xml 文件，并将 RabbitMQ Stream Java 客户端作为依赖项

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>stream-client</artifactId>
    <version>0.16.0</version>
</dependency>

发送​

接下来，让我们为本教程的消息生产者（发送者）和消息消费者（接收者）部分创建两个文件。它们将分别被称为 Send.java 和 Receiver.java。

生产者将连接到 RabbitMQ，发送一条消息，然后退出。消费者将消费并将其打印到标准输出。

在 Send.java 的顶部，导入本教程使用的一些关键类

import com.rabbitmq.stream.*;
import java.io.IOException;

导入这些类后，现在可以实例化一个 Environment

Environment environment = Environment.builder().build();

stream Java 客户端的入口点是 Environment。它用于配置 RabbitMQ stream 发布者、stream 消费者和 stream 本身。

它抽象出了 TCP 或 TLS 套接字连接，并负责协议版本协商和身份验证等。

本教程假设 stream 发布者和消费者连接到本地运行的 RabbitMQ 节点，即 localhost。要连接到不同机器上的节点，只需使用 Environment.builder() 返回的构建器指定目标主机名或 IP 地址。

接下来，让我们创建一个生产者。

生产者还将声明一个 stream，它将向其发布消息，然后发布一条消息

String stream = "hello-java-stream";
environment.streamCreator().stream(stream).maxLengthBytes(ByteCapacity.GB(5)).create();
Producer producer = environment.producerBuilder().stream(stream).build();
producer.send(producer.messageBuilder().addData("Hello, World!".getBytes()).build(), null);
System.out.println(" [x] 'Hello, World!' message sent");

stream 声明操作是幂等的：stream 仅在尚不存在时才会被创建。

Stream 是一个仅追加的日志抽象，允许重复消费消息，直到它们过期。始终定义保留策略是一个好习惯。在上面的示例中，stream 的大小限制为 5 GiB。

消息内容是一个字节数组。应用程序可以使用任何适当的格式（如 JSON、MessagePack 等）对它们需要传输的数据进行编码。

当上面的代码运行完成时，生产者连接和 stream 系统连接将被关闭。这就是我们的生产者。

每次运行生产者时，它都会向服务器发送一条消息，并且该消息将附加到 stream。

完整的 Send.java 文件可以在 GitHub 上找到。

发送不起作用！​

如果这是您第一次使用 RabbitMQ，并且您没有看到 “Sent” 消息，那么您可能会感到困惑，想知道哪里出错了。可能是 broker 启动时没有足够的可用磁盘空间（默认情况下至少需要 50 MB 可用空间），因此拒绝接受消息。检查 broker 日志文件以查看是否有 资源告警记录，并在必要时降低可用磁盘空间阈值。配置指南将向您展示如何设置 disk_free_limit。

另一个原因可能是程序在消息到达 broker 之前退出。在某些客户端库中，发送是异步的：函数立即返回，但消息在通过网络传输之前在 IO 层排队。发送程序要求用户按键以完成该过程：消息有足够的时间到达 broker。stream 协议提供了一种确认机制来确保 broker 接收出站消息，但为了简单起见，本教程没有使用此机制。

接收​

本教程的另一部分，消费者，将连接到 RabbitMQ 节点并等待消息被推送到它。与本教程中发布单条消息并停止的生产者不同，消费者将持续运行，消费 RabbitMQ 将推送给它的消息，并将接收到的有效负载打印出来。

与 Send.java 类似，Receive.java 将需要首先导入一些类

import com.rabbitmq.stream.ByteCapacity;
import com.rabbitmq.stream.Consumer;
import com.rabbitmq.stream.Environment;
import com.rabbitmq.stream.OffsetSpecification;

在初始设置方面，消费者部分与生产者部分非常相似；我们使用默认连接设置并声明消费者将从中消费的 stream。

请注意，stream 名称必须与生产者使用的名称匹配。

Environment environment = Environment.builder().build();
String stream = "hello-java-stream";
environment.streamCreator().stream(stream).maxLengthBytes(ByteCapacity.GB(5)).create();

请注意，消费者部分也声明了 stream。这是为了允许先启动生产者或消费者中的任何一个。

Consumer 类用于实例化 stream 消费者，environment.consumerBuilder() 提供了一个构建器对象来配置它。最后，.messageHandler 方法接受已传递消息的处理程序。

offset 参数定义了消费者的起始点。在这种情况下，消费者从 stream 中可用的第一条消息开始。

Consumer consumer = environment.consumerBuilder()
            .stream(stream)
            .offset(OffsetSpecification.first())
            .messageHandler((unused, message) -> {
                System.out.println("Received message: " + new String(message.getBodyAsBinary()));
            }).build();

完整的 Receive.java 文件可以在 GitHub 上找到。

将它们放在一起​

为了运行这两个示例，请打开两个终端（shell）标签页。

本教程的两个部分可以以任何顺序运行，因为它们都声明了 stream。让我们首先运行消费者，以便在第一个发布者启动时，消费者将打印它

 ./mvnw -q compile exec:java '-Dexec.mainClass=Receive'

然后运行生产者

 ./mvnw -q compile exec:java '-Dexec.mainClass=Send'

消费者将打印它从发布者通过 RabbitMQ 收到的消息。消费者将保持运行，等待新的交付。尝试多次重新运行发布者以观察这一点。

Streams 与队列的不同之处在于，它们是可以重复消费的消息的仅追加日志。当多个消费者从一个 stream 消费时，它们将从第一条可用消息开始。

© . All rights reserved.