Java 中间件之 RabbitMQ 快速入门: 简单队列模型快速入门

概述

用官方 HelloWorld 同款的简单队列模型，把「发一条 hello 到 RabbitMQ、收回来」这件事从头跑通，顺带把连接工厂、Channel、queueDeclare 这几个必写的 API 和控制台验证手段讲清楚。

纲要

• 简单队列模型：publisher / queue / consumer 三个角色，没有交换机参与
• AMQP 与 JMS：一个是线路协议，一个是 Java API 规范，两者不是一个层面的东西
• 工程结构：mq-demo 聚合工程，父工程管依赖，publisher 与 consumer 各是一个子模块
• 发送消息：ConnectionFactory → Connection → Channel → queueDeclare → basicPublish
• 接收消息：basicConsume 注册 DefaultConsumer 回调，handleDelivery 里处理消息体
• 异步机制：basicConsume 是非阻塞的，「等待接收消息」一定比「接收到消息」先打印
• 控制台验证：Queues 页看 Ready/Total，Get messages 偷看消息内容
• 实战坑位：端口 5672 与 15672、队列名大小写、声明参数不一致、消费者没起来导致堆积

简单队列模型

RabbitMQ 官方给了五种消息模型（也常被叫做五种队列模式），HelloWorld 是其中最简单的一种。它的结构里没有交换机：

三个角色的职责：

• publisher：消息发布者，把消息投递到队列，投完就走，不关心谁来处理
• queue：消息队列，负责接收并缓存消息，本质上是 RabbitMQ 里的一个存储结构，归属于某个 virtualHost
• consumer：订阅队列，拿到消息后按业务逻辑处理

注意模型图里 publisher 和 queue 直连只是逻辑视图。物理上 RabbitMQ 一定要经过交换机，HelloWorld 里用的是 RabbitMQ 预置的默认交换机（名字是空字符串 ""），它会按 routing key 的名字把消息路由到同名队列。所以 channel.basicPublish("", queueName, null, body) 第一个参数是空串，不是笔误。

AMQP 与 JMS

刚接触 MQ 很容易把这俩搞混，其实它们不在同一个层次：

对比维度	AMQP	JMS
本质	线路级协议（wire protocol），规定了数据包的字节格式	Java API 规范（JSR 914），只定义了一堆接口
归属	跨语言标准，与厂商无关	Java 平台专属规范
类比	类似于 HTTP	类似于 JDBC
实现	RabbitMQ、Apache Qpid、ActiveMQ 也部分支持	ActiveMQ、HornetQ、OpenMQ 要提供各自的驱动 jar
能否跨实现	只要两边都实现了 AMQP 就能互通	换 Broker 就得换客户端 jar，甚至改代码
消息模型	exchange / queue / binding，路由能力在 Broker 侧	Queue（点对点）/ Topic（发布订阅）两种域
消息结构	BasicProperties + byte[] body，属性写在协议头里	javax.jms.Message，分 header / properties / body

一句话记：AMQP 是怎么传，JMS 是怎么调。RabbitMQ 用的是 AMQP 0-9-1，Java 客户端 amqp-client 是对这套协议的 Java 实现；而 ActiveMQ 用的是 JMS 这套 Java 接口。如果不是纯 Java 的技术栈（比如有 Python、Go 的服务要参与收发），AMQP 这种语言无关的协议会省事很多。

顺带区分一下 spring-boot-starter-amqp 里的三个东西：

• AMQP：协议本身
• amqp-client：RabbitMQ 官方 Java 客户端，下面 HelloWorld 里的 Channel、ConnectionFactory 都出自这里
• SpringAMQP：Spring 对 AMQP 的再封装（RabbitTemplate、@RabbitListener），下一篇会用，这篇先用原生客户端，方便看清底层到底做了什么

工程结构

课前资料里的 mq-demo 是一个 Maven 聚合工程。父工程只做依赖管理，publisher 和 consumer 是两个平级子模块，各自一个 Spring Boot 工程：

