第二篇：消息为什么能够解耦系统？

消息为什么能够解耦系统？

上一篇我们聊了《为什么需要消息队列？》。

我们看到，消息队列出现的原因，并不是因为系统里面缺少一个“传消息的工具”，而是因为随着业务增长，服务之间的直接调用会让系统越来越复杂。

最开始：

订单服务
↓
库存服务

没有问题，但是随着业务增加：

订单服务

├──库存服务
├──优惠券服务
├──积分服务
├──短信服务
├──物流服务
└──营销服务

订单服务逐渐变成了整个业务流程的中心，任何一个下游服务变化，都会影响订单服务，这就是系统耦合。

那么问题来了：为什么引入一个消息队列，就可以解决服务之间的耦合问题？

直接调用的问题是什么？先看最常见的代码。

订单创建：

@Service
public class OrderService {

private final StockService stockService;
private final CouponService couponService;
private final SmsService smsService;

public void createOrder(Order order) {

// 保存订单
orderRepository.save(order);

// 扣库存
stockService.reduce(order);

// 发优惠券
couponService.send(order)

// 发短信
smsService.send(order);
}
}

这段代码的问题，不是写法不好，实际上很多系统最开始都会这么设计。

真正的问题是：订单服务和其他服务形成了强依赖。

订单服务必须知道：

库存服务存在；
优惠券服务存在；
短信服务存在。

如果以后新增：

邮件通知；
用户画像；
数据分析；

那么订单服务必须继续修改。

代码变成：

public void createOrder(Order order) {

save(order);

stockService.reduce(order);

couponService.send(order);

smsService.send(order);

mailService.send(order);

analysisService.record(order);
}

业务越来越多，核心服务越来越臃肿。面向接口为什么不能解决？

很多开发者会想到：既然耦合严重，那抽一个接口。

例如：

public interface OrderListener {

void handle(Order order);

}

然后：

@Component
public class SmsListener implements OrderListener {

public void handle(Order order){

smsService.send(order);

}
}

订单服务：

public class OrderService {

private List<OrderListener> listeners;

public void createOrder(Order order){

save(order);

listeners.forEach(
listener -> listener.handle(order)
);

}
}

看起来优雅很多，但是本质没有改变。因为订单服务仍然负责调用这些监听器。

如果监听器执行失败：

订单创建
↓
调用监听器
↓
某个监听失败

整个流程依然受到影响，问题不是代码结构。

而是：调用关系本身存在依赖。

消息队列改变了什么？

消息队列做了一件非常重要的事情：把服务之间的直接调用，变成事件通知。

以前：

订单服务
↓
库存服务
↓
短信服务

调用关系：我调用你

现在：

订单服务
↓
订单创建消息
↓
消息队列
↓
库存服务
短信服务
积分服务

变成：我告诉你发生了什么。

订单服务只产生一个事件：
{
“event”:“ORDER_CREATED”,
“orderId”:10001
}

它不知道谁消费。也不关心谁消费。

代码会变成什么样？

生产者：

@Service
public class OrderService {

private final MessageProducer producer;

public void createOrder(Order order){

orderRepository.save(order);

producer.send(
"order-created", order
);
}
}

消费者：

库存：

@Component
public class StockConsumer {

public void consume(Order order){

stockService.reduce(order);

}
}

短信：

@Component
public class SmsConsumer {

public void consume(Order order){

smsService.send(order);

}
}

现在：

订单服务不知道：

有没有短信服务；
有没有库存服务；
有没有积分服务。

它只知道：

订单创建完成，需要发布一个事件。但是，为什么 MQ 可以做到这一点？关键就在于 MQ 中间增加了一个角色：

Broker。

很多人理解消息队列：

生产者
↓
消费者

其实不准确。

真实结构：

Producer
↓
Broker
↓
Consumer

Broker 是消息的存储和转发中心。

它负责：

接收消息；
保存消息；
等待消费者读取。
没有 Broker 会怎么样？

假设订单服务直接通知库存服务：

订单服务
↓
库存服务

那么订单服务必须关心：库存服务是否在线。如果库存服务宕机，结果订单创建失败

但是有 Broker：

订单服务
↓
Broker
↓
库存服务

库存服务暂时不可用：

订单服务
↓
消息保存成功
↓
库存服务恢复
↓
继续消费

两个系统的生命周期被分离，这才是真正的解耦。

发布订阅模型为什么重要？消息队列还有一个关键能力：一个消息，可以被多个系统消费。

例如：订单创建事件：

ORDER_CREATED

可能有：

库存系统：扣减库存

积分系统：增加积分

营销系统：发送优惠

三个系统都需要这个消息。

如果使用传统调用，订单服务需要：

stockService.reduce();

pointService.add();

marketingService.push();

但是使用 MQ：

库存消费者
↑
|

订单事件 → Broker
|
↓
积分消费者
|
↓
营销消费者

新增消费者：

只增加一个消费端，生产者不需要修改。

Kafka 和 RocketMQ 是怎么实现这种解耦的？

不同 MQ 实现方式不同，但核心思想类似。

Kafka

Kafka 中：

Topic
↓
Partition
↓
Consumer Group

一个 Topic 保存一类消息。消费者通过 Consumer Group 消费。不同 Group 可以独立消费同一份消息。

例如：

订单 Topic：order-created

库存：stock-group

积分：point-group

两个 Group 都可以读取：
order-created
RocketMQ

RocketMQ：

Topic
↓
MessageQueue
↓
Consumer

同样通过 Topic 对消息分类，消费者订阅自己关心的消息。

两者底层实现不同，但是设计思想一致：生产者只负责生产消息，消费者只负责消费消息。

源码入口在哪里？

以 Java 常用 MQ 客户端为例。

生产消息：

Kafka：

KafkaProducer.send()
↓
RecordAccumulator
↓
Sender

RocketMQ：

DefaultMQProducer.send()
↓
MQClientInstance
↓
Broker

虽然实现不同，但是流程类似：

业务代码
↓
客户端
↓
网络请求
↓
Broker
↓
消息存储

真正解耦的关键，并不是某一个 API，而是 Broker 把生产和消费两个过程拆开了。

总结

消息队列能够解耦系统，本质不是因为它多了一层中间件。

而是它改变了服务之间的关系。

传统调用：

服务 A 必须知道 服务 B，任何变化都会影响调用方。

消息模型：

服务 A只产生事件
↓
消息队列
↓
谁需要谁消费

生产者和消费者拥有了独立的生命周期。这也是为什么大型系统越来越倾向于事件驱动架构。

上一篇：

《为什么需要消息队列？》

下一篇：

《消息队列如何保证消息可靠性？》

我们会继续往下拆：

一条消息从生产者发送出去，到 Broker 保存，再到消费者处理，中间到底有哪些地方可能丢消息？

以及 Kafka、RocketMQ 分别如何解决这个问题。