如何在Spring Cloud Stream函数式应用中关联Kafka状态存储实现去重

实现思路 #

要解决重复消息问题，核心是通过状态存储记录已处理的消息标识，在处理前检查状态存储：若消息未处理过则继续流程并更新状态，否则直接过滤。结合你的需求，我们可以在原有KStream流程前插入去重逻辑，同时关联自定义状态存储。

完整实现代码 #

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Objects;

@Configuration
public class StreamProcessorConfig {

    // 定义去重用的持久化状态存储
    @Bean
    public StoreBuilder<KeyValueStore<String, String>> processedMessagesStore() {
        return Stores.keyValueStoreBuilder(
                Stores.persistentKeyValueStore("processed-messages-store"),
                Serdes.String(),
                Serdes.String()
        );
    }

    @Bean
    public Function<KStream<String, String>, KStream<String, Long>> sampleProcessor() {
        return source -> source
                // 步骤1：基于状态存储实现去重
                .transformValues(() -> new ValueTransformerWithKey<String, String, String>() {
                    private KeyValueStore<String, String> processedStore;

                    @Override
                    public void init(ProcessorContext context) {
                        // 从上下文获取预定义的状态存储
                        this.processedStore = (KeyValueStore<String, String>) context.getStateStore("processed-messages-store");
                    }

                    @Override
                    public String transform(String key, String value) {
                        // 以消息内容作为去重标识（实际场景推荐用唯一消息ID）
                        if (processedStore.get(value) == null) {
                            // 未处理过，写入状态存储标记为已处理
                            processedStore.put(value, value);
                            return value;
                        } else {
                            // 已处理过，返回null后续过滤
                            return null;
                        }
                    }

                    @Override
                    public void close() {
                        // 可选：资源清理操作
                    }
                }, "processed-messages-store") // 指定关联的状态存储名称
                .filter(Objects::nonNull) // 过滤掉重复消息（返回null的条目）
                // 步骤2：原有消息转换逻辑
                .mapValues(value -> value.toUpperCase())
                // 步骤3：分组计数并关联计数存储
                .groupBy((key, value) -> value)
                .count(Materialized.as("count-store"))
                .toStream();
    }
}

代码说明 #

1. 状态存储定义：

   • 创建processed-messages-store作为持久化KeyValueStore，用来存储已处理的消息标识，重启后状态不会丢失，避免重复处理历史消息。
   • 若你的消息有唯一业务ID/消息ID（比如从消息头获取），建议用ID作为状态存储的key，比消息内容更可靠。

2. 去重逻辑：

   • 通过transformValues操作访问状态存储，检查当前消息是否已处理：未处理则写入状态存储并返回消息，已处理则返回null。
   • 后续用filter过滤掉null值，实现重复消息的拦截。

3. 原有逻辑保留：

   • 去重后的消息继续执行转大写、分组计数逻辑，最终将计数结果存入count-store状态存储。

测试验证 #

当发送以下消息：

m1 = Hello
m2 = World
m3 = Hello

• m1、m2会被正常处理，计数存储中HELLO和WORLD的计数均为1。
• m3会被状态存储识别为已处理，直接被过滤，不会进入后续计数流程。

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