mq-demo
├── pom.xml                       # 父工程，packaging=pom，继承 spring-boot-starter-parent
├── publisher                     # 消息发送方
│   ├── pom.xml
│   └── src
│       ├── main
│       │   ├── java/cn/itcast/mq/PublisherApplication.java
│       │   └── resources/application.yml
│       └── test/java/cn/itcast/mq/helloworld/PublisherTest.java   # 本文重点
└── consumer                      # 消息接收方
    ├── pom.xml
    └── src
        ├── main
        │   ├── java/cn/itcast/mq/ConsumerApplication.java
        │   └── resources/application.yml
        └── test/java/cn/itcast/mq/helloworld/ConsumerTest.java    # 本文重点

HelloWorld 这一节只用得到 helloworld 包下的两个类。application.yml 里配的是 SpringAMQP 用的连接地址，原生客户端根本不会去读它；publisher 测试目录里还有个 cn.itcast.mq.spring.SpringAmqpTest，那是后面 SpringAMQP 的用例，现在先不看。

父工程依赖

父工程继承 spring-boot-starter-parent，版本号统一管理，所以子模块的 pom 里不需要写 version：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>cn.itcast.demo</groupId>
    <artifactId>mq-demo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>pom</packaging>

    <modules>
        <module>publisher</module>
        <module>consumer</module>
    </modules>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.9.RELEASE</version>
        <relativePath/>
    </parent>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <!-- AMQP 依赖，内部传递了 com.rabbitmq:amqp-client -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <!-- 单元测试 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
    </dependencies>
</project>

子模块只需要声明父工程，没有自己的额外依赖：

<parent>
    <artifactId>mq-demo</artifactId>
    <groupId>cn.itcast.demo</groupId>
    <version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>publisher</artifactId>

这里有个版本相关的坑：spring-boot-starter-test 在 2.3.x 里默认带着 JUnit 5 的 vintage 引擎，所以源码里的 org.junit.Test（JUnit 4 注解）能正常跑。从 Spring Boot 2.4 开始 vintage 被移除了，如果你的项目是 2.4 及以上，要么把 import org.junit.Test 换成 import org.junit.jupiter.api.Test，要么手动加依赖：

<dependency>
    <groupId>org.junit.vintage</groupId>
    <artifactId>junit-vintage-engine</artifactId>
    <scope>test</scope>
</dependency>

消息发送 publisher

发送流程是五步：建立连接 → 创建 Channel → 声明队列 → 发送消息 → 关闭通道和连接。

package cn.itcast.mq.helloworld;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

public class PublisherTest {

    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1. 建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1. 设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2. 建立连接
        Connection connection = factory.newConnection();

        // 2. 创建通道 Channel
        Channel channel = connection.createChannel();

        // 3. 声明队列（队列不存在就创建，存在就复用）
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4. 发送消息：交换机名传空串表示用默认交换机
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
        System.out.println("发送消息成功：【" + message + "】");

        // 5. 关闭通道和连接
        channel.close();
        connection.close();
    }
}

源码里队列名是 simple.queue、消息体是 hello, rabbitmq!（带逗号和感叹号），比对着看的时候别被视频口播里的 simpleq、helloRabbitMQ 带偏，以源码为准。另外源码用的是 message.getBytes()（平台默认字符集），这里改成了显式的 UTF-8，理由见后面的坑位小节。

几个关键点。

为什么要 Channel 而不是直接用 Connection 发消息？ Connection 是一条真实的 TCP 长连接，建连成本高（握手、认证），而且 TCP 连接数是有上限的。Channel 是复用在一条 Connection 之上的逻辑链路，声明队列、收发消息这些操作全部走 Channel，多个线程各开各的 Channel 互不干扰。生产环境下的典型做法是：一个应用一条 Connection，每个线程一个 Channel。

queueDeclare 的五个参数（后两个常被忽略但很关键）：

参数	值	含义
queue	simple.queue	队列名称，最长 255 字节的 UTF-8 字符
durable	false	是否持久化，false 表示 RabbitMQ 重启后队列消失
exclusive	false	是否排他，true 表示只对首次声明它的 Connection 可见，连接断开即删除
autoDelete	false	是否自动删除，true 表示最后一个消费者断开后队列自动删除
arguments	null	扩展参数，比如设置队列最大长度、死信交换机等

