ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Zipkin整合RabbitMQ实现链路追踪异步收集与削峰

Zipkin整合RabbitMQ实现链路追踪异步收集与削峰

链路追踪接进去容易,扛住生产流量难。Zipkin 这套东西的默认玩法是业务进程里起一个异步线程,把 span 攒一攒,通过 HTTP 直接甩给 collector。压测的时候看不出毛病,等流量真上来、依赖方偶尔抖一抖,你就会发现:业务线程池里冒出几个卡在 socket 写上的线程,collector 那边一个 Full GC,这边的请求耗时跟着起跳。这篇就聊我实际落地过的方案——Zipkin 整合 RabbitMQ 实现异步数据收集,把 trace 数据的上报通道从同步 HTTP 换成消息队列,让链路数据自己慢慢排队、慢慢消化,业务该跑多快跑多快。内容会覆盖为什么要这么改、数据通路怎么设计、Zipkin Server 侧怎么打开 RabbitMQ 接收开关、业务侧三种接入姿势的完整代码、批量与队列参数怎么算、以及与业务 MQ 共存时怎么让 trace 不断链。适合已经把链路追踪跑起来、正在被上报开销或数据丢失问题困扰的同学,也适合刚准备接入、想一步到位选对方案的同行,Java 基础和 RabbitMQ 基本操作能看懂就行。

1. 为什么把链路数据从 HTTP 换成消息队列

1.1 默认上报在压测里暴露的三个问题

Zipkin 官方 SDK 的默认上报通道是 HTTP POST 到 collector 的/api/v2/spans。这个设计在开发环境、小流量环境完全够用,代码量少、不用额外维护中间件。问题全在流量上来之后集中爆发。

第一个问题是连接建立成本。上报用的是独立线程池,但每次 flush 都要走一遍建连或者从连接池借连接,collector 前面通常还挂着网关和负载均衡。业务侧 QPS 一上去,上报线程池会被打满,队列里的 span 越积越多,最后触发丢弃策略。你看着日志里飘过的“reporter dropped N spans”,就明白数据已经不全了。链路数据不全比没有更难受,因为你会拿残缺的调用树去分析耗时,结论还容易反向。

第二个问题是故障传导。链路追踪是旁路系统,理论上不该影响主链路。但 HTTP 上报是业务进程主动发起的网络调用,collector 一旦出现响应变慢、超时,上报线程就会卡住,异步队列迅速堆满,接着开始丢数据,再严重一点内存都跟着涨。我见过最典型的一次是 collector 端磁盘写满,写入阻塞,业务侧上报线程全卡在 read timeout 上,最后业务服务的 GC 时间和 CPU 都被拖了一截——明明是旁路功能,硬生生拖成了主链路故障。

第三个问题是削峰能力缺失。真实流量从来不是平的,大促、定时任务集中触发、上游批量推送,都会在几秒内把 QPS 顶到平时的好几倍。HTTP 上报是“有多少发多少”,collector 的处理能力是固定的,峰值一来要么丢数据,要么被打挂。而链路数据的价值恰恰在异常时段最高,平时你根本不需要看 trace,出问题的时候才发现最想看的那段时间数据没了。

1.2 RabbitMQ 这段通路的真实收益与代价

把 HTTP 换成 RabbitMQ,本质上是给上报链路加了一个削峰缓冲层。业务进程只负责把 span 序列化后扔进队列,扔完就走,不关心谁消费、什么时候消费;collector 按自己的节奏拉取,处理不过来就让消息在队列里待着。这一层解耦带来三个直接收益:上报耗时从“网络往返”降到“一次本地 publish”,业务线程基本感知不到;collector 抖动不再影响业务,最坏情况是队列堆积、数据延迟,而不是业务卡顿;流量峰值被队列吸收,collector 可以按平均吞吐而不是峰值吞吐来规划容量。

但必须说清楚代价,不然容易踩坑。第一,多了一个必须维护的组件,RabbitMQ 本身的可用性、磁盘、内存告警都要纳入监控,链路追踪的可用性从此不再只取决于 Zipkin。第二,数据延迟变成常态,正常情况下延迟可能只有几十毫秒到几秒,积压时可能到几分钟,排查实时问题时要意识到“你看到的是几分钟前的世界”。第三,消息体大小和批量策略需要重新调,HTTP 那套“攒够 5MB 再发”的默认参数直接搬到 AMQP 上会把 broker 的帧切得很碎,性能和内存都不好看。第四,确认机制要重新设计,链路数据到底要不要保证不丢?我的经验是不要,把它当可丢弃数据对待,用容量上限和溢出策略兜底,比开一堆持久化和 confirm 更划算。

