⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处苏三说技术「苏三」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

前言

消息队列（MQ）是分布式系统中不可或缺的技术之一。

对很多小伙伴来说，刚接触MQ时，可能觉得它只是个“传话工具”，但用着用着，你会发现它简直是系统的“润滑剂”。

无论是解耦、削峰，还是异步任务处理，都离不开MQ的身影。

下面我结合实际场景，从简单到复杂，逐一拆解MQ的10种经典使用方式，希望对你会有所帮助。

1. 异步处理：让系统轻松一点

场景

小伙伴们是不是经常遇到这样的情况：用户提交一个操作，比如下单，然后要发送短信通知。

如果直接在主流程里调用短信接口，一旦短信服务响应慢，就会拖累整个操作。

用户等得不耐烦，心态直接崩了。

解决方案

用MQ，把非关键流程抽出来异步处理。下单时，直接把“发短信”这件事丢给MQ，订单服务就能立刻响应用户，而短信的事情让MQ和消费者去搞定。

示例代码

// 订单服务：生产者
Order order = createOrder(); // 订单生成逻辑
rabbitTemplate.convertAndSend("order_exchange", "order_key", order);
System.out.println("订单已生成，发短信任务交给MQ");

// 短信服务：消费者
@RabbitListener(queues = "sms_queue")
public void sendSms(Order order) {
    System.out.println("发送短信，订单ID：" + order.getId());
    // 调用短信服务接口
}

深度解析

这种方式的好处是：主流程解耦，不受慢服务的拖累。订单服务只管自己的事，短信服务挂了也没关系，MQ会把消息暂存，等短信服务恢复后继续处理。

2. 流量削峰：稳住系统别崩

场景

每年的“双十一”电商大促，用户秒杀商品时一窝蜂冲进来。

突然涌入的高并发请求，不仅会压垮应用服务，还会直接让数据库“趴窝”。

解决方案

秒杀请求先写入MQ，后端服务以稳定的速度从MQ中消费消息，处理订单。

这样既能避免系统被瞬时流量压垮，还能提升处理的平稳性。

示例代码

// 用户提交秒杀请求：生产者
rabbitTemplate.convertAndSend("seckill_exchange", "seckill_key", userRequest);
System.out.println("用户秒杀请求已进入队列");

// 秒杀服务：消费者
@RabbitListener(queues = "seckill_queue")
public void processSeckill(UserRequest request) {
    System.out.println("处理秒杀请求，用户ID：" + request.getUserId());
    // 执行秒杀逻辑
}

深度解析

MQ在这里相当于一个缓冲池，把瞬时流量均匀分布到一段时间内处理。系统稳定性提升，用户体验更好。

3. 服务解耦：减少相互牵制

场景

比如一个订单系统需要通知库存系统扣减库存，还要通知支付系统完成扣款。

如果直接用同步接口调用，服务间的依赖性很强，一个服务挂了，整个链条都会被拖垮。

解决方案

订单服务只负责把消息丢到MQ里，库存服务和支付服务各自从MQ中消费消息。

这样订单服务不需要直接依赖它们。

示例代码

// 订单服务：生产者
rabbitTemplate.convertAndSend("order_exchange", "order_key", order);
System.out.println("订单生成消息已发送");

// 库存服务：消费者
@RabbitListener(queues = "stock_queue")
public void updateStock(Order order) {
    System.out.println("扣减库存，订单ID：" + order.getId());
}

// 支付服务：消费者
@RabbitListener(queues = "payment_queue")
public void processPayment(Order order) {
    System.out.println("处理支付，订单ID：" + order.getId());
}

深度解析

通过MQ，各个服务之间可以实现松耦合。

即使库存服务挂了，也不会影响订单生成的流程，大幅提升系统的容错能力。

4. 分布式事务：保证数据一致性

场景

订单服务需要同时生成订单和扣减库存，这涉及两个不同的数据库操作。

如果一个成功一个失败，就会导致数据不一致。

解决方案

通过MQ实现分布式事务。