为什么 producer 和 consumer 都要声明队列？ 两边谁先启动是不确定的。消费者先起来时去找队列，队列还没有就直接报错了。所以两边各自声明一次，成为一种保险措施。重复声明同一个队列不会覆盖也不会新建——只要队列已经存在且参数一致，RabbitMQ 会直接复用。

消息接收 consumer

接收流程和发送的前三步完全一致，区别只有最后一步：不是主动拉，而是注册一个回调函数，把处理逻辑挂到队列上。

package cn.itcast.mq.helloworld;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1. 设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2. 建立连接
        Connection connection = factory.newConnection();

        // 2. 创建通道 Channel
        Channel channel = connection.createChannel();

        // 3. 声明队列（保证队列存在，与 publisher 侧参数必须一致）
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4. 订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5. 处理消息
                String message = new String(body, StandardCharsets.UTF_8);
                System.out.println("接收到消息：【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }
}

源码里 ConsumerTest 不是 JUnit 测试类，而是带 main 方法的普通类，得用 Run 'ConsumerTest.main()' 启动。原因也很实际：测试方法的线程跑完 JUnit 就直接退出了，回调函数根本来不及执行。

basicConsume 的第二个参数 autoAck = true 表示自动确认：消息一投递给消费者，RabbitMQ 就认为这条消息处理成功并立即删除，也就是所谓的「阅后即焚」。业务处理中抛了异常，消息也照样消失——换成手动 ACK 就是为了治这个，那是后话。

异步在哪儿

跑 consumer 的时候会看到这个输出顺序：

等待接收消息。。。。
接收到消息：【hello, rabbitmq!】

basicConsume 写在前面，它的回调却后打印。原因不是执行顺序乱了，是这俩根本不在同一个线程上：

basicConsume 返回得非常快，它只是把回调函数登记到队列上。主线程继续往下走，打印「等待接收消息」；等 RabbitMQ 把消息投递过来，AMQP 客户端的工作线程才回调 handleDelivery。这就是消费侧的异步——生产者发完就断开连接，消费者拿到了才处理，两边互不等待。

完整操作步骤

启动 RabbitMQ

本地裸机或者用 Docker，这里给 Docker 的起法，顺便把用户名密码和后面代码里的保持一致：

docker run -d --name rabbitmq \
  -p 5672:5672 \
  -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=itcast \
  -e RABBITMQ_DEFAULT_PASS=123321 \
  rabbitmq:3.8-management

带 management 后缀的镜像自带 Web 管理插件。启动后访问 http://<你的IP>:15672，用 itcast / 123321 登录。

确认账号在 vhost 上的权限

进 Admin → Users，确认 itcast 存在且 Can access virtual hosts 里有 /。如果列表里没有这个用户，就在这一步补：

# 进入容器执行
docker exec -it rabbitmq rabbitmqctl add_user itcast 123321
# 不加 tag 的用户登录不了 15672 管理台，必须补这一行
docker exec -it rabbitmq rabbitmqctl set_user_tags itcast administrator
docker exec -it rabbitmq rabbitmqctl set_permissions -p / itcast ".*" ".*" ".*"

三对 ".*" 依次是配置权限、写权限、读权限。给了权限才能在 / 这个 vhost 里建队列、发消息。

修改连接参数

把 PublisherTest 和 ConsumerTest 里的 factory.setHost("192.168.150.101") 换成你自己的 IP 或主机名。用 localhost 是不行的——除非 RabbitMQ 就跑在本机。

运行 publisher 发送消息

跑 PublisherTest#testSendMessage。控制台输出：

发送消息成功：【hello, rabbitmq!】

在控制台查看消息

打开 Queues 页，能看到 simple.queue 这一条，Ready 是 1（Total 也是 1）。点进队列名，往下翻到 Get messages，Ack Mode 保持默认的 Nack message requeue true，Messages 填 1，点 Get Message：

hello, rabbitmq!

这一步是「偷看」——消息拿了又退回去，Ready 仍然是 1。如果 Ack Mode 选了 automatic ack，取一次消息就没了。

运行 consumer 消费消息

