Hello World：用Java构建你的第一个RabbitMQ消息队列程序

在现代分布式系统架构中，消息队列扮演着至关重要的角色。作为最流行的开源消息代理软件之一，RabbitMQ以其可靠性、灵活性和易用性赢得了广大开发者的青睐。本文将带你一步步使用Java编写第一个RabbitMQ程序，让你在30分钟内掌握消息队列的基本概念和实战技巧。

什么是RabbitMQ？

RabbitMQ是一个实现了高级消息队列协议（AMQP）的开源消息代理软件。它就像一个高效的邮局系统，负责接收、存储和转发消息，使得应用程序之间的通信变得异步、解耦和可靠。

核心概念：

Producer（生产者）：发送消息的应用程序
Consumer（消费者）：接收消息的应用程序
Queue（队列）：存储消息的缓冲区
Exchange（交换机）：接收生产者发送的消息，并根据规则将消息路由到队列

环境准备

在开始编写代码之前，我们需要完成以下准备工作：

1. 安装RabbitMQ

使用Docker安装（推荐）：

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

在Ubuntu上安装：

sudo apt-get update
sudo apt-get install rabbitmq-server
sudo systemctl enable rabbitmq-server
sudo systemctl start rabbitmq-server

安装完成后，访问 http://localhost:15672 可以打开RabbitMQ的管理界面（默认用户名/密码：guest/guest）。

2. 添加Maven依赖

在项目的pom.xml中添加RabbitMQ客户端依赖：

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

编写第一个RabbitMQ程序

现在让我们开始编写代码，创建一个简单的”Hello World”示例，包含一个生产者和一个消费者。

步骤1：创建连接工具类

首先，我们创建一个用于建立RabbitMQ连接的工具类：

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

public class RabbitMQConnection {
    
    private static final String HOST = "localhost";
    private static final int PORT = 5672;
    private static final String USERNAME = "guest";
    private static final String PASSWORD = "guest";
    
    public static Connection getConnection() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);
        factory.setPort(PORT);
        factory.setUsername(USERNAME);
        factory.setPassword(PASSWORD);
        
        return factory.newConnection();
    }
}

步骤2：实现消息生产者

生产者负责创建并发送消息到RabbitMQ：

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

public class Producer {
    
    private final static String QUEUE_NAME = "hello_queue";
    
    public static void main(String[] args) {
        // 使用try-with-resources确保资源正确关闭
        try (Connection connection = RabbitMQConnection.getConnection();
             Channel channel = connection.createChannel()) {
            
            // 声明一个队列，如果不存在则创建
            // 参数说明：queue - 队列名称
            //         durable - 是否持久化
            //         exclusive - 是否排他（仅当前连接可用）
            //         autoDelete - 是否自动删除（无消费者时自动删除）
            //         arguments - 其他参数
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            
            String message = "Hello RabbitMQ! 当前时间：" + System.currentTimeMillis();
            
            // 发布消息
            // 参数说明：exchange - 交换机名称（空字符串表示默认交换机）
            //         routingKey - 路由键（对于默认交换机，路由键就是队列名称）
            //         props - 消息属性
            //         body - 消息体
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            
            System.out.println(" [x] 发送消息: '" + message + "'");
            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

步骤3：实现消息消费者

消费者从队列中接收并处理消息：

import com.rabbitmq.client.*;
import java.io.IOException;

public class Consumer {
    
    private final static String QUEUE_NAME = "hello_queue";
    
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();
        
        // 声明队列，确保队列存在
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        System.out.println(" [*] 等待接收消息。按 CTRL+C 退出");
        
        // 创建消费者回调
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] 收到消息: '" + message + "'");
            
            // 模拟消息处理
            processMessage(message);
        };
        
        // 开始消费消息
        // 参数说明：queue - 队列名称
        //         autoAck - 是否自动确认（true：自动确认，false：手动确认）
        //         deliverCallback - 消息接收回调
        //         cancelCallback - 取消订阅回调
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
    }
    