注意:链路追踪数据属于“丢了不影响业务正确性”的观测数据。任何为了保证 100% 投递而引入的同步等待、消息持久化、多重确认,都会把旁路系统的成本转嫁回主链路,属于典型的过度设计。

2. 数据通路设计与关键约定

2.1 从 SDK 到 Collector 的三段式结构

整条通路可以拆成三段:生产端(业务进程内的 Reporter)→ 缓冲端(RabbitMQ 队列)→ 消费端(Zipkin Server 的 RabbitMQ Collector)。生产端不做任何网络请求,只做两件事——把zipkin2.Span序列化成字节数组,然后 publish 到指定队列。缓冲端就是一个普通队列,可以配长度上限和溢出策略。消费端是 Zipkin Server 内置的 collector 模块,它启动时会创建一个消费者,从队列里批量拉取消息,用SpanBytesDecoder自动识别编码格式并解码,然后走和 HTTP collector 完全一样的后续流程(存储、索引、查询)。

这个结构的关键在于编解码的约定必须两端一致。Zipkin 的 collector 支持 JSON v2、Proto3、Thrift 三种编码,它会通过SpanBytesDecoderDetector自动探测,所以理论上生产端用哪种都行。但我建议统一用 JSON v2:可读性好,出问题时rabbitmqadmin get抓一条消息出来肉眼就能看懂内容,排查成本低;代价是体积比 Proto3 大 30% 到 50%。如果你的 span 量级很大、带宽和存储敏感,再切 Proto3。

另一个容易被忽略的点是消息的粒度。一条消息里可以只放一个 span,也可以放一批 span(JSON 数组或 Proto3 的ListOfSpans)。我强烈建议批量,原因下面算数据量的时候会详细说,简单讲就是:消息条数直接决定 broker 的入队/出队压力和 confirm 开销,而 span 的条数是你控制不了的。批量把“消息数”从“span 数”降一到两个数量级。

2.2 队列名、编码格式与消息形态

Zipkin Server 的 RabbitMQ collector 有一个默认队列名zipkin,可以通过环境变量改。我的建议是不要用默认名,改成带环境前缀的名字,比如trace.zipkin.prod、trace.zipkin.staging。原因很实际:多套环境共用一个 RabbitMQ 集群是常态,一旦有人本地调试时连错 vhost 或者队列名写错,数据就会串。加前缀之后,队列列表一眼能看出归属,出问题也好定位。

消息形态上,如果走批量 + JSON v2,那么一条消息的 body 就是一个 JSON 数组,形如[{"traceId":"...","id":"...","name":"get /order","timestamp":...},...]。content_type建议设置成application/json,虽然 collector 不依赖这个字段做判断,但运维排查和抓包工具看着舒服。delivery_mode设成 1(非持久),理由前面说过,链路数据不值得为持久化付出磁盘 fsync 的代价。

还有一个细节:队列不要设置太多消费者抢占。Zipkin 的 collector 在消费时是按批处理的,多个 collector 实例同时消费同一个队列会平分消息,这本身没问题(横向扩展就该这样),但如果你在同一台机器上起了多个 collector 并且 concurrency 开得很大,反而会因为频繁的 ack 和 prefetch 调整导致吞吐下降。经验值是单实例 concurrency 控制在 2 到 4,实例数按 CPU 和存储写入能力扩。

2.3 三种接入方式的横向对比

业务侧把 Reporter 从 HTTP 换成 AMQP,历史上存在几种做法,差异挺大,先摆一张表看清楚再动手。

