Redisson分布式锁+阻塞队列+线程池高并发协同方案

1. 项目概述：为什么一个分布式系统里，光靠synchronized根本扛不住真实业务

你有没有遇到过这种场景：某电商大促秒杀活动刚开抢，后台日志疯狂打印“库存扣减成功”，但财务对账时发现实际卖出的商品数量，比订单表里记录的多出好几百单？或者某支付系统在高并发下，同一笔订单被重复创建了三次支付流水，最后还得人工逐条核对、回滚？这些不是玄学，而是典型的分布式环境下的竞态条件（Race Condition）——当多个服务实例、多个线程同时去操作同一份共享资源（比如数据库里的库存字段、缓存里的用户积分），而缺乏统一协调机制时，必然发生的逻辑错乱。

标题里这五个关键词——Java、Redis、Redisson、分布式锁、阻塞队列、线程池——不是随便堆砌的技术名词，它们共同构成了一套经过千万级流量验证的、生产可用的高并发协同控制方案。我带过的几个中型后端团队，在把单体应用拆成微服务后，几乎都卡在这个环节：用synchronized锁住方法，本地测试稳如老狗；一上K8s集群，三台Pod同时处理请求，锁立刻失效，bug满天飞。后来我们彻底放弃“本地锁思维”，转而用Redisson封装好的分布式原语，配合线程池与阻塞队列做任务削峰，才真正把并发从“能跑通”推进到“可预测、可监控、可兜底”。

这个方案解决的不是“能不能用”的问题，而是“在1000+ QPS、5个服务节点、网络延迟波动在20ms~200ms之间的情况下，如何让库存扣减、订单生成、积分更新这些关键动作，像单机程序一样严格串行执行，且不因节点宕机或网络抖动导致锁永久丢失”。它面向的是真实世界的复杂性：Redis主从切换可能丢数据、JVM GC停顿可能让线程卡住100ms、某个服务实例突然OOM退出……所有这些，都必须在设计之初就考虑进去。所以这不是一个“加个注解就能搞定”的小技巧，而是一整套需要理解底层原理、权衡取舍、并亲手调过参数的工程实践。

如果你正在写一个需要强一致性的核心业务模块（比如优惠券发放、抽奖资格校验、分布式ID生成），或者正被线上偶发的数据不一致问题折磨得睡不着觉，那这篇内容就是为你写的。它不讲抽象理论，只讲我在三个不同业务系统里踩过的坑、调过的参数、压测时崩溃又重建的配置方案。接下来，我会带你一层层拆开这套组合拳：为什么选Redisson而不是自己用Redis命令手写锁？阻塞队列和线程池怎么配合才能既抗住流量洪峰，又不让内存爆掉？分布式锁的leaseTime到底设成多少才算安全？这些答案，全来自线上血泪经验。

2. 整体架构设计与技术选型逻辑：为什么是这五件套，而不是其他组合

2.1 为什么必须用Redis作为协调中心，而不是数据库或ZooKeeper？

很多人第一反应是：“我数据库里建个lock_table，update时加where version=xxx，不也能实现分布式锁？”——理论上可以，但实际落地会撞上三堵墙：

• 性能墙：MySQL单点写入TPS通常卡在3000~5000，而一个中等规模的秒杀服务，瞬时写请求轻松破万。每抢一次都要走一次数据库update+select，数据库直接变瓶颈，锁还没拿到，连接池先耗尽。
• 可靠性墙：数据库主从同步有延迟，A节点update成功，B节点还没同步完，B节点又发起update，两个节点都认为自己拿到了锁。这是典型的“脑裂”风险。
• 运维墙：ZooKeeper虽然一致性更强（ZAB协议），但部署复杂、运维成本高。一个5节点ZK集群，光是调优GC参数和网络超时时间，就能让初级运维抓狂。而Redis，几乎所有公司都有现成的集群，运维团队熟，客户端生态全。

Redis胜在高性能+最终一致性+极简运维。它的单线程模型保证了命令执行的原子性（setnx + expire必须用Lua脚本打包），集群模式（Redis Cluster或Codis）能水平扩展，哨兵机制自动故障转移。更重要的是，它天生就是为“快”设计的——锁的获取与释放，必须在毫秒级完成，否则整个业务链路都会被拖慢。我见过最极端的案例：某金融系统用MySQL做分布式锁，平均获取锁耗时47ms，高峰期直接导致支付接口P99延迟飙升到2.3秒，用户投诉电话打爆客服。