    private static void processMessage(String message) {
        try {
            // 模拟消息处理时间
            Thread.sleep(1000);
            System.out.println(" [√] 消息处理完成: " + message);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

运行和测试

现在让我们测试我们的RabbitMQ程序：

首先启动消费者：
```
java Consumer
```
你会看到输出：[*] 等待接收消息。按 CTRL+C 退出
然后运行生产者：
```
java Producer
```
生产者输出：[x] 发送消息: 'Hello RabbitMQ! 当前时间：1641234567890'
观察消费者控制台：消费者会显示：[x] 收到消息: 'Hello RabbitMQ! 当前时间：1641234567890' 然后：[√] 消息处理完成: Hello RabbitMQ! 当前时间：1641234567890

进阶功能：手动消息确认

在实际生产环境中，我们通常使用手动消息确认来确保消息被正确处理：

public class ManualAckConsumer {
    
    private final static String QUEUE_NAME = "hello_queue";
    
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();
        
        // 设置预取计数为1，确保公平分发
        channel.basicQos(1);
        
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        System.out.println(" [*] 等待接收消息（手动确认模式）。按 CTRL+C 退出");
        
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            
            try {
                System.out.println(" [x] 开始处理消息: '" + message + "'");
                processMessage(message);
                
                // 手动确认消息
                // 参数说明：deliveryTag - 消息标签
                //         multiple - 是否批量确认（确认所有比该标签小的消息）
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                System.out.println(" [√] 消息处理完成并确认: " + message);
                
            } catch (Exception e) {
                System.err.println(" [×] 消息处理失败: " + message);
                e.printStackTrace();
                
                // 拒绝消息并重新入队
                // 参数说明：deliveryTag - 消息标签
                //         multiple - 是否批量拒绝
                //         requeue - 是否重新入队
                channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
            }
        };
        
        // 关闭自动确认，使用手动确认
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }
    
    private static void processMessage(String message) throws InterruptedException {
        // 模拟业务处理
        Thread.sleep(2000);
        
        // 模拟随机失败（仅用于演示）
        if (Math.random() < 0.2) {
            throw new RuntimeException("模拟处理失败");
        }
    }
}

实际应用场景

RabbitMQ在实际项目中有广泛的应用：

1. 订单处理系统

// 订单创建后发送消息到队列
public class OrderService {
    public void createOrder(Order order) {
        // 保存订单到数据库
        orderRepository.save(order);
        
        // 发送消息到订单队列
        String message = "ORDER_CREATED:" + order.getId();
        rabbitTemplate.convertAndSend("order_exchange", "order.created", message);
    }
}

2. 邮件发送服务

// 异步发送邮件，提高系统响应速度
public class EmailConsumer {
    @RabbitListener(queues = "email_queue")
    public void processEmail(EmailRequest request) {
        try {
            emailService.send(request);
            // 记录发送日志
            log.info("邮件发送成功: {}", request.getTo());
        } catch (Exception e) {
            log.error("邮件发送失败: {}", request.getTo(), e);
            // 将失败消息转移到死信队列
            throw new AmqpRejectAndDontRequeueException(e);
        }
    }
}

常见问题与解决方案

1. 连接失败

问题：java.net.ConnectException: Connection refused 解决：确保RabbitMQ服务正在运行，检查主机名和端口配置。

2. 队列已存在但属性不匹配

问题：inequivalent arg 'durable' for queue 'hello_queue' 解决：删除已存在的队列或确保声明参数一致。

3. 消息堆积

解决：增加消费者数量或使用消息TTL（Time-To-Live）：

Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 60000); // 60秒后过期
channel.queueDeclare(QUEUE_NAME, true, false, false, args);

总结

通过本文的学习，你已经掌握了：

RabbitMQ的基本概念和核心组件
如何使用Java创建RabbitMQ生产者和消费者
消息确认机制的重要性
RabbitMQ在实际项目中的应用场景

这只是RabbitMQ世界的入门，还有更多高级特性等待探索，如交换机类型、死信队列、集群配置等。建议你在掌握基础后，继续深入学习这些高级功能，让RabbitMQ成为你构建可靠分布式系统的得力工具。

下一步学习建议：

了解不同类型的交换机（Direct、Fanout、Topic、Headers）
学习消息持久化和事务
探索RabbitMQ集群和高可用配置
研究消息确认模式和QoS设置

Happy coding！愿你在消息队列的世界里越走越远！

文档信息

本文作者：JiliangLee
本文链接：https://leejiliang.cn/2025/10/22/Hello-World%E7%94%A8-Java-%E5%86%99%E4%BD%A0%E7%9A%84%E7%AC%AC%E4%B8%80%E4%B8%AA-RabbitMQ-%E7%A8%8B%E5%BA%8F/
版权声明：自由转载-非商用-非衍生-保持署名（创意共享3.0许可证）

CODE

Hello World：用Java构建你的第一个RabbitMQ消息队列程序

Hello World：用Java构建你的第一个RabbitMQ消息队列程序

什么是RabbitMQ？

环境准备

1. 安装RabbitMQ

2. 添加Maven依赖

编写第一个RabbitMQ程序

步骤1：创建连接工具类

步骤2：实现消息生产者

步骤3：实现消息消费者

运行和测试

进阶功能：手动消息确认

实际应用场景

1. 订单处理系统

2. 邮件发送服务

常见问题与解决方案

1. 连接失败

2. 队列已存在但属性不匹配

3. 消息堆积

总结

文档信息

Search

Table of Contents