当前位置：首页 > news >正文

简单入门RabbitMQ

news 2025/5/17 6:43:03

本章将带大家来写一个简单的程序，使用 Java 创建RabbitMQ 的生产者和消费者

依赖引入

在 Maven 仓库中输入 amqp-client：
在这里插入图片描述

找到第一个 RabbitMQ Java Client ，点击进去找到一个合适的版本然后将依赖引入到我们项目中的 pom.xml 文件中。

生产者代码编写

首先我们来回顾 RabbitMQ 的工作流程：
在这里插入图片描述

我们首先要建立连接Connection，通过 ConnectionFactory 来创建一个连接，在建立连接前我们需要设置好RabbitMQ 的参数：主机、端口号、用户名、密码、绑定的虚拟机。

		//建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);factory.setPort(Constants.PORT);factory.setUsername(Constants.NAME);factory.setPassword(Constants.PASSWORD);factory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = factory.newConnection();

接着我们需要开启信道 Channel，因为我们的Connection 是抽象的虚拟连接，并不是真正和RabbitMQ 虚拟机进行了连接，我们需要Connection 上的信道来进行通信：

		//开启信道Channel channel = connection.createChannel();

声明交换机，当我们没有显式地声明交换机的话，RabbitMQ 会使用默认的交换机，RabbitMQ默认会为每个连接自动创建一个名为 “”（空字符串）的直连交换机（Direct Exchange），所有队列都会隐式绑定到该交换机，绑定的路由键（Routing Key）就是队列名称。

声明队列：

//声明队列，使用内置的交换机，如果队列不存在会自动帮我们创建
channel.queueDeclare(Constants.HELLO_QUEUE, true, false, true, null);

参数介绍：

Queue.DeclareOk queueDeclare(String queue,
boolean durable, 
boolean exclusive, 
boolean autoDelete,
Map<String, Object> arguments) throws IOException;

queue: 队列名称

durable: 表示是否需要持久化，true 说明需要持久化，那么我们发送的消息就会进行落盘操作，如果RabbitMQ 重启了也可以找到这个消息，即把消息保存在硬盘中，如果是 false ，表示不需要进行持久化操作，那么我们发送的消息就只会在内存中，如果 RabbitMQ 重启了，消息也就不见了。

exclusive：表示是否独占，true: 表示该队列只对它的连接可见，且连接关闭时队列自动删除。
适用于临时场景（如RPC回调队列），确保队列不被其他连接共享。
注意：若其他连接尝试使用排他队列，会抛异常。

autoDelete：表示是否自动删除，true：当最后一个消费者取消订阅（如断开连接）时，队列自动删除。适用于动态队列（如临时任务队列），无需手动清理。
注意：若队列从未有过消费者，则不会触发删除。

arguments：设置队列的高级参数，在后面的章节中会详细展开。

发送消息：

for (int i = 0; i < 10; i++) {String msg = "hello work queue...." + i;channel.basicPublish("",Constants.HELLO_QUEUE, null, msg.getBytes());
}

参数说明：

void basicPublish(String exchange, 
String routingKey, 
BasicProperties props, 
byte[] body) throws IOException;

exchange：交换机的名称，由于上面我们没有声明交换机，所以这里填 “” 【空字符串】表示默认交换机

routingKey：路由键，也就说消息要发送到哪个队列上，路由键即为标识，在下一章节中会详细介绍。

props：属性配置，这个在后面的章节中也会介绍

body：消息内容

资源释放：

//6. 资源释放
channel.close();
connection.close();

这里我们先释放信道，然后再释放连接。

注意：释放连接之后，就不能释放信道了！！！因为连接都不存在了，信道也不会存在。
因此你也可以只释放连接。

运行程序之后，我们打开 RabbitMQ 的面板：
在这里插入图片描述

会发现我们成功将十条消息发送到队列中

在这里插入图片描述

通过输入Messages 数量，然后点击 Get Message(s)，就可以查看消息的详细信息了。

消费者代码编写

和生产者类似，都需要先设置好虚拟机的参数，然后创建连接，创建信道：

//建立连接
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(Constants.HOST);
factory.setPort(Constants.PORT);
factory.setUsername(Constants.NAME);
factory.setPassword(Constants.PASSWORD);
factory.setVirtualHost(Constants.VIRTUAL_HOST);
Connection connection = factory.newConnection();
//开启信道
Channel channel = connection.createChannel();

声明队列：

//声明队列
channel.queueDeclare(Constants.HELLO_QUEUE, true, false, true, null);

为什么消费者需要声明队列？
因为消费者要消费的前提是有队列可以消费
在分布式系统中，生产者和消费者的代码可能不会在同一台机器上，如果生产者还没有启动，反而消费者先启动了，消费者就会抛出异常，找不到队列
因此为了避免发生上面的异常情况，我们也需要在消费者中声明队列，这样，如果没有队列的话，RabbitMQ 就会帮我们创建队列出来。

接收消息和处理消息：

DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, 
Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到的消息：" + new String(body));}
};
channel.basicConsume(Constants.HELLO_QUEUE, true, consumer);

处理消息的主逻辑我们通过 DefaultConsumer 来进行编写，通过重写 handleDelivery 方法，来实现我们需要的逻辑。

参数介绍：

/*** No-op implementation of {@link Consumer#handleDelivery}.*/
@Override
public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException
{// no work to do
}