如何向新SSE消费者广播指定时间戳起的Kafka消息？

你现在的代码有个致命问题：所有SSE客户端共享同一个Kafka消费组，调用seekToTimestamp会直接修改整个消费组的偏移量，导致不管什么时候连进来的客户端，都会从你指定的那个时间点开始收消息，完全做不到只给新客户端推最近2分钟的内容。而且全局的subscribableChannel会把所有消息推给所有客户端，再靠filter过滤orgid，不仅浪费资源，也根本不符合需求。

换个思路，用Spring Kafka的Reactive客户端（需添加spring-kafka-reactive依赖），给每个SSE连接单独建一条消费流，每条流可以自己设置起始消费时间戳，互相不干扰。

第一步：添加依赖 #

先在pom.xml里加上Reactive Kafka的依赖：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-reactive</artifactId>
    <version>你的Spring Kafka版本号</version>
</dependency>

第二步：配置Reactive消费者 #

@Configuration
public class ReactiveKafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${topic.name.consumer}")
    private String topic;

    @Bean
    public ReactiveKafkaConsumerTemplate<String, String> reactiveKafkaConsumerTemplate() {
        Map<String, Object> consumerProps = new HashMap<>();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        // 每个连接用唯一的groupId，彻底避免偏移量共享
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "sse-group-" + UUID.randomUUID());
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        return new ReactiveKafkaConsumerTemplate<>(
                ReceiverOptions.create(consumerProps)
                        .subscription(Collections.singleton(topic))
        );
    }
}

第三步：修改SSE控制器 #

@RestController
public class ServerController {

    private final ReactiveKafkaConsumerTemplate<String, String> reactiveKafkaConsumerTemplate;
    private static final Logger log = LoggerFactory.getLogger(ServerController.class);

    public ServerController(ReactiveKafkaConsumerTemplate<String, String> reactiveKafkaConsumerTemplate) {
        this.reactiveKafkaConsumerTemplate = reactiveKafkaConsumerTemplate;
    }

    @GetMapping(path = "/stream/{orgid}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    Flux<ServerSentEvent<ServiceEvent>> getFolderWatch(@PathVariable String orgid) {
        // 计算2分钟前的时间戳
        long twoMinutesAgo = System.currentTimeMillis() - 2 * 60 * 1000;

        return reactiveKafkaConsumerTemplate
                // 让每个分区都seek到2分钟前的位置
                .receiveAtOffset(twoMinutesAgo)
                // 取出消息内容
                .map(ConsumerRecord::value)
                // 转成你需要的ServiceEvent
                .map(content -> new ServiceEvent(this, content))
                // 只留当前orgid的消息
                .filter(event -> event.getOrgId().equals(orgid))
                // 包装成SSE格式
                .map(event -> ServerSentEvent.<ServiceEvent>builder(event).build())
                // 连接断了就打个日志，也可以加资源清理逻辑
                .doOnCancel(() -> log.info("SSE连接已关闭，orgid: {}", orgid));
    }
}

备选方案：给每个连接单独建KafkaListener容器 #

要是不想用Reactive，也可以给每个SSE连接动态创建一个KafkaListener容器，每个容器用唯一的groupId，这样每个容器的偏移量都是独立的：

@RestController
public class ServerController {

    private final ConcurrentKafkaListenerContainerFactory<String, String> containerFactory;
    private final Map<String, AbstractMessageListenerContainer<String, String>> activeContainers = new ConcurrentHashMap<>();
    private final SimpMessagingTemplate messagingTemplate;
    private static final Logger log = LoggerFactory.getLogger(ServerController.class);

    public ServerController(ConcurrentKafkaListenerContainerFactory<String, String> containerFactory, SimpMessagingTemplate messagingTemplate) {
        this.containerFactory = containerFactory;
        this.messagingTemplate = messagingTemplate;
    }

    @GetMapping(path = "/stream/{orgid}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    Flux<ServerSentEvent<ServiceEvent>> getFolderWatch(@PathVariable String orgid) {
        // 生成唯一会话ID，用来区分不同连接
        String sessionId = UUID.randomUUID().toString();
        long twoMinutesAgo = System.currentTimeMillis() - 2 * 60 * 1000;

        // 创建独立的消费者容器
        AbstractMessageListenerContainer<String, String> container = containerFactory.createContainer("${topic.name.consumer}");
        container.getContainerProperties().setGroupId("sse-group-" + sessionId);
        // 设置消息监听，只给当前会话推对应orgid的消息
        container.setupMessageListener((MessageListener<String, String>) message -> {
            ServiceEvent event = new ServiceEvent(this, message.value());
            if (event.getOrgId().equals(orgid)) {
                messagingTemplate.convertAndSend("/sse/" + sessionId, event);
            }
        });

        // 启动容器并seek到2分钟前的位置
        container.start();
        container.seekToTimestamp(twoMinutesAgo);
        activeContainers.put(sessionId, container);

        // 生成SSE流，监听当前会话的消息
        return messagingTemplate.receive("/sse/" + sessionId)
                .map(Message::getPayload)
                .cast(ServiceEvent.class)
                .map(event -> ServerSentEvent.<ServiceEvent>builder(event).build())
                // 连接关闭时销毁容器，释放资源
                .doOnCancel(() -> {
                    AbstractMessageListenerContainer<String, String> removedContainer = activeContainers.remove(sessionId);
                    if (removedContainer != null) {
                        removedContainer.stop();
                        log.info("已销毁会话{}对应的Kafka消费者容器", sessionId);
                    }
                });
    }
}

• 别共享消费组：每个SSE连接必须用独一无二的groupId，这样每个消费者的偏移量完全独立，seek操作不会影响其他客户端。
• 独立消费流：每个连接对应一条单独的Kafka消费流，只从指定时间点开始消费，只推当前客户端需要的消息。
• 清理资源：一定要在SSE连接关闭时销毁对应的Kafka消费者或容器，不然会占着资源不放。

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