企业官网建设流程全解析

我先把背景交代清楚。这是一个内部项目，代号就叫“模拟项目X”，目标是用 protobuf 做消息序列化，自己实现一套轻量级的发布订阅消息队列，不依赖外部中间件。第一版是同步模型，跑完压测之后数据很难看，publish 被订阅者回调拖住，吞吐上不去，延迟也抖得厉害。于是有了这一篇的内容：在发布订阅链路中引入异步线程池，把发布操作和订阅者的业务处理彻底解耦。这篇文章会从为什么需要异步、线程池怎么设计、怎么和 protobuf 消息链路对接、压测暴露出哪些问题这几个角度完整复盘，适合正在自己写消息中间件、或者想理解线程池在真实项目里怎么用的朋友。

1. 同步版本的瓶颈数据：publish 不该为订阅者买单

先看第一版的设计。发布者调用 publish(topic, message)，内部同步遍历该 topic 下的所有订阅者，逐个执行订阅者注册的回调函数。整个调用链是：protobuf 序列化 -> 路由查找 -> for 循环调用订阅者回调 -> 回调内部自行反序列化并执行业务逻辑 -> publish 返回。

表面上看没什么问题，消息队列本来不就是这么转发消息的吗？但压测数据一出来就发现问题了。模拟场景是单个 topic 挂 10 个订阅者，每个订阅者的回调里做一次 protobuf 反序列化，加一次简单的内存状态更新，模拟耗时大约 2 毫秒到 5 毫秒不等。压测工具以每秒 100 条的速率持续发布消息，结果 publish 的平均耗时达到了 32 毫秒，P99 直接冲到 68 毫秒。也就是说，发布者线程被订阅者拖住了，publish 的延迟等于所有订阅者回调执行时间之和。

1.1 同步模型的问题本质

发布订阅模型天然是 1 对 N 的扇出结构。同步执行意味着 publish 方法结束时，所有订阅者必须已经处理完这条消息。哪个订阅者慢，发布者就要陪它一起等。要命的是很多订阅者的回调逻辑不在你的控制范围内——有人写日志、有人写库、有人调外部接口，任何一个出现抖动，发布端的 RT 就会跟着抖动。更糟糕的是，如果某个订阅者回调里出现死循环或者长时间阻塞，所有发布者线程都会被堵住，整个队列陷入不可用状态。

同步模型在订阅者数量少、回调逻辑稳定的情况下还能凑合用，但只要订阅者数量上到 5 个以上，或者某个回调变慢，发布端延迟就会迅速劣化。这是结构性问题，靠优化单个回调的性能解决不了。

1.2 为什么不能简单加线程

有人会说，那就让每个订阅者单独开一个线程去执行。思路对了一半，但每个订阅者单独开线程意味着每来一条消息都可能创建一个线程，线程的创建销毁本身就是巨大的开销，而且线程数量不受控，高并发下内存和上下文切换成本会直接榨干 CPU。更麻烦的是，订阅者的注册和取消是动态的，没法预分配固定线程数。

所以正确方向不是“一个订阅者一个线程”，而是把整个回调执行过程变成任务投递过程。publish 只需要完成消息的路由计算和序列化工作，把执行回调这一步丢给一个共享的线程池去完成。这就是异步操作的第一层解耦。

1.3 异步化要付出的代价

当然，异步不是免费的。publish 从同步等待所有订阅者完成，变成只保证消息被正确投递到线程池，这意味着消息的处理结果无法直接在发布端拿到。如果业务场景需要知道“这条消息有没有被正确处理”，就得另外做回执、确认、死信重投的机制。另一个代价是消息的处理顺序不再天然和发布顺序一致：多线程并发执行订阅者回调，先进入队列的任务不一定先完成。对于大多数消息推送类场景这完全不是问题，但对于某些有状态订阅者（比如累计计数、按序写入），就必须做顺序保证的设计，这个后面会专门说。

2. 线程池设计：任务封装、队列选型与线程数量估算

确定要用线程池之后，接下来的问题是怎么设计这个线程池。我没有直接用 JUC 默认的 Executors.newFixedThreadPool，而是自定义了 ThreadPoolExecutor 的核心参数。原因后面展开，先说每个部分是怎么定的。

2.1 任务封装：把事件变成可执行单元

线程池里跑的东西不能是裸的 Runnable，那样没法带上下文。我的任务是这么封装的：