保持控制台页面不动，用 Run 'ConsumerTest.main()' 启动消费者。切回 RabbitMQ 控制台刷新，Ready 从 1 变成 0：autoAck 模式下消息一旦投递，队列立刻把它删掉。

想验证「消费者不在线，消息会攒着」的话，先把 consumer 停掉，连着跑三遍 PublisherTest，控制台会显示 Ready = 3；再启动 consumer，三条消息一次性消费完，Ready 归零。

实战坑位

现象 / 报错	原因	处理
ConnectException: Connection refused	端口写成了 15672	5672 是 AMQP 通信端口，15672 是管理台 UI 端口，代码里必须是 5672
ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN	用户名密码错，或用户在 / 这个 vhost 上没配权限	Admin 页确认用户，或 rabbitmqctl set_permissions -p / itcast ".*" ".*" ".*"
页面两边看到的队列不是一个	队列名区分大小写	simple.queue 和 Simple.Queue 是两个独立队列，代码里写成常量或配置项对着抄
PRECONDITION_FAILED - inequivalent arg 'durable' for queue	两边 queueDeclare 参数不一致	producer/consumer 用完全相同的一组参数，改了一个就得改另一个
消息不报错但队列里查不到	队列还没创建就发了，默认交换机找不到匹配队列就丢掉	养成发布前先 queueDeclare 的习惯；或者配 publisher confirm 监听不可路由消息
控制台 Get Message 之后消息没了	Ack Mode 选成了 automatic ack	想偷看就选 Nack message requeue true
中文消息变问号、乱码	getBytes() / new String() 用的是平台默认字符集	两边都显式写 StandardCharsets.UTF_8
浏览器 Connections 里连接数对不上	publisher 是测试方法，跑完就断连；consumer 的 main 一直挂着	正常现象，consumer 需要手动 stop 才会断开
业务代码抛异常但消息没了	autoAck = true，投递即确认	改成手动 ACK，在业务成功后才 basicAck

还有一个容易混的地方：basicPublish 第一个参数是交换机名而不是队列名，写成 channel.basicPublish(queueName, queueName, ...) 会去找一个叫 simple.queue 的交换机，找不到就直接报 NOT_FOUND - no exchange，这和队列没创建时的静默丢弃是两种不同的表现。

API 速览

原生客户端这一节用到的几个方法：

方法	作用	补充说明
factory.newConnection()	建立 TCP 连接	还有个重载能从 URI 读全部参数：factory.setUri("amqp://itcast:123321@host:5672/")
connection.createChannel()	创建逻辑通道	大部分 AMQP 操作都在 Channel 上，务必用完 close()
channel.queueDeclare(...)	声明队列	幂等，参数不一致才报错
channel.basicPublish(exchange, routingKey, props, body)	发送消息	body 必须是 byte[]，props 可用于设持久化 MessageProperties.PERSISTENT_TEXT_PLAIN
channel.basicConsume(queue, autoAck, callback)	订阅消息	非阻塞，注册完立即返回
DefaultConsumer#handleDelivery(...)	回调方法	body 是消息体字节，envelope 里有 deliveryTag、交换机、routing key
channel.basicAck(deliveryTag, multiple)	手动确认	autoAck 为 false 时必须调，否则消息一直是 Unacked 状态

官方文档

• RabbitMQ Tutorials - Hello World!（Java）
• RabbitMQ Java Client API Guide
• AMQP 0-9-1 协议规范
• Spring AMQP 参考文档

总结

• HelloWorld 是唯一一个不显式使用交换机的模型，走的是默认交换机，靠 routing key 与队列同名完成路由
• Connection 是物理 TCP 连接，Channel 是复用其上的逻辑链路，所有读写操作都在 Channel 上进行
• queueDeclare 是幂等的，生产者和消费者都声明一遍，纯粹是为了规避启动顺序不确定带来的队列缺失
• basicConsume 注册回调后立即返回，handleDelivery 在另一个线程里执行，这是整个模型异步的本质
• autoAck = true 时消息投递即删除，下一篇换成 SpringAMQP 时会一并处理确认机制
• publisher 和 consumer 里重复的连接参数配置很容易写歪，下一步就该交给 SpringAMQP 统一托管了

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/Tyro_java/article/details/167258813