接入方式适用版本优点缺点推荐度
spring.zipkin.sender.type=rabbitSpring Cloud Sleuth 2.x一行配置,零代码只支持 JSON/默认编码,队列名不易定制,版本耦合深中
手工装配AsyncReporter+RabbitMQSenderBrave / Sleuth 2.x / 部分 3.x参数完全可控,编码、队列、批量都能调需要自己管 Channel 生命周期高
自定义SpanHandlerSpring Boot 3.x + Micrometer Tracing适配新版本,能和其他 handler 组合需要绕开自动配置,覆盖默认 handler高
自研 Reporter 直接发 JSON任意版本不依赖 zipkin-reporter 的 AMQP 模块要自己处理批量、重试、线程模型低

第三种和第二种是本文的重点。第一种虽然最省事,但它在 Sleuth 后续版本里已经不是主推路径,新项目不建议押注;第四种除非你有非常特殊的定制需求,否则纯属重复造轮子,AsyncReporter已经把批量攒包、定时 flush、队列满丢弃这些逻辑写得很扎实了。

3. 环境搭建与 Zipkin Server 侧配置

3.1 版本组合与依赖清单

先明确服务端。Zipkin Server 从 2.x 开始就是个 fat jar,RabbitMQ collector 是内置模块,不需要额外装插件,只要启动时把连接信息通过环境变量传进去,它就会自动启用。这一点很多人不知道,还在到处找“怎么给 zipkin 装 rabbitmq 插件”,其实根本不用。

客户端这边,依赖分两块:追踪框架(Brave 或 Micrometer Tracing)和上报通道(zipkin-sender-amqp-client)。后者是官方维护的 AMQP sender 实现,基于 RabbitMQ 的 Java client,坐标是io.zipkin.reporter2:zipkin-sender-amqp-client,版本要和zipkin-reporter主版本对齐。如果你用的是 Spring Cloud Sleuth,spring-cloud-sleuth-zipkin已经把 reporter 带进来了,只需要额外加 AMQP sender 依赖并把spring.zipkin.sender.type设成rabbit。

版本上有个坑必须提:Spring Boot 3 之后 Sleuth 退场,换成 Micrometer Tracing + Brave/OTel 桥接,spring.zipkin.sender.type这个配置项对 RabbitMQ 的支持没有了。management.zipkin.tracing.endpoint只认 HTTP 地址。所以升到 Boot 3 的项目,必须走自定义 Reporter 或自定义 SpanHandler 的路子,别照着老文章抄配置,抄完只会发现数据一条都没进队列。我在一个模拟项目里就因为没注意这点,白折腾了半天。

3.2 让 Collector 打开 RabbitMQ 接收开关

Zipkin Server 侧的配置全部通过环境变量走,核心就几个:

环境变量作用建议值
RABBIT_URI完整连接串,优先级最高amqp://user:pass@host:5672/vhost
RABBIT_ADDRESSES只给地址,用户名密码另配多节点用逗号分隔
RABBIT_QUEUE消费队列名trace.zipkin.prod
RABBIT_CONCURRENCY单实例消费者并发数2 到 4
RABBIT_CONNECTION_TIMEOUT建连超时(毫秒)6000
RABBIT_VIRTUAL_HOSTvhost,URI 里带过就不用设与业务隔离的独立 vhost

用 Docker 起的话大概长这样:

docker run -d --name zipkin \ -p 9411:9411 \ -e RABBIT_URI='amqp://trace_user:trace_pass@mq-host:5672/trace' \ -e RABBIT_QUEUE='trace.zipkin.prod' \ -e RABBIT_CONCURRENCY=3 \ -e STORAGE_TYPE=elasticsearch \ -e ES_HOSTS=http://es-host:9200 \ openzipkin/zipkin:latest

启动日志里出现类似Using RabbitMQ: amqp://...和Collector reading from queue trace.zipkin.prod的字样,就说明开关生效了。如果只看到 HTTP collector 的日志,八成是环境变量名写错或者 URI 格式不对,RABBIT_URI里的 vhost 用的是路径段而不是查询参数,amqp://host:5672/vhost而不是amqp://host:5672?vhost=vhost,这个细节很容易栽。

提示:RABBIT_URI和RABBIT_ADDRESSES同时存在时前者优先。别两个都配,否则改了一个没生效会浪费很多排查时间。

3.3 RabbitMQ 侧的前置准备

在 RabbitMQ 上要做的不是“建队列”这么简单,有几个参数值得提前定好。

vhost 隔离:给链路追踪单独开一个 vhost,比如trace,把用户权限限制在这个 vhost 内。好处是队列、连接数、磁盘占用都能按 vhost 单独观察,也避免有人误删。

