RabbitMQ消息中间件的Java实践指南

什么是RabbitMQ？

RabbitMQ是一个开源的消息队列中间件，由Erlang语言编写。它使用AMQP协议（Advanced Message Queuing Protocol），允许不同的应用程序轻松交换信息。你可以把它想象成一座快递站，各个应用程序就像快递员，它们通过RabbitMQ传递包裹（消息）。

为什么选择RabbitMQ？简单来说，它能帮你实现异步处理、解耦系统、削峰填谷等复杂功能。比如，在电商场景下，用户下单后需要通知库存、支付、物流等多个模块，这些模块可以利用RabbitMQ互相通信，而不是直接依赖彼此。

今天，我们就通过一系列Java代码示例，带你玩转RabbitMQ！

第一步：添加依赖

在开始之前，你需要在Maven项目中引入RabbitMQ的Java客户端库。打开pom.xml文件，添加如下依赖：

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

小贴士：如果你用的是Gradle，可以这样写：

implementation 'com.rabbitmq:amqp-client:5.14.2'

第二步：创建RabbitMQ连接

连接RabbitMQ就像拨通电话一样简单，但这里需要知道RabbitMQ服务器的地址、端口号以及用户名密码。假设服务器地址为localhost，端口为默认的5672，用户名和密码都是guest。

示例代码：创建连接

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

public class RabbitMQConnection {
    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        
        // 设置连接参数
        factory.setHost("localhost"); // RabbitMQ服务器地址
        factory.setPort(5672);        // RabbitMQ服务器端口
        factory.setUsername("guest"); // 用户名
        factory.setPassword("guest"); // 密码
        
        // 建立连接
        try (Connection connection = factory.newConnection()) {
            System.out.println("成功连接到RabbitMQ服务器！");
        }
    }
}

这段代码中，ConnectionFactory负责创建连接，而newConnection()方法返回一个Connection对象。记住，try-with-resources语法会自动关闭连接资源，这是Java 7之后推荐的方式哦！

第三步：发送消息

发送消息就像是投递一个包裹到快递站，RabbitMQ会帮我们找到对应的收件人。

示例代码：发送消息

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

public class RabbitMQProducer {
    private static final String QUEUE_NAME = "test_queue";

    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        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!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("消息已发送：" + message);
        }
    }
}

关键步骤解析：

1. channel.queueDeclare()：声明一个名为test_queue的队列。如果队列已经存在，则不会重复创建。
2. channel.basicPublish()：向指定队列发送消息。第一个参数为空字符串表示交换机名称，默认使用默认交换机。
3. message.getBytes()：将消息转换为字节数组，因为RabbitMQ只支持字节数组形式的数据。

第四步：接收消息

接收消息就像是快递员从快递站取回包裹。这里我们用DeliverCallback定义消息处理逻辑。

示例代码：接收消息

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class RabbitMQConsumer {
    private static final String QUEUE_NAME = "test_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        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);

            // 定义消息处理回调
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println("收到消息：" + message);
            };

            // 开始监听消息
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        }
    }
}

关键步骤解析：

1. DeliverCallback：这是一个函数式接口，用于处理接收到的消息。
2. channel.basicConsume()：启动消息监听。第二个参数设置为true，表示自动确认消息已被消费。
3. consumerTag：可以用来标识消费者，但我们暂时忽略它。

第五步：实战案例

让我们结合生产和消费两个环节，完成一次完整的RabbitMQ交互。

案例：订单处理系统

假设有一个电商网站，用户下单后，我们需要通知库存服务和物流服务。我们可以设计如下流程：

1. 用户下单时，订单服务将消息发送到RabbitMQ。
2. 库存服务和物流服务分别订阅消息并处理各自的逻辑。

生产者代码（订单服务）

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

public class OrderService {
    private static final String EXCHANGE_NAME = "orders_exchange";

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

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");

            // 发送订单消息
            String orderMessage = "用户下单，订单号：12345";
            channel.basicPublish(EXCHANGE_NAME, "order.created", null, orderMessage.getBytes());
            System.out.println("订单消息已发送：" + orderMessage);
        }
    }
}

消费者代码（库存服务）

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class InventoryService {
    private static final String EXCHANGE_NAME = "orders_exchange";

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

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 绑定队列到交换机
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
            String queueName = channel.queueDeclare().getQueue();
            channel.queueBind(queueName, EXCHANGE_NAME, "order.created");

            // 定义消息处理回调
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println("库存服务收到消息：" + message);
                // 模拟库存扣减逻辑
                System.out.println("库存扣减中...");
            };

            // 开始监听消息
            channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
        }
    }
}

小结

通过以上几个步骤，我们完成了RabbitMQ的基本操作：连接、发送消息、接收消息以及实际应用场景。RabbitMQ的强大之处在于它能轻松应对高并发、分布式系统中的消息传递问题。

如果你还有疑问，欢迎随时提问！比如，“如何处理消息丢失？”、“如何设置消息优先级？”、“如何监控RabbitMQ性能？”等等。我会继续用轻松幽默的方式为你解答！

Tags：rabbitmq 默认密码