SkyWalking 实现 Kafka 与 RabbitMQ 消息链路追踪实战

当 Consumer 消费消息时，Agent 会：

1. 从消息头中读取 sw8 值；
2. 反序列化并恢复 Trace Context；
3. 创建新的 Span（类型为 Kafka/Consumer），并将其作为上游 Span 的子 Span。

🔗 SkyWalking 官方文档 - Kafka 插件 提供了详细的配置说明。

Java 代码示例（无需修改业务代码！）

假设你有一个简单的 Spring Boot 应用，使用 Kafka 发送和接收消息：

// 生产者
@RestController
public class OrderController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @PostMapping("/order")
    public String createOrder(@RequestBody Order order) {
        // 业务逻辑：保存订单
        String message = "Order created: " + order.getId();
        // 发送消息（SkyWalking Agent 自动埋点）
        kafkaTemplate.send("order-topic", message);
        return "Order submitted";
    }
}

// 消费者
@Component
public class InventoryConsumer {
    @KafkaListener(topics = "order-topic")
    public void handleOrder(String message) {
        // 业务逻辑：扣减库存
        System.out.println("Processing: " + message);
        // ... 扣库存逻辑
    }
}

只需在启动应用时挂载 SkyWalking Agent：

java -javaagent:/path/to/skywalking-agent/skywalking-agent.jar \
-Dskywalking.agent.service_name=order-service \
-jar your-app.jar

Agent 会自动处理上下文传递，无需任何代码侵入！

验证追踪效果

部署后，在 SkyWalking UI 中可以看到类似如下拓扑：

HTTP POST /order → Kafka Send → Kafka Consume (Inventory) → Kafka Consume (Notification)

点击任意 Trace，可看到完整的 Span 链：

• /order (HTTP)
  • Kafka/Producer/order-topic
    • Kafka/Consumer/order-topic (Inventory)
    • Kafka/Consumer/order-topic (Notification)

每个 Span 都包含耗时、时间戳、标签（如 topic、partition）等信息。

2. 手动埋点（高级场景）

在某些特殊情况下（如自定义序列化器、非标准客户端），自动探针可能无法生效。此时可使用 SkyWalking 提供的 Toolkit API 手动注入/提取上下文。

添加依赖

<dependency>
    <groupId>org.apache.skywalking</groupId>
    <artifactId>apm-toolkit-kafka</artifactId>
    <version>8.16.0</version>
</dependency>

手动注入上下文（Producer）

import org.apache.skywalking.apm.toolkit.kafka.KafkaProducerInterceptor;

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

// 添加拦截器（关键！）
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, KafkaProducerInterceptor.class.getName());

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("order-topic", "order-data"));

手动提取上下文（Consumer）

import org.apache.skywalking.apm.toolkit.kafka.KafkaConsumerInterceptor;

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "inventory-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

// 添加拦截器
props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, KafkaConsumerInterceptor.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("order-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 此处已自动恢复 Trace Context
        // 业务逻辑...
    }
}

⚠️ 注意：手动埋点需确保 Producer 和 Consumer 都正确配置拦截器，否则上下文会断裂。

RabbitMQ 链路追踪支持

RabbitMQ 是基于 AMQP 协议的轻量级消息中间件，以可靠性、灵活路由著称。与 Kafka 不同，RabbitMQ 的消息模型基于 Exchange/Queue/Binding，且不原生支持消息头（Header）的自动透传（需显式设置）。

SkyWalking 对 RabbitMQ 的支持主要通过 手动埋点 实现，因为 RabbitMQ Java Client（amqp-client）未被 Agent 自动插桩（截至 8.16.0 版本）。

工作原理

1. Producer：在发送消息前，将当前 Trace Context 序列化为字符串，并作为 Message Properties 中的 headers 字段。
2. Consumer：在接收消息后，从 headers 中提取 sw8 值，恢复 Trace Context，再执行业务逻辑。

Java 代码示例

添加依赖

<dependency>
    <groupId>org.apache.skywalking</groupId>
    <artifactId>apm-toolkit-rabbitmq</artifactId>
    <version>8.16.0</version>
</dependency>

Producer 端：注入上下文

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.apache.skywalking.apm.toolkit.rabbitmq.RabbitMQMessageHeadersInjector;

public class OrderService {
    public void sendOrderMessage(String orderData) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare("order-exchange", "direct");
            channel.queueDeclare("order-queue", false, false, false, null);
            channel.queueBind("order-queue", "order-exchange", "order.key");

            byte[] body = orderData.getBytes();
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                    .contentType("text/plain")
                    .build();

            // 注入 SkyWalking 上下文到 headers
            Map<String, Object> headers = new HashMap<>();
            RabbitMQMessageHeadersInjector.inject(headers);
            props = props.builder().headers(headers).build();

            channel.basicPublish("order-exchange", "order.key", props, body);
        }
    }
}