队列声明权:Zipkin 的 collector 在启动时会自己声明队列(durable),所以不需要你手工建。但如果你的环境禁止客户端声明队列(有些团队会关掉),就得手工建好同名队列,并且属性要对得上:durable=true、非 auto-delete、非 exclusive,否则 collector 启动会报PRECONDITION_FAILED。

长度上限与溢出策略:这是最值钱的一条。队列一定要设x-max-length或者x-max-length-bytes,配x-overflow: drop-head。链路数据是可丢弃数据,队列堆到上限就丢掉最老的,比你眼睁睁看着 broker 内存水位飙升最后触发流控要安全得多。我一般按“能扛住 10 分钟峰值流量”来估上限,具体算法在第 6 节。

不要开队列的持久化消息:队列本身 durable 没问题,但消息不要持久化。业务侧 publish 时把deliveryMode设成 1,broker 重启时丢掉的那部分链路数据,真的不重要。

4. 业务服务改造:把 Reporter 搬到队列上

4.1 路线A:用 zipkin-sender-amqp-client 装配 AsyncReporter

先说 Brave / Sleuth 2.x 这条线,它是目前存量项目里最常见的组合。核心思路很简单:把默认的URLConnectionSender换成RabbitMQSender,其余全部交给AsyncReporter。

@Configuration public class ZipkinAmqpReporterConfig { // 复用 Spring 的 RabbitMQ 连接工厂,别自己再造一个 @Bean(destroyMethod = "close") public Channel zipkinChannel(ConnectionFactory connectionFactory) throws Exception { Connection connection = connectionFactory.newConnection("zipkin-reporter"); return connection.createChannel(); } @Bean public Sender zipkinSender(Channel channel) { return RabbitMQSender.newBuilder(channel) .queue("trace.zipkin.prod") .encoding(Sender.Encoding.JSON) .build(); } @Bean public Reporter<Span> zipkinReporter(Sender sender) { return AsyncReporter.builder(sender) .messageMaxBytes(256 * 1024) .closeTimeout(5, TimeUnit.SECONDS) .build(); } }

几个必须解释的点。第一,Channel 不要直接复用业务发送消息的那个。AMQP 的 Channel 不是线程安全的,业务侧的 Channel 由 Spring 的RabbitTemplate管理,共享会出乱子。这里单独开一条 Connection 和一条 Channel 给上报用,互相隔离。第二,messageMaxBytes从默认的 MB 级别下调到 256KB,原因后面讲。第三,Reporter的 bean 名称和类型要和自动配置对得上,Sleuth 的自动配置里对Reporter<Span>用了条件注解,你声明同类型的 bean 通常能覆盖;如果发现没覆盖上,就用@Primary或者直接把自动配置的 Sender 排除掉。

如果你用的是spring.zipkin.sender.type=rabbit的零代码方案,其实就是框架帮你做了上面这件事,参数取默认值。能用,但队列名、编码、批量都不好控制,我一般只在临时验证时用。

4.2 路线B:Micrometer Tracing 下替换 SpanHandler

Spring Boot 3 的项目走 Micrometer Tracing,情况变了。AsyncReporter这套 API 还在(Brave 那边没变),但把它接进Tracing的方式变了:默认装配里,Tracing会挂一个SpanHandler负责把 span 交给 Zipkin reporter,你要做的是提供一个自己的 handler 顶掉它。

@Configuration public class TracingAmqpConfig { @Bean public Channel zipkinChannel(ConnectionFactory connectionFactory) throws Exception { return connectionFactory.newConnection("zipkin-reporter").createChannel(); } @Bean public Sender zipkinSender(Channel channel) { return RabbitMQSender.newBuilder(channel) .queue("trace.zipkin.prod") .encoding(Sender.Encoding.JSON) .build(); } @Bean public BraveTracingBuilderCustomizer rabbitSpanHandlerCustomizer(Sender sender) { Reporter<zipkin2.Span> reporter = AsyncReporter.builder(sender) .messageMaxBytes(256 * 1024) .build(); SpanHandler handler = ZipkinSpanHandler.newBuilder(reporter).build(); return builder -> builder.spanHandler(handler); } }

BraveTracingBuilderCustomizer是 Spring Boot 提供的扩展点,用来在Tracing构建前插手Tracing.Builder。注意spanHandler(...)是替换语义而不是追加,如果你还配了别的 handler(比如自定义的指标统计 handler),要用CompositeSpanHandler手动组合,两个 handler 都塞进去,否则先设置的那个会被悄悄覆盖掉,而且不报错,表现为“日志里有 span 但队列里没数据”或者反过来。

还有一个版本注意点:这套写法依赖 Boot 的 tracing 自动配置提供了扩展点,不同小版本在类名和包路径上可能有差异,落地前先在自己的版本上断点确认一下Tracingbean 是怎么构建的,别盲抄。我踩过一次,扩展点没生效也不报错,一直以为是 RabbitMQ 权限问题,最后发现 customizer 压根没被调用。

4.3 编码与批量参数怎么调

messageMaxBytes这个参数是整篇文章里我最想强调的一个。AsyncReporter的默认值是按 HTTP 场景调的,一次 POST 攒到几 MB 很划算,一个请求搞定。但在 AMQP 上,大消息会被拆成多个 frame 传输,broker 内存里要拼装,消费端要等齐所有 frame 才能交付,一旦某条大消息和一堆小消息混在同一个队列里,就会出现明显的队头阻塞。所以我把批量上限压到 256KB 左右,单个 JSON span 大约 500B 到 1KB,一批差不多两三百个 span,够摊销固定开销了,又不至于把 broker 撑到难受。

配合调的是 flush 间隔。默认 1 秒,链路量小的时候会导致大量“一条消息一个 span”的情况,消息条数飙升。我的做法是把AsyncReporter的 flush 间隔适当放宽到 2 到 3 秒,牺牲一点实时性换吞吐;如果排查实时问题有需求,可以按环境区分配置,生产放宽、预发放宽、本地收紧。

编码选型上,JSON 和 Proto3 我都跑过对比。同样一批 200 个 span,JSON 大约 180KB,Proto3 大约 110KB,体积差 40% 左右。如果你的 span 量在每秒几千以下,这 70KB 的差异完全不值得牺牲可读性;如果到了每秒几万个 span,那 Proto3 就值得上,同时可以在 EH 里相应下调批量上限,让单条消息保持在 200KB 以内。

4.4 上送失败时的兜底逻辑

消息队列不是万能的,broker 重启、网络分区、队列被打满,publish 都会失败。AsyncReporter的行为是:发送失败时把这一批 span 重新放回缓冲队列,如果缓冲满了就丢弃并计数。所以你要做两件事。

一是监控丢弃计数。AsyncReporter暴露了reporter.droppedSpans()这类统计(不同版本 API 名略有差异),把它接到指标系统里,告警阈值不需要设得太敏感,因为偶发丢弃是设计内的行为,但如果持续增长就说明队列或者 collector 有问题。

二是给 publish 失败加短路。业务侧的 Channel 如果进入异常状态,后续每次 publish 都会立即抛异常,AsyncReporter会不断重试,反而拖慢自己。稳妥的做法是在配置里加一个简单的断路器,连续失败 N 次后暂停上报一段时间,让 reporter 自己把数据丢掉,恢复后再继续。这段逻辑不复杂,几十行代码,但能防止“追踪系统挂了把业务进程拖垮”这种最坏情况。

注意:不要给链路数据的 publish 加 publisher confirm 然后阻塞等待。这等于把异步链路又变回同步,收益全没了。真要确认投递,用异步 confirm 回调做计数就好。

5. 与业务异步消息共存:让 trace 不断链

5.1 生产者侧把上下文塞进消息头

上一节讲的是“追踪数据走队列”,这一节讲另一个更容易被忽略的问题:业务本身也用 RabbitMQ 的时候,trace 会在消息边界断掉。这个问题跟上报通道无关,但只要你在用 MQ,就一定会遇到,而且比上报通道的影响更大——因为断链意味着你看到的调用树缺了一整段异步分支。

原理很简单:trace 上下文默认存在 ThreadLocal 里,跨线程就没了,跨进程更没。要让异步链路接上,生产者在发送业务消息时,必须把当前的 trace 上下文写进消息头。B3 协议的核心字段就那几个:X-B3-TraceId、X-B3-SpanId、X-B3-ParentSpanId、X-B3-Sampled。如果用的是 W3C 标准,就是traceparent和tracestate。

MessagePostProcessor traceHeaderProcessor(Tracer tracer) { return message -> { Span current = tracer.currentSpan(); if (current != null) { MessageProperties props = message.getMessageProperties(); props.setHeader("X-B3-TraceId", current.context().traceIdString()); props.setHeader("X-B3-SpanId", current.context().spanIdString()); props.setHeader("X-B3-Sampled", current.context().sampled() ? "1" : "0"); } return message; }; } rabbitTemplate.convertAndSend(exchange, routingKey, payload, traceHeaderProcessor(tracer));

用MessagePostProcessor而不是手工setHeader再发,好处是发送逻辑可以统一抽到一个工具类里,业务代码不用关心追踪细节。采样标记一定要传,否则下游会按自己的采样率重新决策,出现“上游采了、下游没采”的断点,排查时特别迷惑。

5.2 消费者侧恢复上下文

消费端要做的是在消息处理方法执行的线程上,用消息头里的信息重建上下文,并且在 finally 里关闭。这里有个坑:用@RabbitListener的时候,消费线程是框架的线程池,不是你的业务线程,所以“在方法开头恢复”要确保恢复的是当前线程。

@RabbitListener(queues = "order.created") public void onMessage(Message message) { SpanContext ctx = extractFromHeaders(message.getMessageProperties()); Span span = tracer.nextSpan().name("consume order.created").extract(ctx).start(); try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) { // 业务逻辑 } finally { span.finish(); } }

如果用的是 Spring Cloud Sleuth 或 Micrometer Tracing 的RabbitTemplate自动装饰(有些版本确实内置了消息头的传播支持),那这部分可能已经自动做了,但你得验证。验证方法很直接:发一条消息,在消费端打日志看traceId是否和生产端一致,不一致就是没接上。我见过不少项目“以为框架会自动传播”,结果链路图里所有异步分支都是独立的新 trace,等于白接。

5.3 跨进程传递的字段清单

不同协议和框架的字段名不一致,混用会直接导致解析失败。下面这张表我贴在项目文档里,供对照:

字段B3 单头B3 多头W3C
追踪 IDb3中的第一段X-B3-TraceIdtraceparent第二段
跨度 IDb3中的第二段X-B3-SpanIdtraceparent第三段
父跨度b3中的第三段X-B3-ParentSpanId由parent-id段体现
采样b3中的第四段X-B3-Sampledtraceparent第一段末位
扩展无X-B3-Flagstracestate

单头格式(b3: traceid-spanid-sampled-parentspanid)体积小,适合消息头空间敏感的场景;多头格式可读性好,排查时一眼能看懂。我的建议是全链路统一一种,别一半单头一半多头,网关层做一次转换就够了。

6. 压测调优与容量估算

6.1 数据量估算公式与实例

容量规划不靠感觉,靠算。基本公式:每秒 span 数 = 业务 QPS × 采样率 × 单请求平均 span 数。举个模拟场景:一个下单链路,峰值 QPS 2000,采样率 10%,一次下单要经过网关、订单、库存、支付、消息消费共 5 个服务,每个服务平均产生 3 个 span,合计 15 个 span。

于是:2000 × 0.1 × 15 = 3000 span/s。按单 span JSON 平均 800B 算,原始数据量 2.4MB/s。如果一条消息只放一个 span,那就是 3000 条消息/秒;按 256KB 批量打包,每条消息大约装 250 个 span,消息数降到约 12 条/秒。这个对比很能说明问题:消息条数从 3000 降到 12,broker 的压力差了 250 倍。很多人抱怨“上了 RabbitMQ 反而更慢了”,根子就在这——用了队列但没开批量,一条一个 span 地发。

队列容量按“扛住 10 分钟峰值”估:2.4MB/s × 600s ≈ 1.4GB。所以x-max-length-bytes设 1.5GB 到 2GB 比较合理,配合drop-head,正常运行时队列深度几乎为 0,只有 collector 故障时才增长,到了上限自动丢最老的。

6.2 collector 并发、prefetch 与消费能力

collector 的消费能力取决于三个因素:并发消费者数、每次拉取的批量、下游存储的写入速度。RABBIT_CONCURRENCY控制第一个,经验值是 CPU 核数的 1 到 2 倍,但不要超过 8,因为每个消费者都会持有独立的 Channel 和缓冲。prefetch 在 Zipkin 的 collector 里是内部控制的,它按批 ack,所以实际预取量会比你预期的大一点,这也是为什么队列长度上限一定要设。

存储侧的瓶颈经常被忽略。如果后端是 Elasticsearch,单条 span 的写入成本远高于解码成本,collector 并发开再大也没用,反而把 ES 的 bulk 队列打满。这时候应该减少 collector 并发、增加 bulk 批量,让写入更“厚重”一点。我做过一次对比:同样 3000 span/s,concurrency 从 6 降到 2、ES bulk 从 1000 提到 5000,写入延迟反而下降了四成,因为减少了小批量刷盘的次数。

6.3 必须盯住的几个指标

上线之后没有指标就是盲飞。下面这几个我默认都会配上告警:

指标来源正常范围异常含义
队列深度RabbitMQ接近 0持续增长说明消费跟不上
入队速率RabbitMQ与估算值接近陡降可能是生产端断了
丢弃消息数队列溢出统计偶发持续增长说明容量不足
reporter 丢弃数业务进程接近 0上报通道异常或队列不可用
publish 耗时 P99业务进程毫秒级升高说明 broker 流控
collector 消费延迟Zipkin 日志秒级分钟级说明积压
ES bulk 拒绝数存储侧0写入被打回,需降并发

最容易被忽略的是“入队速率陡降”。如果某天早上队列深度是 0,看着一切正常,但入队速率从 12 条/秒掉到了 0,说明生产端根本没在发——可能是一次发版把配置覆盖了,也可能是 RabbitMQ 用户密码过期。这种“静默失效”比队列积压更难发现,因为所有监控看起来都是绿的。

7. 排查实录:十有八九是这几类问题

7.1 队列里有消息但 Zipkin 查不到

这是最高频的一类。消息在队列里能看到,甚至消费速率也正常,但 Zipkin UI 里搜不到 trace。按顺序排查:

第一步,确认消息是否被真正消费并 ack。用管理台看队列的messages_unacknowledged,如果这个数字持续增长,说明 collector 拉走了但处理失败、没 ack。常见原因是解码失败——比如生产端用了 Proto3 但消息体被某个中间层改写过,或者批量消息不是合法的数组。

第二步,抓一条消息看 body。用rabbitmqadmin get queue=trace.zipkin.prod count=1 ackmode=reject_requeue_true,把 payload 打印出来。合法 JSON 数组、字段名正确、traceId是 16 或 32 位十六进制,这几项一眼就能验证。

第三步,看 collector 日志里有没有Failed to collect spans之类的报错。如果有,通常是存储侧的问题,比如 ES 索引模板冲突、字段类型不匹配(duration被写成了字符串),跟 RabbitMQ 没关系,但表现是一样的“查不到”。

第四步,检查查询时间范围。异步通路天然有延迟,如果你按“最近 5 分钟”查而队列积压了 20 分钟,那当然查不到,把范围放到最近一小时再看。

7.2 队列里根本没消息

这种情况先分清是生产端没发,还是发到了别的地方。最快的验证方式是在业务进程里打开 reporter 的 debug 日志,看有没有Spans reported之类的记录。没有的话,八成是这几件事:Reporter bean 没生效(自动配置把它覆盖回去了)、采样率设成了 0、或者 trace 根本没开始(比如入口是定时任务,没有 HTTP 入口,也没手动创建 span)。

采样率这个坑值得单独说。有人把spring.sleuth.sampler.probability设成 0.1 之后发现“队列里几乎没数据”,以为是链路坏了,其实是 90% 的请求本来就不采。更隐蔽的是采样决策在入口做,但入口是网关,网关的采样率配得极低,导致下游所有服务都不采。排查链路数据量异常时,第一个要看的永远是采样率,而不是上报通道。

还有一种情况是发到了别的 vhost 或者别的队列。检查业务配置里的spring.rabbitmq.virtual-host和 sender 里写的 queue 名,两边必须和服务端的RABBIT_URI、RABBIT_QUEUE完全一致。这种错误往往发生在多环境切换时,配置中心的某个 profile 没覆盖到。

7.3 trace 断链、span 缺失与重复

断链的典型表现是一个请求在 Zipkin 里显示成两条独立的 trace,中间隔着一次消息发送。原因就是第 5 节讲的上下文没传播,去查消息头里有没有 B3 字段即可。如果是 HTTP 调用断链,那是另一个话题(拦截器没生效),先分清断在哪一跳。

span 缺失有两种。一种是“只缺最后几个 span”,通常是 reporter 缓冲还没 flush,进程就被重启了,属于正常损耗,把closeTimeout调大、在 shutdown hook 里显式 close reporter 能缓解。另一种是“某个服务的 span 时有时无”,大概率是那个服务的批量参数太激进,或者采样率和其他服务不一致,导致部分 span 被丢弃。

span 重复看起来诡异,其实很常见:同一个请求被上报了两次,一次走 HTTP 一次走 AMQP。多见于改造过渡期——旧配置没删干净,spring.zipkin.base-url还在,同时新加了 AMQP sender,于是两份数据都发。排查时看 Zipkin 里同一个 spanId 出现两次、但localEndpoint的 IP 相同,基本可以确定是双通道。解决办法是把 HTTP 的 endpoint 配置删掉,或者用条件注解确保只装配一个 Reporter。

7.4 问题速查表

把上面这些压成一张表,出问题的时候按行对,比从头读文档快得多:

现象最可能原因快速验证处理
队列空,UI 也空Reporter 没生效 / 采样为 0看 reporter 日志与采样配置修 bean 或采样率
队列有消息,UI 空解码失败 / 存储报错抓消息体 + 看 collector 日志统一编码 / 修索引
队列持续增长collector 消费能力不足看消费速率与存储延迟扩实例 / 降并发提批量
业务耗时上升broker 流控 / 大消息阻塞看 publish P99 与帧大小下调批量上限
trace 断成两条消息头没带上下文打印消息头补 B3 传播
span 重复双通道上报看是否有两个 endpoint 配置删掉 HTTP 配置
数据延迟几分钟正常积压看队列深度趋势扩容或接受延迟

8. 灰度上线与回滚的几个实操建议

改造链路追踪这种“看起来不影响业务”的通道,最容易被忽略上线策略。我的做法是分三步走。第一步,先把 AMQP 通道和 HTTP 通道同时开着,让数据走双份,验证队列里的数据和 UI 里的数据对得上——这一步会短暂出现重复 span,无所谓,验证完就切。第二步,把 HTTP 通道关掉,只留 AMQP,观察一到两天,重点看 reporter 丢弃数和业务侧 GC 有没有变化。第三步,压测一次峰值流量,验证队列长度上限和溢出策略确实生效,确认积压十分钟后系统还能自己恢复。

回滚策略也要提前想好。因为上报通道是配置驱动的,回滚基本就是把Sender换回URLConnectionSender,所以关键是别把旧配置删掉,用配置中心留一个开关。另外一个容易忽略的点是队列的生命周期:切回 HTTP 之后,队列里的残留消息没人消费了,记得给队列配 TTL 或者手工清理,否则它会一直占着内存,运维看到告警还得来找你。

还有个细节值得分享:别在压测环境用生产队列名。这个错误我犯过一次,压测脚本跑起来,几万 QPS 的合成数据灌进生产队列,collector 消费完之后在 UI 里出现了一大片莫名其妙的 trace,排查了半小时才想起来是自己干的。给压测环境单独一个队列,甚至单独的 vhost,成本极低。

我在实际落地这套方案的过程中,最大的体会是:异步数据收集这件事,技术难点不在“怎么把消息发到 RabbitMQ”,那部分代码不到五十行。真正花时间的是三件事——搞清楚你的数据量到底有多大(决定了批量参数和队列上限)、搞清楚哪些地方会断链(决定了上下文传播要补哪些位置)、以及搞清楚出问题时该看哪个指标(决定了你半夜被叫起来能不能十分钟定位)。这三件事想明白了,剩下的都是配置和调试。如果你现在的链路追踪还在走同步 HTTP,可以先从“把批量上限调小、flush 间隔放宽”这种低风险改动试起,感受一下数据量和延迟的变化,再决定要不要上消息队列——有时候问题不在通道,而在参数。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进