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 保留的消息缓冲区 代表消费者。

发送

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

importcom.rabbitmq.client.ConnectionFactory;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.Channel;

设置类并命名队列：

publicclassSend{privatefinalstaticStringQUEUE_NAME="hello";publicstaticvoidmain(String[] argv)throwsException{...}}

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

ConnectionFactory factory =newConnectionFactory();
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

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

importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;importjava.nio.charset.StandardCharsets;publicclassSend{privatefinalstaticStringQUEUE_NAME="hello";publicstaticvoidmain(String[] argv)throwsException{ConnectionFactory factory =newConnectionFactory();
        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，所以与发布单个消息的发布者不同，我们将让使用者保持运行以侦听消息并将其打印出来。

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

DeliverCallback

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

publicclassRecv{privatefinalstaticStringQUEUE_NAME="hello";publicstaticvoidmain(String[] argv)throwsException{ConnectionFactory factory =newConnectionFactory();
    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 =newString(delivery.getBody(),"UTF-8");System.out.println(" [x] Received '"+ message +"'");};
channel.basicConsume(QUEUE_NAME,true, deliverCallback, consumerTag ->{});

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

importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;importcom.rabbitmq.client.DeliverCallback;importjava.nio.charset.StandardCharsets;publicclassRecv{privatefinalstaticStringQUEUE_NAME="hello";publicstaticvoidmain(String[] argv)throwsException{ConnectionFactory factory =newConnectionFactory();
        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 =newString(delivery.getBody(),StandardCharsets.UTF_8);System.out.println(" [x] Received '"+ message +"'");};
        channel.basicConsume(QUEUE_NAME,true, deliverCallback, consumerTag ->{});}}