Consumer 端：提取上下文

import com.rabbitmq.client.*;
import org.apache.skywalking.apm.toolkit.rabbitmq.RabbitMQMessageHeadersExtractor;

public class InventoryService {
    public void startConsuming() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("order-queue", false, false, false, null);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            // 从 headers 中提取上下文
            Map<String, Object> headers = delivery.getProperties().getHeaders();
            if (headers != null) {
                RabbitMQMessageHeadersExtractor.extract(headers);
            }

            // 业务逻辑（此时已处于正确的 Trace 上下文中）
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Processing: " + message);
        };

        channel.basicConsume("order-queue", true, deliverCallback, consumerTag -> {});
    }
}

同样需要挂载 SkyWalking Agent：

java -javaagent:/path/to/skywalking-agent/skywalking-agent.jar \
-Dskywalking.agent.service_name=inventory-service \
-jar inventory-service.jar

追踪效果

在 SkyWalking UI 中，RabbitMQ 的链路将显示为：

HTTP → RabbitMQ Publish → Routing → RabbitMQ Consume

每个消息操作都会生成对应的 Span，如 RabbitMQ/Producer 和 RabbitMQ/Consumer。

上下文传递机制详解

无论是 Kafka 还是 RabbitMQ，SkyWalking 的核心在于 Trace Context 的跨进程传递。其内部使用一种紧凑的字符串格式（称为 sw8 协议）来编码上下文信息。

sw8 格式解析

sw8 字符串结构如下（以 Base64 编码）：

1-TRACE_ID-SEGMENT_ID-SPAN_ID-3-PARENT_SERVICE-PARENT_INSTANCE-NEXT_HOP

各字段含义：

🔗 SkyWalking Cross Process Propagation Headers Protocol 详细描述了该协议。

为什么使用消息头而非消息体？

• 透明性：业务逻辑无需感知追踪数据；
• 兼容性：不影响消息序列化/反序列化；
• 性能：头部数据小，传输开销低。

常见问题与解决方案

Q1: 消息被多个消费者消费，Trace 如何表示？

A: SkyWalking 会为每个消费者创建独立的子 Span，形成 分叉（Fork） 结构。在 UI 中，你会看到一个 Producer Span 下挂多个 Consumer Span。

Q2: 消息延迟很高，如何分析？

A: 在 SkyWalking UI 的 Trace 详情页，可查看每个 Span 的开始/结束时间。计算 Consumer Span 开始时间 - Producer Span 结束时间 即为消息在队列中的等待时间。

Q3: 上下文丢失怎么办？

可能原因：

• 消息头被覆盖（如自定义序列化器未保留 headers）；
• 消费者未正确提取上下文；
• Agent 未加载或版本不匹配。

排查步骤：

1. 检查 Producer 发送的消息是否包含 sw8 头（可通过 Kafka/RabbitMQ 管理工具查看）；
2. 确认 Consumer 代码是否调用了 extract()；
3. 查看 Agent 日志（logs/skywalking-api.log）是否有错误。

Q4: 能否追踪消息重试？

A: 可以！每次重试都会生成新的 Consumer Span，但共享同一个 Trace ID。你可以在 Span 标签中看到重试次数（需业务层记录）。

性能影响评估

SkyWalking 的追踪机制对性能的影响非常小：

• CPU：上下文序列化/反序列化开销 < 1%；
• 内存：每个消息增加约 100~200 字节的头部；
• 网络：额外头部数据可忽略不计。

在生产环境中，建议开启 采样率控制（如 10%），避免全量上报造成 OAP 压力。

# agent.config
agent.sample_n_per_3_secs=10

最佳实践建议

1. 统一 Agent 版本：确保所有服务使用相同版本的 SkyWalking Agent；
2. 命名规范：为服务、Topic/Queue 设置清晰的名称，便于拓扑识别；
3. 异常标记：在业务代码中捕获异常时，调用 Span.errorOccurred() 标记失败；
4. 自定义标签：通过 Span.tag("orderId", "12345") 添加业务标识，方便搜索；
5. 监控告警：在 SkyWalking OAP 中配置消息延迟、失败率等告警规则。

与其他追踪系统的对比

SkyWalking 在消息队列追踪方面提供了更开箱即用的体验，尤其适合 Java 技术栈。

结语

通过 SkyWalking 对 Kafka 和 RabbitMQ 的链路追踪支持，我们能够轻松构建端到端的可观测性体系，将原本'黑盒'的消息流转过程变得透明可控。无论是自动探针的零代码侵入，还是 Toolkit 提供的灵活手动埋点，都极大降低了分布式追踪的实施门槛。

在云原生时代，消息驱动架构只会越来越普遍。掌握 SkyWalking 的消息追踪能力，将成为每一位后端工程师提升系统稳定性和运维效率的利器。