专栏RabbitMQ 进阶之路个人主页手握风云目录一、核心概念与工作流程1.1. 核心组件1.2. 标准工作流程1.3. AMQP 协议1.4. 管理后台导航栏二、入门案例2.1. 引入依赖2.2. 通用前置步骤2.3. 生产者核心方法参数1. 队列声明queueDeclare2. 消息发送basicPublish2.4. 生产者代码逻辑2.5. 消费者核心方法参数1. 消息消费basicConsume2. 回调方法handleDelivery2.6. 消费者代码逻辑一、核心概念与工作流程RabbitMQ 是开源消息中间件基于生产者 - 消费者模型负责接收、存储、转发消息整体逻辑可类比邮局 / 物流公司。1.1. 核心组件Producer 生产者作为 RabbitMQ 服务的客户端主要作用是生成业务消息并发送至 RabbitMQ 服务消息会携带路由标签供 MQ 完成路由分发生产者本身不需要知晓最终消费这条消息的客户端信息。Consumer 消费者也同样属于 RabbitMQ 服务的客户端消费者会主动连接服务端订阅指定队列并读取、处理消息在消费消息的过程中消息自带的路由标签会被丢弃消费者无法追溯消息对应的生产者。BrokerBroker 指代 RabbitMQ Server也就是 RabbitMQ 服务节点或服务实例通常可以理解为一台 RabbitMQ 服务器它承担着统一接收、缓存、转发所有客户端消息的核心职责。Connection 连接Connection 是客户端与 RabbitMQ 服务之间建立的底层 TCP 连接是两端通信的基础客户端和服务端之间所有的数据、控制指令都依靠这条 TCP 连接进行传输。Channel 信道 / 通道Channel 名为信道或通道构建在 TCP Connection 之上属于独立的虚拟连接单条 TCP 连接能够创建多个 Channel所有消息收发操作都依托 Channel 完成借助信道复用 TCP 链路可以大幅减少频繁创建、销毁 TCP 连接带来的性能损耗提升整体吞吐量。Virtual host 虚拟主机Virtual host 是虚拟主机属于逻辑隔离单元类似于 MySQL 中的 database一台 Broker 服务器中能够创建多个虚拟主机不同业务、不同用户可以分配独立的 vhost每个虚拟主机内部的交换机、队列相互隔离不会产生数据干扰。Queue 队列Queue 是队列是 RabbitMQ 内部用于缓存、存储消息的对象消息会在队列中排队等待消费者处理同时支持多个消费者同时订阅同一个队列创建队列的 queueDeclare 方法包含五个核心参数queue 代表队列名称不存在则自动创建、已存在则忽略durable 控制队列是否持久化设为 true 时队列会落盘保存MQ 重启后队列不会丢失exclusive 代表队列是否独占仅允许单个消费者监听对应的 TCP 连接关闭后队列会自动删除autoDelete 控制队列自动删除逻辑当队列没有任何消费者订阅时便会销毁arguments 用于存放队列自定义扩展配置参数。Exchange 交换机Exchange 是交换机是消息抵达 Broker 后的第一个处理节点承担消息路由的核心工作它接收生产者发送的消息结合自身类型与消息携带的路由标签将消息分发转发至一个或多个队列可以类比快递分拣中心根据包裹标签完成分流RabbitMQ 启动时会预创建多类内置交换机简单收发场景会使用名称为空字符串 的直连默认交换机使用该交换机时路由 key 必须和队列名称完全匹配消息才能正常路由到目标队列。1.2. 标准工作流程生产者首先生成携带业务数据与路由标签的消息随后生产者连接 RabbitMQ 服务端 Broker建立 TCP 连接并创建信道 Channel接着生产者完成交换机 Exchange 的声明以此定义消息对应的路由规则之后生产者再声明用于临时存储消息的队列 Queue全部准备工作完成后生产者通过信道将消息发送至 RabbitMQ 服务端 BrokerBroker接收到消息后会依据预设的路由规则把消息存入匹配的队列当中如果不存在能够匹配的目标队列 Broker会按照生产者预先配置的策略选择将消息直接丢弃或是退回给生产者。如果把整套 RabbitMQ 服务类比物流公司其中 Broker 对应物流公司总部负责统筹所有包裹的收发与存储工作Virtual Host 则是物流公司划分出的独立业务运营中心用来实现不同业务之间的数据隔离Exchange 相当于站点里的快递分拣中心会根据包裹附带的标签完成分流分发Queue 就是存放待派送包裹的货物仓库Connection 代表客户端与服务端之间的主干通信线路Channel 是主干线路上多条并行的子线路能够同步处理多条消息提升处理效率。1.3. AMQP 协议AMQP的全称为高级消息队列协议该协议标准化了消息交换相关规则统一规定了交换机、队列等各类实体之间的交互规范RabbitMQ是基于Erlang语言对AMQP协议实现的开源消息中间件除此之外它还额外兼容STOMP、MQTT等其他协议AMQP协议定义的核心实体包含生产者、交换机、队列、消费者这些实体和RabbitMQ里的核心组件能够一一对应。1.4. 管理后台导航栏Overview全局总览视图展示节点、连接、消息总量等全局数据Connections管理所有客户端 TCP 连接Channels管理所有信道Exchanges交换机管理Queues队列管理Admin用户、虚拟主机、权限管理。二、入门案例基于 Maven 项目实现简单模式消息收发分为生产者、消费者两部分。2.1. 引入依赖dependency groupIdcom.rabbitmq/groupId artifactIdamqp-client/artifactId version5.20.0/version scopecompile/scope /dependency2.2. 通用前置步骤生产者、消费者前 5 步逻辑一致创建 ConnectionFactory 连接工厂配置连接参数服务 IP、端口、虚拟主机、用户名、密码通过工厂创建 ConnectionTCP 连接基于连接创建 Channel通道调用 queueDeclare 声明队列队列不存在则自动创建存在则跳过。2.3. 生产者核心方法参数1. 队列声明queueDeclarequeueDeclare(String var1, boolean var2, boolean var3, boolean var4, MapString, Object var5)var1队列名称var2队列是否持久化true 服务重启后队列不丢失var3是否独占队列仅单个消费者监听连接关闭则删除队列var4无消费者时队列是否自动删除var5队列扩展配置参数。2. 消息发送basicPublishvoid basicPublish(String var1, String var2, AMQP.BasicProperties var3, byte[] var4)var1交换机名称简单模式使用默认空交换机 var2路由键使用默认交换机时必须和队列名一致var3消息配置属性var4消息正文字节数组。2.4. 生产者代码逻辑完成连接、通道、队列声明调用 basicPublish 向默认交换机发送消息消息发送完成后主动关闭 Channel 和 Connection释放资源。完整代码实现package rabbitmq; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ProducerDemo { public static void main(String[] args) throws IOException, TimeoutException { // 1. 建立连接 ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(8.152.201.3); connectionFactory.setPort(5672); connectionFactory.setUsername(yang); connectionFactory.setPassword(study); connectionFactory.setVirtualHost(test); Connection connection connectionFactory.newConnection(); // 2. 开启信道 Channel channel connection.createChannel(); // 3. 声明交换机使用内置的交换机 // 4. 声明队列 channel.queueDeclare(hello, true, false, false, null); // 5. 发送消息 for (int i 0; i 10; i) { String message Hello RabbitMQ i; channel.basicPublish(, hello, null, message.getBytes()); } System.out.println(消息发送成功); // 6. 释放资源 channel.close(); connection.close(); } }2.5. 消费者核心方法参数1. 消息消费basicConsumeString basicConsume(String var1, boolean var2, Consumer var3)var1目标队列名var2是否自动消息确认消费者收到消息后自动告知 MQvar3消息回调对象使用框架提供的 DefaultConsumer。2. 回调方法handleDeliverypublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { }消费者接收消息后自动触发用于业务处理consumerTag消费者标识envelope消息封装信息交换机、路由键等properties消息配置body消息内容字节数组。2.6. 消费者代码逻辑完成连接、通道、队列声明自定义 DefaultConsumer重写 handleDelivery 方法打印 / 处理消息调用 basicConsume 监听队列消费者属于常驻监听程序一般不主动关闭连接与通道。完整代码实现package rabbitmq; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ConsumerDemo { public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { // 1. 建立连接 ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(公网 IP); connectionFactory.setPort(5672); connectionFactory.setUsername(yang); connectionFactory.setPassword(study); connectionFactory.setVirtualHost(test); Connection connection connectionFactory.newConnection(); // 2. 常见信道 Channel channel connection.createChannel(); // 3. 声明队列可以省略 channel.queueDeclare(hello, true, false, false, null); // 4. 消费信息 DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(接收信息 new String(body)); } }; channel.basicConsume(hello, true, consumer); Thread.sleep(5000); // 5. 释放资源 channel.close(); connection.close(); } }先运行生产者代码可以在管理界面看到可查看到新建队列、临时连接 / 通道队列内存在待消费消息。再运行消费者代码控制台成功打印接收的消息队列消息被消费。如果注释资源关闭逻辑连接和通道会在后台持续运行。