MQ解决高并发下订单问题，实现流量削峰

以下是一个基于 Java 和 RabbitMQ 实现流量削峰的示例，展示如何通过消息队列处理高并发下单请求：

示例：电商秒杀系统中的流量削峰

1. 依赖引入（Maven）

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.14.2</version>
</dependency>

2. 消息队列配置（RabbitMQ）

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class MQConfig {private static final String HOST = "localhost";private static final int PORT = 5672;private static final String USERNAME = "guest";private static final String PASSWORD = "guest";public static final String QUEUE_NAME = "order_queue";// 创建连接工厂public static ConnectionFactory getConnectionFactory() {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setPort(PORT);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);return factory;}
}

3. 生产者：订单服务（接收高并发请求）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;public class OrderService {private final Connection connection;public OrderService() throws IOException, TimeoutException {this.connection = MQConfig.getConnectionFactory().newConnection();}// 处理下单请求（削峰前）public void createOrderDirectly(Long productId, Integer count) {// 传统模式：直接处理订单（高并发时会压垮数据库）System.out.println("直接处理订单：商品ID=" + productId + ", 数量=" + count);// 模拟数据库操作try {Thread.sleep(200); // 假设处理一个订单需要200ms} catch (InterruptedException e) {e.printStackTrace();}}// 处理下单请求（削峰后）public void createOrderWithMQ(Long productId, Integer count) throws IOException {try (Channel channel = connection.createChannel()) {// 声明队列（如果不存在则创建）channel.queueDeclare(MQConfig.QUEUE_NAME, false, false, false, null);// 封装订单信息为JSONString orderInfo = "{\"productId\":" + productId + ",\"count\":" + count + "}";// 发送消息到队列channel.basicPublish("", MQConfig.QUEUE_NAME, null, orderInfo.getBytes());System.out.println("订单已放入队列：" + orderInfo);}}
}

4. 消费者：库存服务（按系统容量处理订单）

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;public class InventoryService {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = MQConfig.getConnectionFactory();try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明队列channel.queueDeclare(MQConfig.QUEUE_NAME, false, false, false, null);// 设置消费者每次只处理1条消息（限流）channel.basicQos(1);System.out.println("库存服务已启动，等待订单消息...");// 创建消费者DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("收到订单：" + message);try {// 模拟处理订单（扣库存、更新数据库等）processOrder(message);// 手动确认消息已处理channel.basicAck(envelope.getDeliveryTag(), false);} catch (Exception e) {e.printStackTrace();// 处理失败，拒绝消息并重新入队channel.basicNack(envelope.getDeliveryTag(), false, true);}}};// 启动消费者（手动确认模式）channel.basicConsume(MQConfig.QUEUE_NAME, false, consumer);}}private static void processOrder(String orderInfo) {try {// 模拟处理订单耗时（如扣减库存、写入订单表）Thread.sleep(500);System.out.println("订单处理完成：" + orderInfo);} catch (InterruptedException e) {e.printStackTrace();}}
}

5. 模拟高并发测试

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;public class TrafficPeakTest {public static void main(String[] args) throws IOException, TimeoutException {OrderService orderService = new OrderService();ExecutorService executor = Executors.newFixedThreadPool(100); // 模拟100个并发用户// 模拟1000个并发下单请求（流量峰值）for (int i = 0; i < 1000; i++) {final int orderId = i;executor.submit(() -> {try {// 削峰前：直接处理订单（可能导致系统崩溃）// orderService.createOrderDirectly(1001L, 1);// 削峰后：通过MQ异步处理订单orderService.createOrderWithMQ(1001L, 1);} catch (Exception e) {e.printStackTrace();}});}executor.shutdown();}
}

关键技术点解析

1. 流量削峰的实现

• 生产者端：将订单请求快速放入队列后立即返回，避免请求堆积
• 消费者端：
  • 通过 channel.basicQos(1) 限制每次只处理 1 条消息
  • 单线程消费（可扩展为多线程），每秒处理约 2 个订单（500ms / 订单）
• 效果：1000 个并发请求被队列缓冲，系统按自身容量（2TPS）平稳处理

2. 消息可靠性保障

• 持久化：RabbitMQ 默认将消息存储在内存中，可配置持久化到磁盘
• 手动确认：消费者处理完成后手动 basicAck，失败则 basicNack 并重试
• 死信队列：可配置死信队列存储多次处理失败的消息

3. 削峰前后对比

指标	无 MQ（传统模式）	有 MQ（流量削峰）
最大并发处理量	受数据库连接数限制（如 100）	队列可缓冲无限量请求
响应时间	平均 200ms（直接处理）	立即返回（<10ms）
系统稳定性	峰值时易崩溃	平稳处理，无崩溃风险
资源利用率	峰值时资源耗尽，平时闲置	按固定速率使用资源

生产环境优化建议

1. 队列监控：

   • 监控队列长度，设置告警阈值（如超过 10 万条未处理消息）
   • 使用 RabbitMQ Management 插件或 Prometheus + Grafana 监控

2. 消费者扩容：

   • 垂直扩容：增加消费者机器配置
   • 水平扩容：增加消费者实例数（需注意幂等性）

3. 降级策略：

   • 队列过长时，拒绝新请求并返回 “系统繁忙”
   • 非核心业务降级（如暂时关闭短信通知）

4. 持久化配置：

   // 设置队列持久化
   boolean durable = true;
   channel.queueDeclare(MQConfig.QUEUE_NAME, durable, false, false, null);// 设置消息持久化
   AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().deliveryMode(2) // 2表示持久化.build();
   channel.basicPublish("", MQConfig.QUEUE_NAME, properties, message.getBytes());
   

其他 MQ 选型参考

消息队列	吞吐量	优势场景	示例项目
RabbitMQ	万级 TPS	强一致性、支持事务、灵活路由	金融系统订单处理
Kafka	百万级 TPS	大数据实时处理	日志收集、实时数据流处理
RocketMQ	十万级 TPS	高可用、顺序消息	电商订单、物流系统
Pulsar	百万级 TPS	云原生、多租户	分布式微服务架构

根据业务场景选择合适的 MQ，本例使用 RabbitMQ 是因其易用性和可靠性。