简介这是一份基于 Java 的多协议服务处理器MPSP开源项目源码面向网络编程学习者与中高级 Java 开发工程师适合用来研究 TCP/IP、UDP、HTTP 等网络协议的并发处理与服务端架构。压缩包共 23 个文件其中 15 个 Java 源文件按 src/test 目录组织另含 Gradle 构建脚本、配置文件、gradlew 启动脚本及 README 说明整体仅 18KB结构紧凑导入工程后即可快速定位核心代码。已有 443 人学习下载。通过源码可系统掌握套接字编程、NIO 非阻塞 IO、线程池设计、协议解析流程以及工厂、单例、观察者等设计模式的工程落地并体现清晰的包结构与分层设计思路项目中的单元测试、CI 配置和日志监控选型也为构建健壮、可扩展的网络服务提供了完整的参考骨架适合进阶学习和二次开发。1. MPSP 消息处理平台一套能扛突发流量又不丢消息的 Java 异步骨架第一次接到消息处理需求时我的第一反应很简单消费者拉一批消息for 循环里挨个调用业务方法不就行了。真上线才发现这种裸写方式在流量小时没毛病流量一起来就暴露出三个致命伤——消息重复消费没人管、一批里只要一条失败就把整批拖垮、线程池参数全靠猜。MPSP 就是这样一套基于 Java 的轻量消息处理平台骨架它不替代 Kafka 或 RocketMQ只负责业务侧从拉取、分发、幂等到可靠重试的整条链路。适合正在写事件驱动模块、支付回调或异步任务的开发者拿来即用也看得见每个参数为什么存在。2. MPSP 的整体结构消息模型、线程模型与为什么不能裸写线程池2.1 消息模型Header 与 Body 分离状态字段就是排查问题的黑匣子MPSP 里所有消息都统一成同一个 Message 对象Header 和 Body 分开存放。Header 只放路由和状态信息Body 放业务 JSON。这样设计的好处很直白线上出问题时不需要把每条消息的 body 都拉出来翻一遍只看 Header 就能判断这条消息属于哪个主题、对应哪笔业务、重试了几次、是什么时候产生的。// 核心消息模型字段刻意精简 public class Message { private String msgId; // 全局唯一生成后不再变化幂等依赖它 private String topic; // 业务主题如 order_event private String bizId; // 业务主键顺序路由和去重都看它 private int retryCount; // 已重试次数超过阈值进死信 private long produceTime; // 生产时间戳端到端延迟靠它算 private String body; // 业务数据 JSON // getter / setter 省略 }注意 msgId 和 bizId 是两个东西。msgId 是消息自身的唯一编号一条消息重复投递 n 次msgId 都不变bizId 是业务侧的主键比如订单号。幂等去重用 msgId顺序路由用 bizId两者职责不能混。我见过有人拿订单号当 msgId 用同一个订单发两条不同事件时直接撞了唯一约束这种翻车在接入第一天就能遇到。2.2 线程模型单线程拉取 有界缓冲 工作线程池很多人一上来就把拉取和业务处理放在同一个线程里拉一条、处理一条、再拉一条。流量低时看不出问题一旦业务处理里有个外部 HTTP 调用耗时 500ms拉取线程就被卡住MQ 客户端的心跳超时消费组直接判定你离线触发 rebalance。这就是裸写消费逻辑最常见的线上事故。MPSP 把链路拆成三段一个拉取线程负责从 MQ 批量拿消息放到一个有界缓冲队列里工作线程池负责真正执行业务逻辑拉取线程和业务线程之间只通过缓冲队列通信互不阻塞。// 拉取线程把批量消息搬进有界缓冲缓冲满时阻塞拉取天然背压 public class PullWorker implements Runnable { private final BlockingQueueMessage buffer; // 有界缓冲默认 2000 private final ThreadPoolExecutor workers; // 业务工作线程池 Override public void run() { while (running) { // 拉取一批超时 500ms避免空转 ListMessage batch batchPuller.pull(batchSize, 500); for (Message msg : batch) { buffer.put(msg); // 缓冲满会阻塞在这里让 MQ 端自然积压 workers.execute(() - handle(msg)); // 丢给业务线程 } } } }这里的关键在buffer.put(msg)。有界队列满的时候拉取线程会阻塞在 put 上不再从 MQ 拉新消息。这比线程池的拒绝策略温和得多——拒绝策略是直接抛弃任务而阻塞是让压力顺着链路往回传由 MQ 端做真正的积压。buffer 容量就是整个系统允许的最大在途消息数一般按期望吞吐 × 单条处理耗时 × 2估算。2.3 为什么不用多消费者直接竞争顺序性由 bizId 路由保证也有人说既然有工作线程池我多开几个消费者客户端直接竞争消费不就行了技术上能跑但会引入两个问题第一同一分区的消息会被多个线程并发处理订单状态机这类强顺序场景直接乱套第二幂等去重逻辑分散在每个消费线程里重复消息可能同时到达去重形同虚设。MPSP 的做法是把并发收拢到缓冲队列之后再做一次路由。每个 bizId 的哈希值固定映射到一个工作线程同一个订单的后续事件永远落在同一个线程上顺序天然保证。// 按 bizId 哈希路由保证同一业务的事件始终由同一线程处理 int threadIndex Math.floorMod(msg.getBizId().hashCode(), workerCount);代价是线程之间可能负载不均某个大订单流量特别集中时单线程会成为瓶颈。但从生产实践看绝大多数业务的顺序要求都集中在单个业务主键内部这种路由方式比全局锁或者分区重排简单可靠得多。3. 接入 MPSP 的完整链路生产者、消费者与幂等去重3.1 生产者端事务内先落库再投递失败走本地补偿接入 MPSP 时生产者端最容易踩的坑是“先发消息再更新业务状态”。一旦发消息成功、更新数据库失败消费者就会拿到一个业务上并不存在的事件而且你还没法撤回。MPSP 的常规做法是本地消息表业务表和消息状态表在同一个事务里写入消息状态先置为 PENDING事务提交后再真正发送到 MQ发送成功才把状态改成 SENT。Transactional public boolean sendOrderEvent(OrderEvent event) { // 1. 业务表与消息状态表同事务写入msgId 由这里统一生成 String msgId UUID.randomUUID().toString().replace(-, ); messageStateDao.save(new MessageState(msgId, event.getOrderId(), PENDING)); // 2. 事务提交后发送到 MQ boolean ok mqProducer.send(buildMessage(msgId, event)); if (ok) { messageStateDao.updateStatus(msgId, SENT); } return ok; }注意这里是事务方法但 MQ 发送放在了事务内。严格来说事务未提交时 MQ 消息已经发出去了消费端如果恰好立刻拉取可能看不到业务数据。常见做法是让 MQ 发送在事务提交后执行用TransactionSynchronizationManager.registerSynchronization挂一个 afterCommit 回调。同时本地任务表里始终保留 PENDING 记录定时任务每隔几秒扫描一次把 PENDING 超过 5 秒的消息重新投递投递成功再更新状态。这套补偿机制才是可靠性兜底MQ 本身只保证不丢不保证时序。3.2 消费者端批量拉取、逐条确认不整批提交位点消费者端的第一个原则是批量拉取、逐条处理。批量拉取能降低 MQ 交互次数但提交位点时不能一批一起提交——一批 200 条消息里只要有一条处理失败整批滚动回去重新拉取其余 199 条都会被重复消费这是经典的重试放大问题。// 拉批 → 逐条处理 → 逐条确认状态表落库 ListMessage batch batchPuller.pull(batchSize, pollTimeoutMs); for (Message msg : batch) { try { process(msg); // 真正的业务处理 stateDao.markSuccess(msg.getMsgId()); // 本条已完成 } catch (Exception e) { stateDao.markFailed(msg.getMsgId()); // 记录失败进重试 retryService.push(msg); // 交给退避重试 } }注意这里提交的“位点”其实是本地消息状态表不是 MQ 客户端自己的 offset。为什么因为 MQ 的 offset 按分区维度整批提交业务层没法精确到单条。MPSP 的方案是让消费逻辑幂等地处理每一条重启后从 MQ 重新拉到的消息先查状态表已经 SUCCESS 的直接跳过。这样即使 MQ 侧发生重复投递业务层也只执行一次。3.3 幂等设计业务去重表 内存最近值缓存两层兜底幂等是整套平台里最不能省的一层。框架只能保证“至少一次”投递少一次都保证不了更别说精确一次。重复消费的源头主要有三个生产者重试导致同一条消息发多次、消费者处理成功后提交状态前崩溃导致重拉、MQ 自身在网络抖动时的重投。所以幂等必须放在业务侧用 msgId 做唯一键。// 幂等判断先查内存最近值再查库两层兜底 public boolean isDuplicate(String msgId) { if (recentCache.containsKey(msgId)) { return true; // 最近处理过直接跳过 } // 状态表里 msg_id 有唯一约束已存在且状态为 SUCCESS 则重复 return stateDao.countByMsgIdAndStatus(msgId, SUCCESS) 0; }内存缓存解决的是高频重复状态表唯一约束解决的是重启后的兜底。只做内存缓存会出问题——进程一重启缓存清空重复消息全部漏过去只查库又扛不住高并发下同一批消息的重复校验。两层叠起来单条消息的判重开销基本被内存缓存吸收只有缓存未命中才落库。4. 可靠性与参数调优重试、死信、限流要和线程池一起调才不翻车4.1 重试策略指数退避加抖动别用固定间隔固定间隔重试是新手最容易写出来的方案重试 5 次、每次间隔 3 秒。问题是如果下游数据库刚好在故障恢复窗口固定间隔会让所有失败消息在同一秒集中重试把刚恢复的数据库又打回去。指数退避加抖动是生产环境的标准做法。// 计算第 n 次重试的等待时间1s → 2s → 4s → 8s → ... 封顶 60s public static long nextBackoff(int retryCount) { long base 1000L; // 初始 1 秒 long delay base * (1L Math.min(retryCount, 6)); // 2^n 退避最多 64 秒 long jitter ThreadLocalRandom.current().nextLong(delay / 10); // 加 10% 抖动 return Math.min(delay jitter, 60_000L); // 绝对上限 60 秒 }为什么加抖动指数退避在没有抖动的场景下所有消息会在同一时间点打到下游形成浪涌。抖动打散了这个聚集效应。实际接入时retryCount要从消息 Header 里读而不是从内存里记因为进程重启后重试计数器会丢从 Header 读才能保持重试次数连续。4.2 死信队列给失败消息一张后悔药但死信也要有人消费重试次数用尽仍然失败的消息不能无限重试也不能直接丢弃要进死信队列。MPSP 里死信不投递给第三方系统只做两件事持久化存储 触发告警。这样后续排查问题时死信记录就是原始现场。// 重试次数达到上限进入死信并告警 if (msg.getRetryCount() maxRetry) { dlqSink.push(msg); // 写入死信存储 alertService.warn(dlq, msg); // 触发告警必须人工介入 return; }死信队列本身不复杂复杂的是很多人只写推送不写消费。死信积压到几万条时告警还没人数等发现时已经过了业务补偿窗口。所以死信也要挂一个消费逻辑周期性扫描对可以补偿的消息重新发起投递对无法补偿的消息做标记归档。死信不是终点只是一个带人工介入的缓冲区。4.3 限流与背压线程池饱和时选错策略消息直接丢Java 线程池的饱和策略有四种接入消息平台时选错一个就丢消息。默认的 AbortPolicy 在线程池满时会直接抛 RejectedExecutionException如果拉取线程没有捕获整批消息就丢了。DiscardPolicy 更离谱静默丢弃。消息处理场景里可靠的做法是 CallerRunsPolicy或者在前面加一层有界缓冲。// 工作线程池配置核心 8最大 12队列 2000饱和策略由调用方处理 ThreadPoolExecutor workers new ThreadPoolExecutor( 8, 12, 60, TimeUnit.SECONDS, new ArrayBlockingQueue(2000), new ThreadPoolExecutor.CallerRunsPolicy() );CallerRunsPolicy 的语义是“线程池满了任务返回给调用线程执行”。在 MPSP 里调用线程是拉取线程拉取线程会去执行业务方法相当于把压力传回拉取链路拉取线程被业务占住后自然不再向 MQ 拉取新消息。这个策略保证任务不丢只是吞吐暂时降下来。代价是拉取线程也在跑业务心跳可能受影响所以超时时间要留足够余量。4.4 一张参数表从吞吐优先到可靠性优先怎么调MPSP 的每个参数都不是独立存在的。线程池核心数决定基本吞吐最大线程数决定突发上限队列容量决定在途积压批量大小决定单批失败的影响面重试参数决定故障恢复速度。调参时先定单线程处理耗时再反推其他参数。参数默认值吞吐优先调法可靠性优先调法核心线程数CPU × 2处理耗时 IO 密集时再乘 2保持默认避免抢占业务线程最大线程数核心 × 1.5提到核心 × 2容忍突发和核心数一致宁可排队队列容量2000调大到 5000缓冲更多调小到 1000触发更早背压拉取批量200消息体小时调到 500保持 200缩小失败放大面拉取超时500ms可以降到 200ms保持 500ms减少空转最大重试5 次3 次快速失败快速人工介入8 次给下游更多恢复时间退避上限60s30s缩短整体延迟120s更保守这个表格不是拍脑袋填的是按处理耗时推出来的。单条消息处理耗时 30ms单线程每秒能处理约 33 条8 个核心线程理想吞吐约 260 条/秒队列 2000 相当于 7 秒左右的缓冲。如果你的场景是支付回调处理耗时可能到 200ms同样的参数组合吞吐就掉到 40 条/秒这时候要先优化单条处理链路而不是加线程。5. MPSP 落地避坑五个生产事故级别的常见问题5.1 无界队列导致内存被积压消息打爆现象线上 JVM 老年代持续上涨GC 频繁 Full GC最后 OOM 重启重启后消费位点落后巨大。原因有开发者为了“提升吞吐”把缓冲队列换成了无界队列。正常情况下吞吐确实上去了但下游某个服务抖动时消费速度跟不上生产速度消息无限堆积在堆内存里内存被打爆。无界队列还隐藏了背压信号MQ 端永远不知道消费端已经扛不住。解决缓冲队列必须用有界队列容量按期望吞吐 × 单条耗时 × 3估算满时阻塞拉取线程让压力传到 MQ 端做持久化积压而不是在 JVM 堆里积压。同时监控队列剩余容量低于 20% 就告警。5.2 批量提交位点放大了失败重试的影响面现象日志里显示某条消息处理失败后接下来几秒钟内前面已经成功处理过的几百条消息又被重新调用了一次业务侧出现大量“重复操作”告警。原因最开始图省事一个批次处理完后统一提交 MQ 位点。批次里有一条失败整个批次的位点都不提交下次拉取时整批重新推送。已经处理成功的 199 条消息被重复消费如果业务没做幂等就是一笔重复扣款级别的事故。解决批次只负责拉取不负责位点管理。每处理完一条先把本地消息状态表置为 SUCCESS再按消息粒度更新本地确认记录。MQ 位点仍然按批提交但由于业务层有状态表兜底重新拉取的消息会被幂等拦截掉。5.3 幂等只做内存缓存重启后重复消费现象某次服务发布重启后同一笔订单的待办通知被触发了两次用户连续收到两条一样的内容。原因内存幂等缓存只记录进程存活期间处理过的 msgId重启后缓存清空。MQ 在重启窗口内重投了旧消息缓存不命中查库逻辑又没接上重复消息就漏过去了。解决幂等必须两层内存缓存加速判重状态表唯一约束兜底。状态表里 msg_id 加唯一索引插入冲突直接视为重复。这样即使缓存重建期间有重复消息进来数据库这一层也能拦住。5.4 固定间隔重试在故障恢复时引发风暴现象下游数据库抖动 10 分钟恢复后第一分钟MPSP 瞬间打进来几千条重试消息数据库又被打出抖动形成二次故障。原因重试逻辑写成了“失败后固定等 5 秒再重试”所有失败消息都在同一时刻集中重试恢复窗口变成新的压力峰值。解决改成指数退避加抖动。第一次重试等 1 秒左右第二次 2 秒第三次 4 秒每次叠加 10% 的随机抖动。这样失败时间相近的消息退避之后的重试时刻会被自然分散开。5.5 死信队列只投不消积压成大坑现象死信表数据量一周涨了十几万条磁盘占用暴增查询越来越慢但没有任何系统告警。原因死信逻辑只写了“重试次数超限就投递到死信”没人写消费端。死信成了只进不出的黑洞越积越多。解决死信必须配套消费任务。每五分钟扫描一次死信表区分可补偿和不可补偿两类可补偿的重新投递原主题并重置 retryCount连续补偿失败超过三次就冻结报警不可补偿的做归档标注保留现场但不再占用活表空间。同时死信积压数量超过阈值必须告警这是人工介入的信号。6. 验证 MPSP 是否扛得住压测脚本与三个核心监控指标6.1 一个 50 行的简易压测脚本接入 MPSP 后别急着上线先用压测脚本把参数验证一遍。下面是一个按固定速率发送消息的简易工具能同时验证吞吐和积压行为。// 用法java LoadGen 4 200 表示 4 个发送线程每线程每秒 200 条 public class LoadGen { private static final AtomicBoolean stop new AtomicBoolean(false); public static void main(String[] args) throws InterruptedException { int sendThreads Integer.parseInt(args[0]); int ratePerProducer Integer.parseInt(args[1]); long intervalMs 1000L / ratePerProducer; CountDownLatch latch new CountDownLatch(sendThreads); for (int i 0; i sendThreads; i) { new Thread(() - { while (!stop.get()) { long t0 System.currentTimeMillis(); producer.send(buildMessage()); // 走 MPSP 生产者链路 long cost System.currentTimeMillis() - t0; StatRecorder.record(cost); // 记录发送耗时 long sleep intervalMs - (System.currentTimeMillis() - t0); if (sleep 0) Thread.sleep(sleep); // 按速率节流 } latch.countDown(); }).start(); } Thread.sleep(60_000L); // 跑 60 秒 stop.set(true); latch.await(); StatRecorder.print(); // 输出 TPS、P99、P999 } }脚本核心是间隔计算1000 / ratePerProducer是每秒可发送的间隔毫秒数减掉实际发送耗时后 sleep 剩余时间保证发送速率基本恒定。压测时先跑低速率比如 4 线程 × 200 条每秒观察消费是否稳定再把速率提到目标峰值的两倍观察缓冲队列和拒绝策略是否按预期工作。6.2 三个核心监控指标端到端延迟、积压量、重试率压测结束后的数据不能只看平均值要看三个指标端到端延迟的 P99、缓冲队列积压深度、消息重试率。P99 用当前时间 - produceTime计算能反映真实用户体验积压深度看缓冲队列剩余容量和状态表 PENDING 计数持续上涨说明消费能力不足重试率看 retryCount 分布正常情况下重试 0 次的消息占比要超过 99%持续偏低就要查下游资源和幂等逻辑。我一般会按这个顺序回归先量化单条消息的处理耗时再定线程数和队列容量最后调重试和死信参数绝不一次性全改。从那以后每次调整线程池或重试参数我都强制走一遍这套压测验证流程确认三个指标曲线全部平稳才敢合入线上。希望帮到你。本文还有配套的精品资源点击获取
简介这是一份基于 Java 的多协议服务处理器MPSP开源项目源码面向网络编程学习者与中高级 Java 开发工程师适合用来研究 TCP/IP、UDP、HTTP 等网络协议的并发处理与服务端架构。压缩包共 23 个文件其中 15 个 Java 源文件按 src/test 目录组织另含 Gradle 构建脚本、配置文件、gradlew 启动脚本及 README 说明整体仅 18KB结构紧凑导入工程后即可快速定位核心代码。已有 443 人学习下载。通过源码可系统掌握套接字编程、NIO 非阻塞 IO、线程池设计、协议解析流程以及工厂、单例、观察者等设计模式的工程落地并体现清晰的包结构与分层设计思路项目中的单元测试、CI 配置和日志监控选型也为构建健壮、可扩展的网络服务提供了完整的参考骨架适合进阶学习和二次开发。1. MPSP 消息处理平台一套能扛突发流量又不丢消息的 Java 异步骨架第一次接到消息处理需求时我的第一反应很简单消费者拉一批消息for 循环里挨个调用业务方法不就行了。真上线才发现这种裸写方式在流量小时没毛病流量一起来就暴露出三个致命伤——消息重复消费没人管、一批里只要一条失败就把整批拖垮、线程池参数全靠猜。MPSP 就是这样一套基于 Java 的轻量消息处理平台骨架它不替代 Kafka 或 RocketMQ只负责业务侧从拉取、分发、幂等到可靠重试的整条链路。适合正在写事件驱动模块、支付回调或异步任务的开发者拿来即用也看得见每个参数为什么存在。2. MPSP 的整体结构消息模型、线程模型与为什么不能裸写线程池2.1 消息模型Header 与 Body 分离状态字段就是排查问题的黑匣子MPSP 里所有消息都统一成同一个 Message 对象Header 和 Body 分开存放。Header 只放路由和状态信息Body 放业务 JSON。这样设计的好处很直白线上出问题时不需要把每条消息的 body 都拉出来翻一遍只看 Header 就能判断这条消息属于哪个主题、对应哪笔业务、重试了几次、是什么时候产生的。// 核心消息模型字段刻意精简 public class Message { private String msgId; // 全局唯一生成后不再变化幂等依赖它 private String topic; // 业务主题如 order_event private String bizId; // 业务主键顺序路由和去重都看它 private int retryCount; // 已重试次数超过阈值进死信 private long produceTime; // 生产时间戳端到端延迟靠它算 private String body; // 业务数据 JSON // getter / setter 省略 }注意 msgId 和 bizId 是两个东西。msgId 是消息自身的唯一编号一条消息重复投递 n 次msgId 都不变bizId 是业务侧的主键比如订单号。幂等去重用 msgId顺序路由用 bizId两者职责不能混。我见过有人拿订单号当 msgId 用同一个订单发两条不同事件时直接撞了唯一约束这种翻车在接入第一天就能遇到。2.2 线程模型单线程拉取 有界缓冲 工作线程池很多人一上来就把拉取和业务处理放在同一个线程里拉一条、处理一条、再拉一条。流量低时看不出问题一旦业务处理里有个外部 HTTP 调用耗时 500ms拉取线程就被卡住MQ 客户端的心跳超时消费组直接判定你离线触发 rebalance。这就是裸写消费逻辑最常见的线上事故。MPSP 把链路拆成三段一个拉取线程负责从 MQ 批量拿消息放到一个有界缓冲队列里工作线程池负责真正执行业务逻辑拉取线程和业务线程之间只通过缓冲队列通信互不阻塞。// 拉取线程把批量消息搬进有界缓冲缓冲满时阻塞拉取天然背压 public class PullWorker implements Runnable { private final BlockingQueueMessage buffer; // 有界缓冲默认 2000 private final ThreadPoolExecutor workers; // 业务工作线程池 Override public void run() { while (running) { // 拉取一批超时 500ms避免空转 ListMessage batch batchPuller.pull(batchSize, 500); for (Message msg : batch) { buffer.put(msg); // 缓冲满会阻塞在这里让 MQ 端自然积压 workers.execute(() - handle(msg)); // 丢给业务线程 } } } }这里的关键在buffer.put(msg)。有界队列满的时候拉取线程会阻塞在 put 上不再从 MQ 拉新消息。这比线程池的拒绝策略温和得多——拒绝策略是直接抛弃任务而阻塞是让压力顺着链路往回传由 MQ 端做真正的积压。buffer 容量就是整个系统允许的最大在途消息数一般按期望吞吐 × 单条处理耗时 × 2估算。2.3 为什么不用多消费者直接竞争顺序性由 bizId 路由保证也有人说既然有工作线程池我多开几个消费者客户端直接竞争消费不就行了技术上能跑但会引入两个问题第一同一分区的消息会被多个线程并发处理订单状态机这类强顺序场景直接乱套第二幂等去重逻辑分散在每个消费线程里重复消息可能同时到达去重形同虚设。MPSP 的做法是把并发收拢到缓冲队列之后再做一次路由。每个 bizId 的哈希值固定映射到一个工作线程同一个订单的后续事件永远落在同一个线程上顺序天然保证。// 按 bizId 哈希路由保证同一业务的事件始终由同一线程处理 int threadIndex Math.floorMod(msg.getBizId().hashCode(), workerCount);代价是线程之间可能负载不均某个大订单流量特别集中时单线程会成为瓶颈。但从生产实践看绝大多数业务的顺序要求都集中在单个业务主键内部这种路由方式比全局锁或者分区重排简单可靠得多。3. 接入 MPSP 的完整链路生产者、消费者与幂等去重3.1 生产者端事务内先落库再投递失败走本地补偿接入 MPSP 时生产者端最容易踩的坑是“先发消息再更新业务状态”。一旦发消息成功、更新数据库失败消费者就会拿到一个业务上并不存在的事件而且你还没法撤回。MPSP 的常规做法是本地消息表业务表和消息状态表在同一个事务里写入消息状态先置为 PENDING事务提交后再真正发送到 MQ发送成功才把状态改成 SENT。Transactional public boolean sendOrderEvent(OrderEvent event) { // 1. 业务表与消息状态表同事务写入msgId 由这里统一生成 String msgId UUID.randomUUID().toString().replace(-, ); messageStateDao.save(new MessageState(msgId, event.getOrderId(), PENDING)); // 2. 事务提交后发送到 MQ boolean ok mqProducer.send(buildMessage(msgId, event)); if (ok) { messageStateDao.updateStatus(msgId, SENT); } return ok; }注意这里是事务方法但 MQ 发送放在了事务内。严格来说事务未提交时 MQ 消息已经发出去了消费端如果恰好立刻拉取可能看不到业务数据。常见做法是让 MQ 发送在事务提交后执行用TransactionSynchronizationManager.registerSynchronization挂一个 afterCommit 回调。同时本地任务表里始终保留 PENDING 记录定时任务每隔几秒扫描一次把 PENDING 超过 5 秒的消息重新投递投递成功再更新状态。这套补偿机制才是可靠性兜底MQ 本身只保证不丢不保证时序。3.2 消费者端批量拉取、逐条确认不整批提交位点消费者端的第一个原则是批量拉取、逐条处理。批量拉取能降低 MQ 交互次数但提交位点时不能一批一起提交——一批 200 条消息里只要有一条处理失败整批滚动回去重新拉取其余 199 条都会被重复消费这是经典的重试放大问题。// 拉批 → 逐条处理 → 逐条确认状态表落库 ListMessage batch batchPuller.pull(batchSize, pollTimeoutMs); for (Message msg : batch) { try { process(msg); // 真正的业务处理 stateDao.markSuccess(msg.getMsgId()); // 本条已完成 } catch (Exception e) { stateDao.markFailed(msg.getMsgId()); // 记录失败进重试 retryService.push(msg); // 交给退避重试 } }注意这里提交的“位点”其实是本地消息状态表不是 MQ 客户端自己的 offset。为什么因为 MQ 的 offset 按分区维度整批提交业务层没法精确到单条。MPSP 的方案是让消费逻辑幂等地处理每一条重启后从 MQ 重新拉到的消息先查状态表已经 SUCCESS 的直接跳过。这样即使 MQ 侧发生重复投递业务层也只执行一次。3.3 幂等设计业务去重表 内存最近值缓存两层兜底幂等是整套平台里最不能省的一层。框架只能保证“至少一次”投递少一次都保证不了更别说精确一次。重复消费的源头主要有三个生产者重试导致同一条消息发多次、消费者处理成功后提交状态前崩溃导致重拉、MQ 自身在网络抖动时的重投。所以幂等必须放在业务侧用 msgId 做唯一键。// 幂等判断先查内存最近值再查库两层兜底 public boolean isDuplicate(String msgId) { if (recentCache.containsKey(msgId)) { return true; // 最近处理过直接跳过 } // 状态表里 msg_id 有唯一约束已存在且状态为 SUCCESS 则重复 return stateDao.countByMsgIdAndStatus(msgId, SUCCESS) 0; }内存缓存解决的是高频重复状态表唯一约束解决的是重启后的兜底。只做内存缓存会出问题——进程一重启缓存清空重复消息全部漏过去只查库又扛不住高并发下同一批消息的重复校验。两层叠起来单条消息的判重开销基本被内存缓存吸收只有缓存未命中才落库。4. 可靠性与参数调优重试、死信、限流要和线程池一起调才不翻车4.1 重试策略指数退避加抖动别用固定间隔固定间隔重试是新手最容易写出来的方案重试 5 次、每次间隔 3 秒。问题是如果下游数据库刚好在故障恢复窗口固定间隔会让所有失败消息在同一秒集中重试把刚恢复的数据库又打回去。指数退避加抖动是生产环境的标准做法。// 计算第 n 次重试的等待时间1s → 2s → 4s → 8s → ... 封顶 60s public static long nextBackoff(int retryCount) { long base 1000L; // 初始 1 秒 long delay base * (1L Math.min(retryCount, 6)); // 2^n 退避最多 64 秒 long jitter ThreadLocalRandom.current().nextLong(delay / 10); // 加 10% 抖动 return Math.min(delay jitter, 60_000L); // 绝对上限 60 秒 }为什么加抖动指数退避在没有抖动的场景下所有消息会在同一时间点打到下游形成浪涌。抖动打散了这个聚集效应。实际接入时retryCount要从消息 Header 里读而不是从内存里记因为进程重启后重试计数器会丢从 Header 读才能保持重试次数连续。4.2 死信队列给失败消息一张后悔药但死信也要有人消费重试次数用尽仍然失败的消息不能无限重试也不能直接丢弃要进死信队列。MPSP 里死信不投递给第三方系统只做两件事持久化存储 触发告警。这样后续排查问题时死信记录就是原始现场。// 重试次数达到上限进入死信并告警 if (msg.getRetryCount() maxRetry) { dlqSink.push(msg); // 写入死信存储 alertService.warn(dlq, msg); // 触发告警必须人工介入 return; }死信队列本身不复杂复杂的是很多人只写推送不写消费。死信积压到几万条时告警还没人数等发现时已经过了业务补偿窗口。所以死信也要挂一个消费逻辑周期性扫描对可以补偿的消息重新发起投递对无法补偿的消息做标记归档。死信不是终点只是一个带人工介入的缓冲区。4.3 限流与背压线程池饱和时选错策略消息直接丢Java 线程池的饱和策略有四种接入消息平台时选错一个就丢消息。默认的 AbortPolicy 在线程池满时会直接抛 RejectedExecutionException如果拉取线程没有捕获整批消息就丢了。DiscardPolicy 更离谱静默丢弃。消息处理场景里可靠的做法是 CallerRunsPolicy或者在前面加一层有界缓冲。// 工作线程池配置核心 8最大 12队列 2000饱和策略由调用方处理 ThreadPoolExecutor workers new ThreadPoolExecutor( 8, 12, 60, TimeUnit.SECONDS, new ArrayBlockingQueue(2000), new ThreadPoolExecutor.CallerRunsPolicy() );CallerRunsPolicy 的语义是“线程池满了任务返回给调用线程执行”。在 MPSP 里调用线程是拉取线程拉取线程会去执行业务方法相当于把压力传回拉取链路拉取线程被业务占住后自然不再向 MQ 拉取新消息。这个策略保证任务不丢只是吞吐暂时降下来。代价是拉取线程也在跑业务心跳可能受影响所以超时时间要留足够余量。4.4 一张参数表从吞吐优先到可靠性优先怎么调MPSP 的每个参数都不是独立存在的。线程池核心数决定基本吞吐最大线程数决定突发上限队列容量决定在途积压批量大小决定单批失败的影响面重试参数决定故障恢复速度。调参时先定单线程处理耗时再反推其他参数。参数默认值吞吐优先调法可靠性优先调法核心线程数CPU × 2处理耗时 IO 密集时再乘 2保持默认避免抢占业务线程最大线程数核心 × 1.5提到核心 × 2容忍突发和核心数一致宁可排队队列容量2000调大到 5000缓冲更多调小到 1000触发更早背压拉取批量200消息体小时调到 500保持 200缩小失败放大面拉取超时500ms可以降到 200ms保持 500ms减少空转最大重试5 次3 次快速失败快速人工介入8 次给下游更多恢复时间退避上限60s30s缩短整体延迟120s更保守这个表格不是拍脑袋填的是按处理耗时推出来的。单条消息处理耗时 30ms单线程每秒能处理约 33 条8 个核心线程理想吞吐约 260 条/秒队列 2000 相当于 7 秒左右的缓冲。如果你的场景是支付回调处理耗时可能到 200ms同样的参数组合吞吐就掉到 40 条/秒这时候要先优化单条处理链路而不是加线程。5. MPSP 落地避坑五个生产事故级别的常见问题5.1 无界队列导致内存被积压消息打爆现象线上 JVM 老年代持续上涨GC 频繁 Full GC最后 OOM 重启重启后消费位点落后巨大。原因有开发者为了“提升吞吐”把缓冲队列换成了无界队列。正常情况下吞吐确实上去了但下游某个服务抖动时消费速度跟不上生产速度消息无限堆积在堆内存里内存被打爆。无界队列还隐藏了背压信号MQ 端永远不知道消费端已经扛不住。解决缓冲队列必须用有界队列容量按期望吞吐 × 单条耗时 × 3估算满时阻塞拉取线程让压力传到 MQ 端做持久化积压而不是在 JVM 堆里积压。同时监控队列剩余容量低于 20% 就告警。5.2 批量提交位点放大了失败重试的影响面现象日志里显示某条消息处理失败后接下来几秒钟内前面已经成功处理过的几百条消息又被重新调用了一次业务侧出现大量“重复操作”告警。原因最开始图省事一个批次处理完后统一提交 MQ 位点。批次里有一条失败整个批次的位点都不提交下次拉取时整批重新推送。已经处理成功的 199 条消息被重复消费如果业务没做幂等就是一笔重复扣款级别的事故。解决批次只负责拉取不负责位点管理。每处理完一条先把本地消息状态表置为 SUCCESS再按消息粒度更新本地确认记录。MQ 位点仍然按批提交但由于业务层有状态表兜底重新拉取的消息会被幂等拦截掉。5.3 幂等只做内存缓存重启后重复消费现象某次服务发布重启后同一笔订单的待办通知被触发了两次用户连续收到两条一样的内容。原因内存幂等缓存只记录进程存活期间处理过的 msgId重启后缓存清空。MQ 在重启窗口内重投了旧消息缓存不命中查库逻辑又没接上重复消息就漏过去了。解决幂等必须两层内存缓存加速判重状态表唯一约束兜底。状态表里 msg_id 加唯一索引插入冲突直接视为重复。这样即使缓存重建期间有重复消息进来数据库这一层也能拦住。5.4 固定间隔重试在故障恢复时引发风暴现象下游数据库抖动 10 分钟恢复后第一分钟MPSP 瞬间打进来几千条重试消息数据库又被打出抖动形成二次故障。原因重试逻辑写成了“失败后固定等 5 秒再重试”所有失败消息都在同一时刻集中重试恢复窗口变成新的压力峰值。解决改成指数退避加抖动。第一次重试等 1 秒左右第二次 2 秒第三次 4 秒每次叠加 10% 的随机抖动。这样失败时间相近的消息退避之后的重试时刻会被自然分散开。5.5 死信队列只投不消积压成大坑现象死信表数据量一周涨了十几万条磁盘占用暴增查询越来越慢但没有任何系统告警。原因死信逻辑只写了“重试次数超限就投递到死信”没人写消费端。死信成了只进不出的黑洞越积越多。解决死信必须配套消费任务。每五分钟扫描一次死信表区分可补偿和不可补偿两类可补偿的重新投递原主题并重置 retryCount连续补偿失败超过三次就冻结报警不可补偿的做归档标注保留现场但不再占用活表空间。同时死信积压数量超过阈值必须告警这是人工介入的信号。6. 验证 MPSP 是否扛得住压测脚本与三个核心监控指标6.1 一个 50 行的简易压测脚本接入 MPSP 后别急着上线先用压测脚本把参数验证一遍。下面是一个按固定速率发送消息的简易工具能同时验证吞吐和积压行为。// 用法java LoadGen 4 200 表示 4 个发送线程每线程每秒 200 条 public class LoadGen { private static final AtomicBoolean stop new AtomicBoolean(false); public static void main(String[] args) throws InterruptedException { int sendThreads Integer.parseInt(args[0]); int ratePerProducer Integer.parseInt(args[1]); long intervalMs 1000L / ratePerProducer; CountDownLatch latch new CountDownLatch(sendThreads); for (int i 0; i sendThreads; i) { new Thread(() - { while (!stop.get()) { long t0 System.currentTimeMillis(); producer.send(buildMessage()); // 走 MPSP 生产者链路 long cost System.currentTimeMillis() - t0; StatRecorder.record(cost); // 记录发送耗时 long sleep intervalMs - (System.currentTimeMillis() - t0); if (sleep 0) Thread.sleep(sleep); // 按速率节流 } latch.countDown(); }).start(); } Thread.sleep(60_000L); // 跑 60 秒 stop.set(true); latch.await(); StatRecorder.print(); // 输出 TPS、P99、P999 } }脚本核心是间隔计算1000 / ratePerProducer是每秒可发送的间隔毫秒数减掉实际发送耗时后 sleep 剩余时间保证发送速率基本恒定。压测时先跑低速率比如 4 线程 × 200 条每秒观察消费是否稳定再把速率提到目标峰值的两倍观察缓冲队列和拒绝策略是否按预期工作。6.2 三个核心监控指标端到端延迟、积压量、重试率压测结束后的数据不能只看平均值要看三个指标端到端延迟的 P99、缓冲队列积压深度、消息重试率。P99 用当前时间 - produceTime计算能反映真实用户体验积压深度看缓冲队列剩余容量和状态表 PENDING 计数持续上涨说明消费能力不足重试率看 retryCount 分布正常情况下重试 0 次的消息占比要超过 99%持续偏低就要查下游资源和幂等逻辑。我一般会按这个顺序回归先量化单条消息的处理耗时再定线程数和队列容量最后调重试和死信参数绝不一次性全改。从那以后每次调整线程池或重试参数我都强制走一遍这套压测验证流程确认三个指标曲线全部平稳才敢合入线上。希望帮到你。本文还有配套的精品资源点击获取