提示：Redis不是万能的。它不提供强一致性（CAP中的CP），所以不适合银行核心账务这类要求“绝对不丢不重”的场景。但对于电商库存、社交点赞、内容发布这类“允许短暂不一致，但最终必须收敛”的业务，它是目前综合成本最低、落地最成熟的方案。

2.2 为什么选Redisson而不是Jedis/Lettuce手写锁？

Jedis和Lettuce是Redis的Java客户端，功能强大，但它们只负责“发命令”，不负责“管逻辑”。你自己写一个可靠的分布式锁，至少要处理以下7个问题：

1. 加锁原子性：setnx + expire必须在一个Lua脚本里执行，否则setnx成功但expire失败，锁就永远不释放。
2. 锁续期（WatchDog）：业务处理时间不确定，如果锁过期了但线程还在跑，其他线程就会进来，造成并发冲突。必须有个后台线程定时给锁“续命”。
3. 可重入性：同一个线程多次加锁，不能被自己阻塞。需要记录线程ID和加锁次数。
4. 锁释放安全性：只能由加锁的线程释放，不能A线程加的锁被B线程误删。要用Lua脚本比对锁值。
5. 阻塞等待：线程拿不到锁时，是忙等（while循环）还是挂起（wait/notify）？忙等耗CPU，挂起要管理线程状态。
6. 锁失效处理：网络超时、Redis节点宕机，客户端怎么感知并清理本地状态？
7. RedLock争议：Redisson默认不启用RedLock（多Redis实例加锁），因为Martin Kleppmann等专家指出其在分区网络下仍有风险。Redisson选择更务实的单Redis集群+高可用保障。

Redisson把这些全封装好了。它提供的RLock接口，用法和Java原生ReentrantLock几乎一样：lock()、tryLock()、unlock()，背后自动处理了Lua脚本、WatchDog续期、线程ID绑定、异常清理。我对比过：手写一个生产级分布式锁，代码量200+行，测试用例要覆盖12种异常场景；用Redisson，核心代码就3行。省下的时间，够你多写两个业务接口。

注意：Redisson不是银弹。它的WatchDog默认续期时间是30秒（lockWatchdogTimeout），如果业务逻辑真的超过30秒没处理完，WatchDog会停止续期，锁自动过期。这时你需要评估：是优化业务逻辑（比如把大事务拆成小步骤），还是手动关闭WatchDog，改用tryLock(long waitTime, long leaseTime, TimeUnit unit)指定固定租期，并确保业务在leaseTime内必完成。

2.3 阻塞队列与线程池：为什么不是“锁住就干”，而是要“排队再干”？

分布式锁解决的是“谁来干”的问题，但没解决“干不过来怎么办”。想象一下：10万用户同时点“立即支付”，后端服务瞬间收到10万请求。如果每个请求都去争一把Redis锁，然后立刻执行扣库存、生订单、发消息——结果就是Redis连接被打满，锁竞争激烈到90%的请求都在空转等待，CPU飙到95%，真正干活的线程反而很少。

这就是典型的“请求洪峰”与“处理能力”不匹配。解决方案是引入缓冲层：用阻塞队列（BlockingQueue）做请求暂存，用线程池（ThreadPoolExecutor）做可控消费。

• 阻塞队列（如LinkedBlockingQueue）：它像一个有容量限制的“等候大厅”。当大厅满了（队列满），新来的请求要么被拒绝（快速失败），要么等待（阻塞）。这天然实现了流量整形（Traffic Shaping），把瞬时尖刺变成平滑曲线。
• 线程池（如ScheduledThreadPoolExecutor）：它像大厅里的“服务员”。你设定好最多5个服务员（corePoolSize=5），每人每次只服务1位顾客（一个Runnable任务）。服务员干完活不会下班，而是立刻接下一位。这样，无论外面涌来多少人，系统内部始终只有5个线程在稳定工作，内存、CPU、数据库连接数都可控。