订单服务生成订单后，将扣减库存的任务交给MQ，最终实现数据的一致性。

示例代码

// 订单服务：生产者
rabbitTemplate.convertAndSend("order_exchange", "order_key", order);
System.out.println("订单创建消息已发送");

// 库存服务：消费者
@RabbitListener(queues = "stock_queue")
public void updateStock(Order order) {
    System.out.println("更新库存，订单ID：" + order.getId());
    // 执行扣减库存逻辑
}

深度解析

通过“最终一致性”解决了分布式事务的难题，虽然短时间内可能有数据不一致，但最终状态一定是正确的。

5. 广播通知：一条消息，通知多个服务

场景

比如商品价格调整，库存、搜索、推荐服务都需要同步更新。

如果每个服务都要单独通知，工作量会很大。

解决方案

MQ的广播模式（Fanout）可以让多个消费者订阅同一条消息，实现消息的“一发多收”。

示例代码

// 生产者：广播消息
rabbitTemplate.convertAndSend("price_update_exchange", "", priceUpdate);
System.out.println("商品价格更新消息已广播");

// 消费者1：库存服务
@RabbitListener(queues = "stock_queue")
public void updateStockPrice(PriceUpdate priceUpdate) {
    System.out.println("库存价格更新：" + priceUpdate.getProductId());
}

// 消费者2：搜索服务
@RabbitListener(queues = "search_queue")
public void updateSearchPrice(PriceUpdate priceUpdate) {
    System.out.println("搜索价格更新：" + priceUpdate.getProductId());
}

深度解析

这种模式让多个服务都能接收到同一条消息，扩展性非常强。

6. 日志收集：分布式日志集中化

场景

多个服务产生的日志需要统一存储和分析。

如果直接写数据库，可能导致性能瓶颈。

解决方案

各服务将日志写入MQ，日志分析系统从MQ中消费消息并统一处理。

示例代码

// 服务端：生产者
rabbitTemplate.convertAndSend("log_exchange", "log_key", logEntry);
System.out.println("日志已发送");

// 日志分析服务：消费者
@RabbitListener(queues = "log_queue")
public void processLog(LogEntry log) {
    System.out.println("日志处理：" + log.getMessage());
    // 存储或分析逻辑
}

7. 延迟任务：定时触发操作

场景

用户下单后，如果30分钟内未支付，需要自动取消订单。

解决方案

使用MQ的延迟队列功能，设置消息延迟消费的时间。

示例代码

// 生产者：发送延迟消息
rabbitTemplate.convertAndSend("delay_exchange", "delay_key", order, message -> {
    message.getMessageProperties().setDelay(30 * 60 * 1000); // 延迟30分钟
    return message;
});
System.out.println("订单取消任务已设置");

// 消费者：处理延迟消息
@RabbitListener(queues = "delay_queue")
public void cancelOrder(Order order) {
    System.out.println("取消订单：" + order.getId());
    // 取消订单逻辑
}

8. 数据同步：跨系统保持数据一致

场景

在一个分布式系统中，多个服务依赖同一份数据源。

例如，电商平台的订单状态更新后，需要同步到缓存系统和推荐系统。

如果让每个服务直接从数据库拉取数据，会增加数据库压力，还可能出现延迟或不一致的问题。

解决方案

利用MQ进行数据同步。订单服务更新订单状态后，将更新信息发送到MQ，缓存服务和推荐服务从MQ中消费消息并同步数据。

示例代码

订单服务：生产者

// 更新订单状态后，将消息发送到MQ
Order order = updateOrderStatus(orderId, "PAID"); // 更新订单状态为已支付
rabbitTemplate.convertAndSend("order_exchange", "order_status_key", order);
System.out.println("订单状态更新消息已发送：" + order.getId());

缓存服务：消费者

@RabbitListener(queues = "cache_update_queue")
public void updateCache(Order order) {
    System.out.println("更新缓存，订单ID：" + order.getId() + " 状态：" + order.getStatus());
    // 更新缓存逻辑
    cacheService.update(order.getId(), order.getStatus());
}

