深入浅出 RabbitMQ:简单队列实战指南

发布于:2025-08-05 ⋅ 阅读:(17) ⋅ 点赞:(0)
大家好,我是工藤学编程 🦉 一个正在努力学习的小博主,期待你的关注
实战代码系列最新文章😉 C++实现图书管理系统(Qt C++ GUI界面版)
SpringBoot实战系列🐷 【SpringBoot实战系列】SpringBoot3.X 整合 MinIO 存储原生方案
分库分表 分库分表之实战-sharding-JDBC分库分表执行流程原理剖析
消息队列 深入浅出 RabbitMQ-核心概念介绍与容器化部署

前情摘要:

1、深入浅出 RabbitMQ-核心概念介绍与容器化部署


【亲测宝藏】发现一个让 AI 学习秒变轻松的神站!不用啃高数、不用怕编程,高中生都能看懂的人工智能教程来啦!

👉点击跳转,和 thousands of 小伙伴一起用快乐学习法征服 AI,说不定下一个开发出爆款 AI 程序的就是你!


深入浅出 RabbitMQ:简单队列实战指南

一、什么是RabbitMQ简单队列

RabbitMQ作为主流的消息中间件,其核心功能是实现系统间的异步通信。简单队列(Simple Queue)是RabbitMQ中最基础的通信模式,采用"点对点"的消息传递方式:一个生产者发送消息到队列,一个消费者从队列中获取消息,适用于简单的异步通信场景。

二、环境准备

  1. RabbitMQ服务:参考深入浅出 RabbitMQ-核心概念介绍与容器化部署)

同时,为了方便环境之间的隔离(开发、测试、上线),我们需要按照如图添加一个虚拟环境

在这里插入图片描述

  1. 依赖配置:在Maven项目中添加AMQP客户端依赖
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.10.0</version>
</dependency>

三、实战代码实现

3.1 消息生产者(Send)

生产者负责创建连接、声明队列并发送消息,核心代码如下:

public class Send {
    // 队列名称(生产者和消费者需使用相同名称)
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        // 1. 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 配置连接参数
        factory.setHost("192.168.229.128");    // 服务器地址
        factory.setPort(5672);             // 端口(默认5672)
        factory.setUsername("admin");      // 用户名
        factory.setPassword("password");   // 密码
        factory.setVirtualHost("/dev");    // 虚拟主机(需提前创建)

        // 2. 创建连接和信道(使用try-with-resources自动关闭资源)
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 3. 声明队列(不存在则创建)
            /**
             * 参数说明:
             * 1. queue:队列名称
             * 2. durable:是否持久化(重启RabbitMQ后队列依然存在)
             * 3. exclusive:是否独占(仅当前连接可访问,连接关闭后自动删除)
             * 4. autoDelete:是否自动删除(当最后一个消费者断开后自动删除)
             * 5. arguments:额外参数(如队列长度限制等)
             */
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            // 4. 发送消息
            String message = "Hello World!";
            /**
             * 参数说明:
             * 1. exchange:交换机名称(使用默认交换机"")
             * 2. routingKey:路由键(默认交换机下需与队列名称一致)
             * 3. props:消息属性(如优先级、过期时间等)
             * 4. body:消息体(字节数组)
             */
            channel.basicPublish("", QUEUE_NAME, null, 
                               message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] 已发送消息: '" + message + "'");
        }
    }
}

3.2 消息消费者(Recv)

消费者负责监听队列并处理消息,核心代码如下:

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

    public static void main(String[] argv) throws Exception {
        // 1. 创建连接工厂(配置与生产者一致)
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.229.128");
        factory.setUsername("admin");
        factory.setPassword("password");
        factory.setVirtualHost("/dev");  // 注意:需与生产者使用相同虚拟主机
        factory.setPort(5672);

        // 2. 创建连接和信道(消费者通常保持长连接,不使用自动关闭)
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 3. 声明队列(与生产者保持一致,确保队列存在)
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] 等待接收消息中...(按CTRL+C退出)");

        // 4. 定义消息处理回调
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, 
                                      Envelope envelope, 
                                      AMQP.BasicProperties properties, 
                                      byte[] body) throws IOException {
                // 消费标识(可用于标识当前会话)
                System.out.println("消费标识: " + consumerTag);
                // 消息元数据(包含交换机、路由键、消息ID等)
                System.out.println("元数据: " + envelope);
                // 消息属性(如发送时间、优先级等)
                System.out.println("属性: " + properties);
                // 消息内容
                System.out.println("接收内容: " + new String(body, "UTF-8"));
            }
        };

        // 5. 开始消费消息
        /**
         * 参数说明:
         * 1. queue:队列名称
         * 2. autoAck:是否自动确认消息(消费成功后自动通知MQ删除消息)
         * 3. consumer:消息处理回调
         */
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

四、关键知识点解析

  1. 连接配置要点

    • 虚拟主机(VirtualHost):用于隔离不同项目的消息队列,生产者和消费者必须使用相同虚拟主机
    • 端口说明:5672是AMQP协议端口,15672是管理界面端口
  2. 队列声明参数

    • 持久化(durable):仅队列本身持久化,消息需额外设置持久化属性
    • 独占性(exclusive):通常用于临时队列,生产环境一般设为false
    • 自动删除(autoDelete):当最后一个消费者断开后自动清理队列,适合临时任务
  3. 消息确认机制

    • 代码中使用autoAck=true(自动确认):消息被接收后立即从队列删除
    • 生产环境建议使用autoAck=false(手动确认):确保消息被正确处理后再删除,避免消息丢失

五、测试步骤

  1. 启动RabbitMQ服务:rabbitmq-server start

  2. 访问管理界面(http://localhost:15672),确认虚拟主机/dev已创建

  3. 先运行消费者(Recv):启动后进入监听状态
    在这里插入图片描述

  4. 再运行生产者(Send):发送消息后控制台输出发送成功
    在这里插入图片描述

  5. 查看消费者控制台:成功接收并打印消息详情
    在这里插入图片描述

六、总结

通过本文实战,我们掌握了RabbitMQ简单队列的核心用法:

  • 生产者通过信道发送消息到队列
  • 消费者通过回调函数异步处理消息
  • 虚拟主机和队列参数的正确配置是通信基础

简单队列适合一对一的消息通信场景,下一篇我们将讲解工作队列(Work Queue),敬请关注!

觉得有用请点赞收藏!
如果有相关问题,欢迎评论区留言讨论~