关键在于，队列和线程池的参数必须联动设计。比如，你设队列容量为1000，线程池核心线程数为10，那么最大积压任务就是1000，但处理速度是10个/秒。如果QPS持续100，10秒后队列就满了，后续请求开始被拒绝。所以你要根据历史峰值QPS、平均处理耗时、可接受的最大等待时间，反向推算出合理的队列大小和线程数。我一般用这个公式粗略估算：队列容量 ≈ 峰值QPS × 平均处理耗时 × 2（留100%余量）。比如峰值QPS=500，平均处理200ms，那队列至少要500 × 0.2 × 2 = 200。

实操心得：千万别用无界队列（如LinkedBlockingQueue不设容量）。我亲眼见过一个系统用new LinkedBlockingQueue()（默认Integer.MAX_VALUE），结果上游突发流量，队列疯狂堆积，JVM堆内存几分钟内涨到8GB，Full GC频繁，最后OOM。一定要设上限，并配好拒绝策略（如CallerRunsPolicy，让调用线程自己执行任务，起到反压作用）。

2.4 五件套如何拧成一股绳：一个典型的工作流

现在把这五个技术点串起来，看它们在真实请求中如何协作。以“用户提交订单”为例：

1. 入口层（Controller）：接收HTTP请求，不做任何业务处理，只做参数校验和基础风控（如IP限频）。校验通过后，将订单数据封装成一个OrderTask对象，丢进阻塞队列（orderQueue.offer(task)）。
2. 消费层（线程池）：线程池里的工作线程（比如叫order-consumer-1）从队列里poll()取出一个OrderTask。此时，它开始执行真正的业务逻辑。
3. 临界区保护（Redisson锁）：在扣减库存前，工作线程调用RLock lock = redisson.getLock("stock:sku:" + skuId); lock.lock(30, TimeUnit.SECONDS);。如果锁被其他线程持有，当前线程会在队列里阻塞等待（Redisson默认用Semaphore模拟阻塞，不消耗线程），直到拿到锁或超时。
4. 业务执行与锁释放：拿到锁后，线程执行DB更新、Redis缓存修改、MQ消息发送等一系列操作。全部成功后，显式调用lock.unlock()。Redisson会自动删除锁Key，并停止对应的WatchDog。
5. 异常兜底：如果业务执行中抛出未捕获异常，必须在finally块里unlock()。Redisson也提供了lock.lockInterruptibly()支持中断，但生产环境慎用，避免中断导致锁未释放。

整个过程，锁只保护了“库存扣减”这一小段最敏感的代码，而不是整个订单创建流程。这极大缩短了锁持有时间，提升了吞吐量。而队列和线程池，则把不可控的请求洪峰，转化成了可控的、可监控的内部任务流。

3. 核心细节解析与实操要点：从代码到配置，每一个参数都有它的脾气

3.1 Redisson客户端初始化：不只是new一个对象那么简单

Redisson不是Spring Boot Starter那种“加个依赖就自动配置”的框架。它的初始化直接影响到锁的稳定性、连接复用率和故障恢复能力。下面这段代码，是我在线上跑了三年、经历过多次Redis集群扩容和网络抖动考验的配置模板：

