RabbitMQ下载与配置

安装Erlang

Erlang 下载地址如下：

https://erlang.org/download/otp_versions_tree.html

安装 RabbitMQ

RabbitMQ 下载地址如下：

https://www.rabbitmq.com/install-windows.html

查看服务，服务已经正常启动

打开Command Prompt

输入rabbitmqctl.bat status,出现问题

Error: unable to perform an operation on node 'rabbit@LAPTOP-8VJSJ45O'. Please see diagnostics information and suggestions below.
Authentication failed (rejected by the remote node), please check the Erlang cookie

解决问题

• 使用旧版本Erlang
• cookie替换到正确的位置

打开路径C:\Windows\System32\config\systemprofile、

这里有一个.erlang.cookie，复制这个文件到C:\Users\你的用户名下，覆盖文件

再次输入

rabbitmqctl.bat status

可以输入以下命令来启用客户端管理 UI 插件：

rabbitmq-plugins enable rabbitmq_management

在浏览器地址栏输入 http://localhost:15672/ 可以进入管理端界面

默认的用户名和密码是“guest”和“guest”。

在 Java 中使用 RabbitMQ 

第一步，在项目中添加 RabbitMQ 客户端依赖：

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

第二步，我们来模拟一个最简单的场景，一个生产者发送消息到队列中，一个消费者从队列中读取消息并打印。

新建生产者类 Wanger ：

public class Wanger {private final static String QUEUE_NAME = "queue";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, false, false, false, null);String message = "发送消息";channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));System.out.println("发送 '" + message + "'");}}
}

1）QUEUE_NAME 为队列名，也就是说，生产者发送的消息会放到 queue队列中。

2）创建服务器连接：ConnectionFactory 是一个非常方便的工厂类，可用来创建到 RabbitMQ 的默认连接（主机名为“localhost”）。然后，创建一个通道（ Channel）来发送消息。Connection 和 Channel 类都实现了 Closeable 接口，所以可以使用 try-with-resource 语句

3）在发送消息的时候，必须设置队列名称，通过 queueDeclare() 方法设置。

4）basicPublish() 方法用于发布消息

• 第一个参数为交换机（exchange），当前场景不需要，因此设置为空字符串；
• 第二个参数为路由关键字（routingKey），暂时使用队列名填充；
• 第三个参数为消息的其他参数（BasicProperties），暂时不配置；
• 第四个参数为消息的主体，这里为 UTF-8 格式的字节数组，可以有效地杜绝中文乱码。

接下来新建消费者类 XiaoXinag：

public class XiaoXiang {private final static String QUEUE_NAME = "love";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);System.out.println("等待接收消息");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" 接收到的消息 '" + message + "'");};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });}
}

1）没有使用 try-with-resource 语句来自动关闭连接和通道，因为我们希望消费者能够一直保持连接，直到我们强制关闭它。

2）在接收消息的时候，必须设置队列名称，通过 queueDeclare() 方法设置。

3）由于 RabbitMQ 将会通过异步的方式向我们推送消息，因此我们需要提供了一个回调，该回调将对消息进行缓冲，直到我们做好准备接收它们为止。

basicConsume() 方法用于接收消息：

• 第一个参数为队列名（queue），和生产者相匹配。

• 第二个参数为 autoAck，如果为 true 的话，表明服务器要一次性交付消息。

• 第三个参数为 DeliverCallback，也就是消息的回调函数。

• 第四个参数为 CancelCallback

在消息发送的过程中，也可以使用 RabbitMQ 的管理面板查看到消息的走势图，如下所示。