RabbitMQ 入门篇总结
微服务架构基础知识的总结:微服务架构基础知识的总结 微服务保护与分布式事务:微服务保护与分布式事务
在微服务架构中，服务与服务之间如何通信，是绕不开的话题。 常见方式有两种：同步调用 和 异步调用（消息队列）。
📚 目录（点击跳转对应章节）
一、同步调用 二、异步调用（消息队列） 三、RabbitMQ 基本介绍 四、SpringAMQP 接发消息步骤 五、消费者消息推送机制 六、Work 模型（工作队列模式） 七、交换机（Exchange） 八、交换机类型 九、声明队列和交换机 十、消息转换器 十一、改造黑马商城业务
一、同步调用
1. 同步调用流程（黑马商城）
在同步模式下：
- 用户发起请求（如下单）
- 订单服务调用库存服务
- 再调用支付服务
- 所有服务都处理完成后，才返回结果
本质上是：请求-响应模型（Request-Response）
2. 同步调用的优势
- 逻辑清晰
- 调用关系直观
- 便于理解和调试
- 适合实时强一致场景
3. 同步调用的问题
1）耦合度高
订单服务必须知道库存服务、支付服务的地址。 一旦某个服务变更，调用方也要修改。
2）性能差
假设：
- 调库存 200ms
- 调支付 300ms
总耗时至少 500ms。
所有调用是串行阻塞的。
3）稳定性差
只要一个服务挂了：
- 整个调用链全部失败
- 产生雪崩效应
二、异步调用（消息队列）
1. 异步结构
异步模式的核心思想：
发送者不关心接收者是否在线，也不等待处理结果。
引入一个中间组件 —— 消息队列（MQ）
结构变成：
- 订单服务 → 发送消息 → MQ
- 库存服务 → 从 MQ 获取消息
- 支付服务 → 从 MQ 获取消息
2. 异步调用流程（黑马商城）
下单流程：
订单服务不需要等待其他服务完成。
3. 异步调用的优势
1）解耦
发送方只管发消息，不关心谁处理。
2）提高性能
发送消息是毫秒级。 后续处理异步执行。
3）提高可用性
某个服务挂了：
- 消息仍然保存在队列中
- 服务恢复后继续消费
4）缓存消息，削峰填谷
消息队列可以缓冲高峰请求，平滑处理流量。 
4. 异步调用的问题
1）一致性问题
同步调用天然强一致。 异步模式需要设计补偿机制。
2）复杂度提高
需要引入：
- 消息可靠性
- 消息幂等性
- 死信队列
- 重试机制
由于以上问题，所以我们选择使用 RabbitMQ 作为消息队列，它提供了完善的机制来解决这些问题。
三、RabbitMQ 基本介绍
RabbitMQ 是：
- 基于 AMQP 协议
- 用 Erlang 开发
- 高可靠、高性能的消息中间件
核心组件包括：
- Producer（生产者）
- Consumer（消费者）
- Queue（队列）
- Exchange（交换机）
- Binding（绑定）
- RoutingKey（路由键）
四、SpringAMQP 接发消息步骤
在 Spring Boot 项目中使用 RabbitMQ，通常借助 SpringAMQP。
1. 引入依赖
添加依赖：
<!– AMQP依赖，包含RabbitMQ –>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2. 配置 RabbitMQ
在 application.yml 中配置：
spring:
rabbitmq:
host: localhost # 主机名称
port: 5672 # 端口
virtual-host: /hmall # 虚拟主机
username: guest # 用户名
password: guest # 密码
3. 使用 RabbitTemplate 发送消息
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendMessage() {
//队列名称
String queueName = "queueName";
//消息
String message = "hello rabbitmq";
//发送消息
rabbitTemplate.convertAndSend(queueName, message);
}
核心方法：
convertAndSend(交换机, routingKey, 消息)
4. 使用 @RabbitListener 监听消息
@RabbitListener(queues = "queueName")
public void listen(String message) {
System.out.println("接收到消息：" + message);
}
只要队列中有消息，就会自动触发该方法。
五、消费者消息推送机制
RabbitMQ 默认采用：
推模式（Push）
服务器主动将消息推送给消费者。
但为了避免消费者被压垮，引入：
- 预取机制（prefetch）
- 手动确认（ack）
这样可以控制一次最多处理多少条消息。
六、Work 模型（工作队列模式）
结构特点：
- 一个队列
- 多个消费者
消息会被平均分配给消费者。
适用于：
- 高并发任务处理
- 分布式任务执行
注意：
- 默认是轮询分发
- 可以设置公平分发（prefetch=1）
七、交换机（Exchange）
生产者其实不直接把消息发给队列。
真实流程是：
Producer → Exchange → Queue → Consumer
交换机负责：
八、交换机类型
1. Fanout（广播模式）
特点：
- 不需要 routingKey
- 所有绑定的队列都会收到消息
适用于：
- 日志广播
- 全局通知
2. Direct（精确匹配）
特点：
- routingKey 精确匹配
- 只发送到匹配的队列
适用于：
- 订单状态分类
- 精准路由
Direct和Fanout的区别
3. Topic（通配符匹配）
支持通配符：
- *：匹配一个单词
- #：匹配多个单词
例如：
order.create
order.update
order.delete
可以设计：
order.*
order.#
适用于：
- 复杂业务路由
- 多层级分类
Topic和Direct的区别
九、声明队列和交换机
可以使用：
- 管理界面
- 代码方式声明
Spring 中声明方式示例：
@Bean
public Queue queue() {
return new Queue("queueName");
}
@Bean
public DirectExchange exchange() {
return new DirectExchange("exchangeName");
}
@Bean
public Binding binding() {
return BindingBuilder.bind(queue())
.to(exchange())
.with("routingKey");
}
SpringAMQP还有注解的方式声明队列和交换机： 
十、消息转换器
默认情况下：
- 发送的是字节数组
- 接收也是字节数组
如果要发送对象，需要配置消息转换器。 
常用方式：
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
这样就可以直接发送对象：
rabbitTemplate.convertAndSend("exchange", "key", user);
消费者直接接收对象：
@RabbitListener(queues = "queueName")
public void listen(User user) {
System.out.println(user);
}
十一、改造黑马商城业务
前提条件
1.添加依赖
<!–消息发送–>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2.配置MQ地址
spring:
rabbitmq:
host: 192.168.150.101 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: /hmall # 虚拟主机
username: hmall # 用户名
password: 123 # 密码
接收信息
1.在trade-service服务中定义一个消息监听类： 
2.具体代码如下：
package com.hmall.trade.listener;
import com.hmall.trade.service.IOrderService;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RequiredArgsConstructor
public class PayStatusListener {
private final IOrderService orderService;
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "trade.pay.success.queue", durable = "true"),
exchange = @Exchange(name = "pay.direct"),
key = "pay.success"
))
public void listenPaySuccess(Long orderId){
orderService.markOrderPaySuccess(orderId);
}
}
发送消息
修改pay-service服务下的com.hmall.pay.service.impl.PayOrderServiceImpl类中的tryPayOrderByBalance方法：
private final RabbitTemplate rabbitTemplate;
@Override
@Transactional
public void tryPayOrderByBalance(PayOrderDTO payOrderDTO) {
// 1.查询支付单
PayOrder po = getById(payOrderDTO.getId());
// 2.判断状态
if(!PayStatus.WAIT_BUYER_PAY.equalsValue(po.getStatus())){
// 订单不是未支付，状态异常
throw new BizIllegalException("交易已支付或关闭！");
}
// 3.尝试扣减余额
userClient.deductMoney(payOrderDTO.getPw(), po.getAmount());
// 4.修改支付单状态
boolean success = markPayOrderSuccess(payOrderDTO.getId(), LocalDateTime.now());
if (!success) {
throw new BizIllegalException("交易已支付或关闭！");
}
// 5.修改订单状态
// tradeClient.markOrderPaySuccess(po.getBizOrderNo());
try {
rabbitTemplate.convertAndSend("pay.direct", "pay.success", po.getBizOrderNo());
} catch (Exception e) {
log.error("支付成功的消息发送失败，支付单id：{}， 交易单id：{}", po.getId(), po.getBizOrderNo(), e);
}
}