// 任务封装示例 public class DispatchTask implements Runnable { private final String topic; private final ByteString payload; // protobuf 序列化后的字节 private final Subscriber subscriber; // 目标订阅者 private final long enqueueTime; // 入队时间戳 private final long seqId; // 消息序号，用于顺序控制 public DispatchTask(String topic, ByteString payload, Subscriber subscriber, long seqId) { this.topic = topic; this.payload = payload; this.subscriber = subscriber; this.seqId = seqId; this.enqueueTime = System.nanoTime(); } @Override public void run() { // 订阅者回调里做反序列化和业务处理 subscriber.getCallback().onMessage(topic, payload); } }

这里的 payload 用 ByteString 而不是原始 byte[]，是因为 protobuf 生成的类直接支持 ByteString 类型，避免在投递过程中反复拷贝。消息在发布端完成序列化，订阅者拿到字节数据后自行 ParseFrom，这保证了发布线程池和订阅线程池之间的数据传递是只读的、无锁的。

2.2 队列选型：有界队列是底线

线程池内部的任务队列，我选择了有界的 LinkedBlockingQueue，容量设置为 20000。为什么不选无界队列？无界队列有两个问题：一是内存不可控，如果订阅者处理速度跟不上发布速度，任务无限堆积，最终 OOM；二是线程池在任务堆积时不会创建更多线程，因为线程数的扩容条件之一是队列满了，无界队列永远不会触达这个条件，导致你配置的最大线程数形同虚设。

有界队列搭配的拒绝策略也很关键。我一开始用的是 AbortPolicy（直接抛异常），压测的时候发现发布端会间歇性出现 RejectedExecutionException。后来改成了 CallerRunsPolicy，即队列满时由调用者线程直接执行该任务。这样做的逻辑是：队列已满说明订阅者的消费能力已经到顶，追加任务只会加剧堆积，不如让发布者自己执行一个任务，天然起到背压作用。

有界队列 + CallerRunsPolicy 的组合，是消息队列场景下的稳妥选择。它不丢消息，又能通过“发布者线程亲自执行”的方式把压力反向传导给发布端。

2.3 线程数量怎么定：先估算，再压测修正

线程数量不能用拍脑袋的方式定。我用了标准的估算公式，再结合压测结果修正。这个项目的消息处理是混合负载：有 CPU 计算（protobuf 反序列化、路由匹配），有 IO 操作（订阅者写日志、写缓存），所以不能按纯 CPU 密集型的N+1来定。

估算公式是N = C * U * (1 + W/C)，其中 C 是 CPU 核心数，U 是目标 CPU 利用率，W 是等待时间，C 是计算时间。我的机器是 8 核，目标利用率 0.8，订阅者回调里 IO 和计算大概是 6：4 的关系，代入公式就是8 * 0.8 * (1 + 1.5) = 16。所以核心线程数定在 16，最大线程数定在 32，队列容量 20000。

这只是起点。实际压测显示 16 个核心线程在纯消息转发场景下 CPU 利用率只有 50% 左右，因为大部分时间线程都在等待任务而阻塞；但当订阅者回调里加入真实的日志写入和缓存更新后，16 个线程很快就不够用了，队列深度持续上涨。最终我把核心线程数调整到了 24，效果最均衡。

2.4 线程池生命周期管理的坑

这个项目刚开始有个很隐蔽的问题：关机时线程池一直无法正常退出。原因是任务里订阅者的回调可能会再调用 publish 方法，而 publish 又依赖同一个线程池，形成依赖循环。调用 shutdown() 之后，线程池要等待所有任务执行完毕，但队列里又不断有新任务进来，关闭过程永远完不成。

解决办法是给线程池包一层管理器，提供两个方法：stopConsuming() 先把消息入口关闭，不再接受新的发布请求，再调用 shutdown() 等待已有任务排空。另外，所有任务跑完之后统一调用 awaitTermination 并设置超时时间，超时后强制 shutdownNow()，避免进程卡死在退出阶段。

3. 与 protobuf 消息链路对接：从字节流到任务投递的完整时序

线程池本身只是执行引擎，真正麻烦的是怎么把 protobuf 的序列化消息链路和异步投递流程顺畅地接在一起。这块我拆成三个节点来设计：报文解帧、路由查找、任务提交。

3.1 报文解帧：长度前缀才是亲爹

消息到达之后，首先面对的是 TCP 流式传输的粘包问题。protobuf 本身不解决消息边界，所以传输层必须自己做报文帧格式。我采用的是最经典的长度前缀方案：4 字节的大端整数标记 payload 长度，后面跟着 protobuf 序列化后的字节数组。解帧逻辑如下：

// 消息解帧示例（简化版） public static ByteBuffer decodeFrame(byte[] data) { if (data.length < 4) { return null; // 数据不完整，继续等 } int length = ((data[0] & 0xFF) << 24) | ((data[1] & 0xFF) << 16) | ((data[2] & 0xFF) << 8) | (data[3] & 0xFF); if (data.length < length + 4) { return null; // 帧未收满，交给后续累积 } return ByteBuffer.wrap(data, 4, length); }