推荐服务：消费者

@RabbitListener(queues = "recommendation_queue")
public void updateRecommendation(Order order) {
    System.out.println("更新推荐系统，订单ID：" + order.getId() + " 状态：" + order.getStatus());
    // 更新推荐服务逻辑
    recommendationService.updateOrderStatus(order);
}

深度解析

通过MQ实现数据同步的好处是：

减轻数据库压力：避免多个服务同时查询数据库。
最终一致性：即使某个服务处理延迟，MQ也能保障消息不丢失，最终所有服务的数据状态一致。

9. 分布式任务调度

场景

有些任务需要定时执行，比如每天凌晨清理过期订单。

这些订单可能分布在多个服务中，如果每个服务独立运行定时任务，可能会出现重复处理或任务遗漏的问题。

解决方案

使用MQ统一分发调度任务，每个服务根据自身的业务需求，从MQ中消费任务并执行。

示例代码

任务调度服务：生产者

// 定时任务生成器
@Scheduled(cron = "0 0 0 * * ?") // 每天凌晨触发
public void generateTasks() {
    List<Task> expiredTasks = taskService.getExpiredTasks();
    for (Task task : expiredTasks) {
        rabbitTemplate.convertAndSend("task_exchange", "task_routing_key", task);
        System.out.println("任务已发送：" + task.getId());
    }
}

订单服务：消费者

@RabbitListener(queues = "order_task_queue")
public void processOrderTask(Task task) {
    System.out.println("处理订单任务：" + task.getId());
    // 执行订单清理逻辑
    orderService.cleanExpiredOrder(task);
}

库存服务：消费者

@RabbitListener(queues = "stock_task_queue")
public void processStockTask(Task task) {
    System.out.println("处理库存任务：" + task.getId());
    // 执行库存释放逻辑
    stockService.releaseStock(task);
}

深度解析

分布式任务调度可以解决：

重复执行：每个服务只处理自己队列中的任务。
任务遗漏：MQ确保任务可靠传递，防止任务丢失。

10. 文件处理：异步执行大文件任务

场景

用户上传一个大文件后，需要对文件进行处理（如格式转换、压缩等）并存储。

如果同步执行这些任务，前端页面可能会一直加载，导致用户体验差。

解决方案

用户上传文件后，立即将任务写入MQ，后台异步处理文件，处理完成后通知用户或更新状态。

示例代码

上传服务：生产者

// 上传文件后，将任务写入MQ
FileTask fileTask = new FileTask();
fileTask.setFileId(fileId);
fileTask.setOperation("COMPRESS");
rabbitTemplate.convertAndSend("file_task_exchange", "file_task_key", fileTask);
System.out.println("文件处理任务已发送，文件ID：" + fileId);

文件处理服务：消费者

@RabbitListener(queues = "file_task_queue")
public void processFileTask(FileTask fileTask) {
    System.out.println("处理文件任务：" + fileTask.getFileId() + " 操作：" + fileTask.getOperation());
    // 模拟文件处理逻辑
    if ("COMPRESS".equals(fileTask.getOperation())) {
        fileService.compressFile(fileTask.getFileId());
    } else if ("CONVERT".equals(fileTask.getOperation())) {
        fileService.convertFileFormat(fileTask.getFileId());
    }
    // 更新任务状态
    taskService.updateTaskStatus(fileTask.getFileId(), "COMPLETED");
}

前端轮询或回调通知

// 前端轮询文件处理状态
setInterval(() => {
    fetch(`/file/status?fileId=${fileId}`)
        .then(response => response.json())
        .then(status => {
            if (status === "COMPLETED") {
                alert("文件处理完成！");
            }
        });
}, 5000);

深度解析

异步文件处理的优势：

提升用户体验：主线程迅速返回，减少用户等待时间。
后台任务灵活扩展：支持多种操作逻辑，适应复杂文件处理需求。

总结

消息队列不只是传递消息的工具，更是系统解耦、提升稳定性和扩展性的利器。

在这10种经典场景中，每一种都能解决特定的业务痛点。