Spring Cloud Streams中如何正确添加消息Header？现有实现存在异常

问题背景 #

Spring Cloud Streams官方文档仅提供Header的访问示例，无添加或修改Header的相关示例。网上使用ProcessorContext添加Header的示例，会导致消息Header添加结果不一致。

当前实现代码 #

EventHeaderTransformer类 #

public class EventHeaderTransformer implements Transformer<String, RequestEvent, KeyValue<String, RequestEvent>>
{
    private static final String EVENT_HEADER_NAME = "event";
    ProcessorContext context;

    public EventHeaderTransformer() { }

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
    }

    @Override
    public KeyValue<String, RequestEvent> transform(String key, RequestEvent value) {
        context.headers().add(EVENT_HEADER_NAME, value.getEventName().getBytes());
        return new KeyValue<>(key, value);
    }

    @Override
    public void close() {
        // nothing here
    }

}

流处理函数 #

public Function<KStream<String, Request>, KStream<String, RequestEvent>> streamRequests() {
        return input -> input
            .transform(() -> unrelatedTransformer)
            .filter(unrelatedFilter)
            // 存在问题的transformer
            .transform(() -> eventHeaderTransformer);
            // 该transformer后的调试输出显示结果不一致
    }

配置文件 #

streamRequests-in-0:
  destination: queue.unmanaged.requests
  group: streamRequests
  consumer:
    partitioned: true
    concurrency: 3
streamRequests-out-0:
  destination: queue.core.requests

测试异常情况 #

测试9条消息后，Header添加存在异常（p=分区，[N]=偏移量）：

• p0[0] = 无Header消息
• p1[0] = 无Header消息
• p2[0] = 带Header消息
• p0[0] = 无Header消息
• p1[0] = 无Header消息
• p2[0] = 无Header消息
• p0[0] = 无Header消息
• p1[0] = 无Header消息
• p2[0] = 带Header消息

调试信息显示有时Header未添加成功或Header为空等异常。

解决方案 #

问题根源 #

当前的EventHeaderTransformer是单例实例，在多并发（concurrency=3）场景下，多个线程共享同一个ProcessorContext实例，导致Header操作出现线程安全问题，结果不一致。

正确实现方式 #

方式1：确保每个线程使用独立的Transformer实例

修改流处理函数，每次调用transform时创建新的EventHeaderTransformer实例，避免线程共享：

public Function<KStream<String, Request>, KStream<String, RequestEvent>> streamRequests() {
    return input -> input
        .transform(() -> unrelatedTransformer)
        .filter(unrelatedFilter)
        // 每次创建新的transformer实例，保证线程隔离
        .transform(EventHeaderTransformer::new);
}

EventHeaderTransformer本身无需修改，现在每个线程会拥有自己的实例和对应的ProcessorContext。

方式2：使用map操作构建带Header的消息（更简洁）

如果不需要复杂状态管理，直接用map操作通过ProducerRecord构建新消息，线程安全且直观：

public Function<KStream<String, Request>, KStream<String, RequestEvent>> streamRequests() {
    return input -> input
        .transform(() -> unrelatedTransformer)
        .filter(unrelatedFilter)
        .map((key, value) -> {
            ProducerRecord<String, RequestEvent> record = new ProducerRecord<>(
                "queue.core.requests",
                key,
                value
            );
            record.headers().add(EVENT_HEADER_NAME, value.getEventName().getBytes());
            return record;
        });
}

方式3：基于Spring Messaging的Message转换

如果使用Spring Messaging的Message模型，可直接操作MessageHeaders：

@Bean
public Function<Message<RequestEvent>, Message<RequestEvent>> addHeader() {
    return message -> {
        // 复制原Header并添加新Header
        Map<String, Object> headerMap = new LinkedHashMap<>(message.getHeaders());
        headerMap.put(EVENT_HEADER_NAME, message.getPayload().getEventName().getBytes());
        return MessageBuilder.createMessage(message.getPayload(), new MessageHeaders(headerMap));
    };
}

关键注意事项 #

• 线程安全优先：多并发消费场景下，Transformer必须保证线程安全，要么每个线程拥有独立实例，要么不在Transformer中共享状态。
• 禁止共享ProcessorContext：ProcessorContext绑定到单个线程/任务，不能在多个实例间共享。
• 优先无状态操作：简单Header修改用map比transform更简洁，也更不容易出错。

内容的提问来源于stack exchange，提问作者StevenPG