Java连接RabbitMQ实现消息队列教程

Java 使用 RabbitMQ 实现消息队列通信，主要通过其官方提供的客户端库 amqp-client 来完成。整个流程包括连接 RabbitMQ 服务器、创建通道、声明队列、发送和接收消息等步骤。

1. 添加依赖

如果你使用的是 Maven 项目，在 pom.xml 中添加 RabbitMQ 的 Java 客户端依赖：

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

2. 建立连接并发送消息（生产者）

生产者负责将消息发送到指定的队列中。以下是一个简单的发送消息示例：

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

public class MessageProducer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost"); // RabbitMQ 服务地址
        factory.setPort(5672);       // 端口
        factory.setUsername("guest"); // 默认用户名
        factory.setPassword("guest"); // 默认密码

        // 建立连接和通道
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 声明队列（如果不存在则创建）
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            // 消息内容
            String message = "Hello RabbitMQ from Java!";

            // 发送消息
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("已发送: " + message);
        }
    }
}

3. 接收消息（消费者）

消费者监听队列，一旦有消息就进行处理：

import com.rabbitmq.client.*;

public class MessageConsumer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");

        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 -> {});
    }
}

4. 注意事项与常见配置

• RabbitMQ 服务必须运行：确保本地或远程的 RabbitMQ 服务正在运行，默认端口为 5672。
• 连接参数可配置：可根据实际环境修改 host、port、username、password。
• 队列声明是幂等的：多次声明相同名称的队列不会报错，只有第一次会创建。
• basicConsume 的 autoAck 参数：设为 true 表示自动确认消息；false 则需手动调用 channel.basicAck() 避免消息丢失。
• 资源释放：使用 try-with-resources 或手动关闭 connection 和 channel。

5. 扩展：使用交换机和路由

实际应用中常使用交换机（Exchange）来实现更灵活的路由。例如发布/订阅模式：

// 发送端绑定到 exchange
channel.exchangeDeclare("logs", "fanout");
channel.basicPublish("logs", "", null, message.getBytes());

// 接收端创建临时队列并绑定
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, "logs", "");

这样可以实现广播式消息分发。