解码器把完整消息 ByteString 提取出来后，再交给上层解析出 MessageEnvelope，里面包含消息类型和 topic 字段。

3.2 路由表：并发安全的订阅者注册中心

路由表的结构是ConcurrentHashMap<String, CopyOnWriteArrayList<Subscriber>>。主题到订阅者列表的映射，key 是 topic 字符串，value 是订阅者列表。选择 CopyOnWriteArrayList 是因为订阅关系变更的频率极低，但遍历频率极高——每来一条消息都要遍历一次列表。写时复制正好把遍历开销降到最低，读操作完全无锁。

路由查找在所有环节之前执行，找到订阅者列表之后，如果列表为空，消息直接丢弃或者走死信逻辑。这里注意一点：发布订阅模型里，消息不要求订阅者预先存在，所以消息来了没有订阅者是正常情况，不应该算报错。

3.3 任务提交链路：一次完整投递的代码级拆解

拿到订阅者列表之后，publish 方法的核心动作就变成了依次把 DispatchTask 提交给线程池。完整流程是这样的：

// publish 方法核心逻辑 public void publish(String topic, Message message) { ByteString payload = message.toByteString(); List<Subscriber> subscribers = topicRouter.get(topic); if (subscribers == null || subscribers.isEmpty()) { return; } long seqId = seqGenerator.incrementAndGet(); for (Subscriber subscriber : subscribers) { DispatchTask task = new DispatchTask(topic, payload, subscriber, seqId); try { executor.execute(task); } catch (RejectedExecutionException e) { // 队列满时的兜底策略：异步写失败日志，同时触发背压指标 metrics.recordRejectedTask(topic); publishTaskSyncFallback(task); } } }

注意这里发送给所有订阅者的是同一个 ByteString 对象，不拷贝。反序列化在每个订阅者的回调里各自执行，所以不存在共享数据被并发修改的问题。每个订阅者拿到的都是同一个不可变字节数组，谁先谁后不影响正确性。

3.4 处理结果的回收

任务在线程池跑完了，结果怎么回收？我没有用 Future，因为 fire-and-forget 才是消息队列的主要模式，用 Future 会导致调用方阻塞等待，那又回到同步模型了。

我的做法是：任务内部把执行结果写入一个无锁环形缓冲统计区，包括执行耗时、成功/失败、反序列化耗时等指标。外部有一个定时任务每隔 5 秒扫描统计区，把数据聚合成 QPS、P99 延迟、线程利用率等监控指标。这样既不影响主流程，又能拿到异步执行的可观测数据，排查问题的时候心里有底。

4. 流量冲击下暴露的问题：乱序、饥饿与堆积的排查记录

写这部分之前我得先说明，下面这三个问题不是我从文档里看来的，是真的在压测阶段一个个踩出来的。如果你也在做类似的异步消息队列，大概率会碰到其中某个。

4.1 订阅者回调里二次发布导致的线程饥饿

压测到第 12 个小时的时候，发现一个诡异的现象：发布端延迟缓慢上升，从平均 3 毫秒一路涨到 300 多毫秒，线程池的 activeCount 一直是满的，但队列深度并没有明显涨。用 jstack 看线程栈，发现所有线程都在执行某个订阅者回调，而这个回调内部又调用了 publish 方法，publish 又往同一个线程池提交任务。由于线程池所有线程都在执行回调，新提交的任务只能排队，而回调本身又在等新任务执行完成后才返回，于是死锁就出现了——线程饥饿导致整个池子被堵死。

修复方案很直接：引入两个独立的线程池。一个负责消息接收和任务分发，称为分发线程池；另一个负责执行订阅者回调，称为业务线程池。publish 只往业务线程池投递任务，订阅者回调里无论做什么都不会影响分发线程池的线程。这条设计准则后来写进了项目规范：任何回调处理中严禁向同一个业务线程池重新提交任务。

4.2 单订阅者消息乱序：序号器与保序队列

乱序问题是在启动多个消费者进程后暴露的。有一个订阅者的业务逻辑是统计实时在线的用户数，每次收到新消息都要基于上一轮的统计结果做增量更新。异步化之后，两条先后到达的消息可能被两个线程同时执行，后一条先完成，前一条后完成，统计结果就错了。这是典型的“消息处理顺序”和“消息到达顺序”不一致导致的。