@Configuration public class RedissonConfig { @Value("${redisson.address}") private String redisAddress; // 如 redis://192.168.1.100:6379 @Value("${redisson.password:}") private String redisPassword; @Bean(destroyMethod = "shutdown") public RedissonClient redissonClient() { Config config = new Config(); // 1. 单Redis节点配置（生产环境强烈建议用集群模式） // config.useSingleServer() // .setAddress(redisAddress) // .setPassword(redisPassword) // .setConnectionMinimumIdleSize(32) // 连接池最小空闲连接数 // .setConnectionPoolSize(64) // 连接池最大连接数 // .setIdleConnectionTimeout(10000) // 空闲连接超时，毫秒 // .setConnectTimeout(10000) // 连接超时 // .setTimeout(3000); // 命令超时 // 2. Redis集群模式（推荐！） config.useClusterServers() .addNodeAddress(redisAddress) // 可添加多个节点地址，如 redis://192.168.1.101:6379, redis://192.168.1.102:6379 .setPassword(redisPassword) .setScanInterval(2000) // 集群状态扫描间隔，毫秒 .setReadMode(ReadMode.SLAVE) // 读取模式：SLAVE（从节点读）、MASTER（主节点读）、MASTER_SLAVE（优先从节点） .setSubscriptionMode(SubscriptionMode.SLAVE) // 订阅模式 .setRetryAttempts(3) // 命令失败重试次数 .setRetryInterval(1500) // 重试间隔，毫秒 .setMasterConnectionPoolSize(64) // 主节点连接池大小 .setSlaveConnectionPoolSize(64) // 从节点连接池大小 .setIdleConnectionTimeout(10000) .setConnectTimeout(10000) .setTimeout(3000); // 3. 全局配置 config.setCodec(new JsonJacksonCodec()); // 使用JSON序列化，比默认的FST更易调试 config.setEventLoopGroup(new NioEventLoopGroup(4)); // Netty事件循环组，线程数建议CPU核心数*2 return Redisson.create(config); } }

关键参数详解与取舍理由：

• useClusterServers()vsuseSingleServer()：单节点配置简单，但存在单点故障风险。集群模式（Redis Cluster）能自动分片、故障转移，是生产环境唯一选择。addNodeAddress只需配置部分节点地址，Redisson会自动发现整个集群拓扑。
• setReadMode(ReadMode.SLAVE)：读操作默认走从节点，大幅降低主节点压力。但要注意：Redis主从同步有延迟（通常<100ms），如果业务要求“写后立刻读”，必须用ReadMode.MASTER，或在写操作后加getLock().forceUnlock()强制刷新。
• setRetryAttempts(3)和setRetryInterval(1500)：网络抖动时，Redisson会自动重试。3次重试+1.5秒间隔，基本能覆盖大部分瞬时网络闪断。设太多（如10次）会导致请求耗时拉长；设太少（如1次）则抖动时失败率高。
• setConnectionPoolSize(64)：这是每个Redis节点的连接池大小。计算公式：总连接数 ≈ (线程池核心线程数 + 队列消费者线程数) × 2。比如你有10个订单消费者线程，每个线程可能同时发2个命令（先查再改），那就需要20个连接。留3倍余量，设64是稳妥的。
• setCodec(new JsonJacksonCodec())：Redisson默认用FST序列化，速度快但调试困难（二进制看不懂）。换成JSON，Redis CLI里get命令直接看到明文，排查问题效率翻倍。性能损失在可接受范围（<5%）。

注意：@Bean(destroyMethod = "shutdown")至关重要。它确保Spring容器关闭时，Redisson能优雅释放Netty连接、停止WatchDog线程。漏掉这行，应用重启时旧连接可能残留，Redis服务器连接数缓慢上涨，几天后达到上限。

3.2 分布式锁的正确使用姿势：lock()、tryLock()、lockInterruptibly() 的实战边界

Redisson的RLock提供了三种加锁方式，新手常混淆它们的适用场景。下面用真实业务片段说明：

@Service public class OrderService { @Autowired private RedissonClient redissonClient; // 场景1：必须拿到锁才能继续，否则一直等（适合核心流程，如库存扣减） public void createOrderWithBlock(String orderId, String skuId) { RLock lock = redissonClient.getLock("order:lock:" + skuId); try { // 等待最多3秒，拿到锁后自动续期30秒 boolean isLocked = lock.tryLock(3, 30, TimeUnit.SECONDS); if (!isLocked) { throw new RuntimeException("获取库存锁失败，请稍后重试"); } // 执行扣库存DB操作... 扣减库存(skuId); } catch (InterruptedException e) { Thread.currentThread().interrupt(); // 恢复中断状态 throw new RuntimeException("线程被中断", e); } finally { if (lock.isHeldByCurrentThread()) { // 安全判断：只释放自己持有的锁 lock.unlock(); } } } // 场景2：尝试获取锁，不阻塞（适合非核心、可降级的逻辑，如日志记录） public void logOrderAsync(String orderId) { RLock lock = redissonClient.getLock("log:lock:" + orderId); if (lock.tryLock()) { // 不等待，立即返回true/false try { // 记录日志到DB或文件... writeLog(orderId); } finally { lock.unlock(); } } else { // 拿不到锁，直接异步发MQ，由另一个消费者处理，不阻塞主流程 mqProducer.sendAsyncLog(orderId); } } // 场景3：支持线程中断（极少用！仅用于需要精确控制线程生命周期的场景） public void cancelOrderWithInterrupt(String orderId) throws InterruptedException { RLock lock = redissonClient.getLock("cancel:lock:" + orderId); try { // 如果线程在等待锁时被interrupt，此方法会抛出InterruptedException lock.lockInterruptibly(10, TimeUnit.SECONDS); // 执行取消订单逻辑... } finally { if (lock.isHeldByCurrentThread()) { lock.unlock(); } } } }

核心原则：

• tryLock(long waitTime, long leaseTime, TimeUnit unit)是生产环境首选。它明确指定了“最多等多久”和“锁最多活多久”，避免了无限等待（饿死）和锁永不过期（死锁）两大风险。waitTime建议设为业务超时时间的1/3（如接口超时3秒，这里设1秒），leaseTime设为业务最长处理时间的2倍（如扣库存DB操作最长500ms，这里设1秒）。
• 永远不要用无参的lock()。它会无限等待，一旦Redis网络不通，线程就永远卡住，成为“僵尸线程”，拖垮整个线程池。
• lockInterruptibly()要慎用。Java的interrupt()机制在复杂框架（如Spring AOP、Dubbo）中容易被拦截或忽略，导致中断信号丢失，锁无法释放。除非你完全掌控线程生命周期（如自定义调度器），否则不用。
• finally块里的isHeldByCurrentThread()检查是保命符。它防止了“锁被其他线程释放”的灾难。比如A线程加锁后，因GC停顿10秒，锁自动过期；B线程拿到锁并执行完，调用unlock()；此时A线程醒来，也调用unlock()——如果没有这个检查，A会误删B的锁，导致后续并发。

实操心得：我在线上加了一个全局锁监控埋点。每次tryLock()成功，记录lockKey、threadId、acquireTime；每次unlock()，记录releaseTime。然后用Prometheus采集lock_acquire_duration_seconds（获取耗时）和lock_held_duration_seconds（持有耗时）两个指标。当发现某个lockKey的平均持有时间突然从200ms涨到2秒，立刻告警——这往往意味着下游DB慢了，或者代码里有隐藏的IO阻塞。

3.3 阻塞队列与线程池的黄金搭档：参数计算与拒绝策略实战

线程池不是Executors.newFixedThreadPool(10)这么简单。它有七个核心参数，每个都影响系统生死。下面是一个为订单服务量身定制的线程池配置：

@Configuration public class ThreadPoolConfig { @Value("${order.threadpool.core-size:5}") private int corePoolSize; @Value("${order.threadpool.max-size:20}") private int maxPoolSize; @Value("${order.threadpool.queue-capacity:1000}") private int queueCapacity; @Value("${order.threadpool.keep-alive:60}") private long keepAliveTime; @Bean("orderTaskExecutor") public ThreadPoolTaskExecutor orderTaskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(corePoolSize); // 核心线程数 executor.setMaxPoolSize(maxPoolSize); // 最大线程数 executor.setQueueCapacity(queueCapacity); // 队列容量 executor.setKeepAliveSeconds(keepAliveTime); // 非核心线程空闲存活时间 executor.setThreadNamePrefix("order-consumer-"); // 线程名前缀，方便日志追踪 executor.setRejectedExecutionHandler(new CustomRejectedHandler()); // 自定义拒绝策略 executor.setWaitForTasksToCompleteOnShutdown(true); // 关闭时等待任务完成 executor.setAwaitTerminationSeconds(60); // 等待超时时间 executor.initialize(); return executor; } /** * 自定义拒绝策略：当队列满且线程数已达max，新任务被拒绝时触发 */ static class CustomRejectedHandler implements RejectedExecutionHandler { private static final Logger logger = LoggerFactory.getLogger(CustomRejectedHandler.class); @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // 1. 记录告警日志 logger.error("订单任务队列已满！当前队列大小:{}, 活跃线程数:{}, 任务被拒绝:{}", executor.getQueue().size(), executor.getActiveCount(), r.getClass().getSimpleName()); // 2. 发送企业微信/钉钉告警（伪代码） // alertService.send("订单队列告警", "队列已满，需紧急扩容"); // 3. 尝试降级：把任务发到Kafka，由低优先级消费者异步处理 // kafkaTemplate.send("order-fallback-topic", r); // 4. 最终，抛出异常让上游感知（如返回503 Service Unavailable） throw new RuntimeException("系统繁忙，请稍后重试"); } } }

参数计算逻辑（必须手算，不能拍脑袋）：

假设你的订单服务历史数据如下：

• 日常QPS：800
• 大促峰值QPS：5000
• 平均单个订单处理耗时：300ms（含DB、Redis、MQ调用）
• 可接受的最大任务等待时间：2秒（用户感知超时）

那么：

• 核心线程数（corePoolSize）=日常QPS × 平均耗时=800 × 0.3 = 240→ 但这是理论值，实际要结合机器CPU。一台16核机器，设corePoolSize=16（CPU核心数），让线程数与CPU并行度匹配，避免过多线程上下文切换。所以这里取min(240, 16) = 16。
• 最大线程数（maxPoolSize）=峰值QPS × 平均耗时=5000 × 0.3 = 1500→ 同样受限于机器资源，设为16 × 2 = 32（预留100%弹性）。
• 队列容量（queueCapacity）=可接受最大等待时间 × 峰值QPS=2 × 5000 = 10000。但注意：队列太大，内存占用高；太小，拒绝率高。折中取5000，并配好拒绝策略。
• keepAliveTime：设60秒足够。非核心线程空闲60秒后回收，避免资源浪费。

为什么拒绝策略必须自定义？

AbortPolicy（默认）直接抛异常，上游可能没处理，导致请求无声失败；CallerRunsPolicy让调用线程自己执行，可能拖慢Web容器线程（如Tomcat的http-nio-8080-exec-1），引发雪崩。自定义策略能做三件事：记录、告警、降级。比如把被拒任务发到Kafka，由一个低优先级的“补单服务”异步处理，保证不丢单，只是延迟几秒。

注意：setWaitForTasksToCompleteOnShutdown(true)和setAwaitTerminationSeconds(60)是优雅停机的关键。应用发布时，Spring会先停止接收新请求，然后等待60秒，让线程池里正在跑的任务自然结束。如果没有这个，强行kill进程，正在执行的订单可能只扣了库存没生成订单，数据就永久不一致了。

4. 实操过程与核心环节实现：从零搭建一个可运行的订单服务Demo

4.1 环境准备：三步搞定本地开发环境

要跑通这个Demo，你不需要搭一套真实的Redis集群。用Docker一条命令就能启动一个功能完整的Redis服务：

# 1. 启动Redis（带密码，模拟生产环境） docker run -d --name my-redis -p 6379:6379 -e REDIS_PASSWORD=redis123 redis:7-alpine # 2. 验证是否启动成功 docker exec -it my-redis redis-cli -a redis123 ping # 返回 "PONG" 即成功 # 3. （可选）安装RedisInsight，可视化查看锁Key # 下载地址：https://redis.com/redis-enterprise/redis-insight/ # 连接时填：Host=127.0.0.1, Port=6379, Password=redis123

Maven依赖（pom.xml）：
只保留最精简的核心依赖，避免版本冲突：

<dependencies> <!-- Spring Boot Web --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!-- Redisson --> <dependency> <groupId>org.redisson</groupId> <artifactId>redisson-spring-boot-starter</artifactId> <version>3.23.3</version> <!-- 选最新稳定版，3.23.x系列对Redis 7兼容性最好 --> </dependency> <!-- Lombok，简化POJO --> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> </dependencies>

配置文件（application.yml）：
所有可配置项都外置，方便不同环境切换：

server: port: 8080 spring: redis: host: 127.0.0.1 port: 6379 password: redis123 timeout: 3000 # Redisson专用配置 redisson: address: redis://127.0.0.1:6379 password: redis123 # 订单线程池配置 order: threadpool: core-size: 4 max-size: 8 queue-capacity: 100 keep-alive: 60

4.2 核心代码实现：一个可直接运行的订单创建服务

第一步：定义订单任务实体（OrderTask）

import lombok.Data; import java.math.BigDecimal; @Data public class OrderTask { private String orderId; private String userId; private String skuId; private Integer quantity; private BigDecimal amount; // 构造函数、toString等略 }

第二步：创建阻塞队列与线程池消费者

@Service @Slf4j public class OrderConsumerService { // 使用Spring管理的线程池 @Autowired @Qualifier("orderTaskExecutor") private ThreadPoolTaskExecutor taskExecutor; // 内存队列（实际项目中，这里应是注入的BlockingQueue bean） private final BlockingQueue<OrderTask> orderQueue = new LinkedBlockingQueue<>(100); // 模拟库存服务（实际应是DB或Redis） private final Map<String, Integer> inventoryMap = new ConcurrentHashMap<>(); @PostConstruct public void initInventory() { // 初始化一些测试商品库存 inventoryMap.put("SKU001", 1000); inventoryMap.put("SKU002", 500); } /** * 入口：接收HTTP请求，入队 */ @PostMapping("/api/order") public ResponseEntity<String> submitOrder(@RequestBody OrderTask task) { try { // 1. 参数校验 if (task.getQuantity() == null || task.getQuantity() <= 0) { return ResponseEntity.badRequest().body("数量必须大于0"); } if (!inventoryMap.containsKey(task.getSkuId())) { return ResponseEntity.badRequest().body("商品不存在"); } // 2. 入队，如果队列满则拒绝 boolean offered = orderQueue.offer(task); if (!offered) { log.warn("订单队列已满，拒绝任务: {}", task.getOrderId()); return ResponseEntity.status(503).body("系统繁忙，请稍后重试"); } log.info("订单[{}]已入队，当前队列大小: {}", task.getOrderId(), orderQueue.size()); return ResponseEntity.ok("订单已提交，正在处理"); } catch (Exception e) { log.error("提交订单异常", e); return ResponseEntity.status(500).body("系统错误"); } } /** * 消费者：后台线程持续从队列取任务执行 */ @EventListener(ApplicationReadyEvent.class) public void startConsuming() { log.info("订单消费者启动，开始监听队列..."); // 启动一个守护线程，持续消费 Thread consumerThread = new Thread(() -> { while (!Thread.currentThread().isInterrupted()) { try { OrderTask task = orderQueue.poll(1, TimeUnit.SECONDS); // 等待1秒，避免空转 if (task != null) { // 提交到线程池执行 taskExecutor.execute(() -> processOrder(task)); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { log.error("消费者线程异常", e); } } }, "order-consumer-main"); consumerThread.setDaemon(true); consumerThread.start(); } /** * 真正的订单处理逻辑（带分布式锁） */ private void processOrder(OrderTask task) { RLock lock = null; String lockKey = "inventory:lock:" + task.getSkuId(); try { lock = redissonClient.getLock(lockKey); // 尝试获取锁，最多等1秒，锁有效期2秒（足够扣库存） boolean isLocked = lock.tryLock(1, 2, TimeUnit.SECONDS); if (!isLocked) { log.warn("获取库存锁失败，订单[{}]被拒绝", task.getOrderId()); // 这里可以发MQ通知前端“秒杀失败” return; } log.info("订单[{}]成功获取锁，开始处理...", task.getOrderId()); // 模拟DB查询库存 Integer currentStock = inventoryMap.get(task.getSkuId()); log.info("商品[{}]当前库存: {}", task.getSkuId(), currentStock); if (currentStock == null || currentStock < task.getQuantity()) { log.warn("商品[{}]库存不足，当前:{}，需:{}", task.getSkuId(), currentStock, task.getQuantity()); return; } // 扣减库存（模拟DB update） inventoryMap.put(task.getSkuId(), currentStock - task.getQuantity()); log.info("商品[{}]库存已扣减，剩余: {}", task.getSkuId(), inventoryMap.get(task.getSkuId())); // 模拟生成订单（实际是插入orders表） log.info("订单[{}]创建成功！"); } catch (Exception e) { log.error("处理订单[{}]异常", task.getOrderId(), e); } finally { // 安全释放锁 if (lock != null && lock.isHeldByCurrentThread()) { lock.unlock(); log.info("订单[{}]锁已释放", task.getOrderId()); } } } }

第三步：编写Controller暴露API

@RestController @RequestMapping("/api") @Slf4j public class OrderController { @Autowired private OrderConsumerService orderConsumer