RabbitMQ--Hello World（基础详解）

news2024/10/7 2:25:50

在这里插入图片描述

文章目录

先决条件
RabbitMQ 初识
RabbitMQ--Hello World
- 发送
- 接收

更多相关内容可查看

先决条件

本教程假定 RabbitMQ 已安装并在标准端口（5672）上运行。如果你使用不同的主机、端口或凭据，连接设置将需要调整。如未安装可查看Windows下载安装RabbitMQ客户端（2024最新篇）

RabbitMQ 初识

RabbitMQ 是一个消息代理：它接受和转发消息。你可以想想关于它作为邮局：当您将要投递的邮件放入邮箱，您可以确定信函承运人最终会送达邮寄给您的收件人。在这个类比中，RabbitMQ 是一个邮箱，一个帖子办公室和信使。

RabbitMQ 和邮局之间的主要区别在于它没有而是处理纸张，而是接受、存储和转发二进制数据 blob \u2012 消息。

RabbitMQ 和一般的消息传递使用一些术语。

生产无非是发送。发送消息的程序是制片人
队列是 RabbitMQ 中邮箱的名称。虽然消息会流动通过 RabbitMQ和您的应用程序，它们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束，它是本质上是一个大型消息缓冲区。
许多生产者可以发送发送到一个队列的消息，许多使用者可以尝试从一个队列接收数据。
表示队列的方式：queue_name
消费与接受具有相似的含义。消费者就是一个程序主要等待接收消息

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

RabbitMQ–Hello World

我们将用 Java 编写两个程序;一个发送单条消息的生产者，以及接收消息并打印出来。我们将在 Java API，专注于这个非常简单的事情，只是为了得到开始。这是一个消息传递的“Hello World”。

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

P
你好
C

发送

P
hello

我们将消息发布者（发送者）和消息使用者（接收者）称为。发布者将连接到 RabbitMQ，发送一条消息，然后退出。
我们需要导入一些类：


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

设置类并命名队列：

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

然后，我们可以创建与服务器的连接

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

该连接抽象了套接字连接，并负责处理协议版本协商和认证等为我们服务。这里我们连接到本地机器上的 RabbitMQ 节点 - 因此是 localhost。如果我们想连接到另一个节点上的节点机器，我们只需在此处指定其主机名或 IP 地址。

接下来我们创建一个通道，这是大多数 API 用于获取完成的事情驻留。请注意，我们可以使用 try-with-resources 语句因为两者都实现了. 这样，我们就不需要在代码中显式关闭它们。ConnectionChanneljava.lang.AutoCloseable

要发送，我们必须声明一个队列供我们发送到;然后我们可以发布一条消息到队列中，所有这些都在 try-with-resources 语句中：

channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "Hello World!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");

声明队列是幂等的 - 只有在不声明的情况下才会创建它已经存在。消息内容是字节数组，因此可以进行编码随便你喜欢什么。

发送不起作用！
如果这是您第一次使用 RabbitMQ，并且您没有看到“已发送” 消息，那么你可能会挠头，想知道什么可以错了。也许代理是在没有足够的可用磁盘空间的情况下启动的（默认情况下，它至少需要 200 MB 可用空间），因此拒绝接受消息。检查代理日志文件以确认并减少必要时限制。配置文件文档将向您展示如何设置 .disk_free_limit

生产者总体代码如下（便于大家复制粘贴）

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

import java.nio.charset.StandardCharsets;

public class Send {

    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            String message = "Hello World!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

接收

我们的消费者会收听来自 RabbitMQ，所以与发布单个消息的发布者不同，我们将让使用者保持运行以侦听消息并将其打印出来。

你好
C

我们将用于缓冲服务器推送给我们的消息的额外界面。DeliverCallback

设置与发布者相同;我们打开一个连接和一个 channel，并声明我们将要从中使用的队列。请注意，这与发布到的队列匹配。

public class Recv {

  private final static String QUEUE_NAME = "hello";

  public static void main(String[] argv) 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);
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

  }
}

请注意，我们也在此处声明队列。因为我们可能会开始消费者先于发布者，我们要确保队列存在在我们尝试从中消费消息之前。

我们为什么不使用 try-with-resource 语句来自动关闭通道和连接？通过这样做，我们只需制作程序即可继续前进，关闭所有东西，然后退出！这会很尴尬，因为我们希望在消费者倾听的同时，这个过程仍然有效异步地让消息到达。

我们将告诉服务器将来自队列。由于它将异步推送我们消息，因此我们提供了一个以对象的形式进行回调，该对象将缓冲消息，直到我们已准备好使用它们。这就是子类的作用。DeliverCallback

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    System.out.println(" [x] Received '" + message + "'");
};
channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });

消费者总体代码（便于大家复制粘贴）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;

public class Recv {

    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) 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);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x] Received '" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
    }
}