设计方案是在消息进入队列前为每个订阅者维护一个序号器，给消息编号，订阅者的线程池按序号顺序处理，同一时刻只允许一个线程执行该订阅者的回调。具体是实现一个串行调度器：为每个订阅者分配一个独立的小队列，业务线程池从这些队列里逐个取任务执行，保证同一订阅者的任务严格按提交顺序执行。代价是单个订阅者的吞吐会有上限（单线程执行），但我们这里的场景本来就是低并发高可靠，完全可接受。

4.3 突发流量打满队列：拒绝率与背压配合

第三次压测用了更大规模的流量，每秒发布 3000 条消息到 20 个订阅者。因为每个订阅者的反序列化和业务处理都需要时间，整体消费速率一度低于生产速率，有界队列很快被打满，拒绝策略被触发。开始是 AbortPolicy，日志里频繁出现RejectedExecutionException，而且由于任务提交是在 publish 循环中，一个 topic 下的 20 个订阅者会排队提交，前面几个提交成功了，后面几个可能就失败了，造成同一批消息部分订阅者收到、部分没收到。

改成 CallerRunsPolicy 之后，发布者线程被迫参与执行任务，生产速率立即被消费速率钳制，队列深度稳定在 18000 左右不再上涨。同时我加了背压信号——当队列深度超过阈值的 80%，发布接口直接抛出一个 BusyException 通知上游。这比让消息无限堆积更健康：用户能明确感知到系统到达瓶颈，而不是被一大堆排队消息掩盖。

4.4 另一个隐患：共享 ByteString 的复用风险

最后说一个和前三个问题不在一个量级、但很容易犯的错误。因为所有订阅者共享同一个 ByteString 对象，如果某个订阅者不小心把 ByteString 的内容转成了可变 byte[] 修改，其他订阅者拿到的数据就会错乱。protobuf 的 ByteString 设计上是不可变的，但通过toByteArray()拿到的新数组就不会影响原对象。关键是不要在回调里直接修改从消息体里 getByte() 返回的底层数组，我在代码审查时发现过两次这种情况，最后在回调入口强制规范：所有反序列化统一使用message.toBuilder()生成的副本，不允许直接修改原始消息对象。

5. 压测结果对比与参数调整记录

异步化改造完成之后，我对同一套压测场景重新跑了一遍，数据对比很明显。首版同步模型在 10 个订阅者场景下的 publish 平均耗时是 32 毫秒，P99 68 毫秒，吞吐量被压在每秒 200 条以下。异步化之后，发布端 publish 平均耗时降到 0.8 毫秒，P99 约 2.5 毫秒，因为 publish 只需要完成序列化加路由加任务提交三个动作，剩下的全是线程池的事。整体吞吐在 8 核机器上跑到每秒 2000 条仍有余量，主要瓶颈变成了订阅者的业务逻辑，而不是队列本身。

有个参数值得一提：线程池核心线程从 16 调到 24 的过程不是一蹴而就的。我做了三组对照压测，16 线程、24 线程、32 线程，结果很有意思。16 线程时队列深度上涨明显，说明消费速度跟不上；24 线程时队列深度平稳，P99 最优；32 线程虽然消费速度更快，但线程切换成本增加，CPU 总占用反而高了 15%，P99 劣化到接近 16 线程的水平。这说明线程数不是越多越好，要踩在“队列不堆积”和“CPU 不过载”的交汇点上。

线程池的监控指标，我整理了一份清单，后续维护时依赖这些数据做判断：

这套指标现在已经成了发布订阅队列的常驻监控项。每次改动线程池参数之后，我都会先跑 5 分钟的小流量压测看这些指标的形状，再决定要不要继续调。

还有两个细节值得分享。第一个是线程池线程名的设置，我在构造 ThreadPoolExecutor 时用自定义 ThreadFactory，为每个线程命名成mq-biz-1、mq-biz-2这种格式。这个习惯帮我省了太多排查时间，jstack 抓下来一眼就能分清哪些线程在跑消息回调，哪些线程在做网络 IO。第二个是任务执行结果的异常捕获，所有订阅者回调的外层统一包一层 try/catch，保证单个订阅者抛异常不会导致线程池线程被污染，也不会影响同批其他订阅者的执行。异常会被打进单独的错误日志，再按定时任务汇总抛出。

这个项目做到这里，异步线程池部分算是稳定了。回想一下从同步模型到异步模型的改造过程，最大的收获可能不是代码本身，而是理解了“解耦”在消息系统中的含义：发布者只负责把消息正确地交给执行引擎，而不负责消息最终被执行完。这份职责边界清晰之后，很多设计决策都变得简单了。如果你正好也在做类似的消息队列或者异步系统，希望这篇对你有帮助。