Kafka 3+Spring Boot 3多生产者同分区消息乱序问题求解

针对多生产者实例导致同一Partition消息乱序的问题，结合Kafka 3和Spring Boot 3，以下是几种可行的解决思路及实现示例：

1. 全局序列号+消费者窗口重排 #

给每个业务事件（如用户资料修改）生成全局唯一的递增序列号，消费者端通过时间窗口收集消息，按序列号排序后再处理，同时保留少量缓冲应对延迟到达的消息。

生产者实现（Spring Boot） #

@Service
public class UserProfileProducer {
    private final KafkaTemplate<String, UserProfileEvent> kafkaTemplate;
    private final SequenceGenerator sequenceGenerator; // 基于Redis/数据库实现的全局序列号生成器

    public UserProfileProducer(KafkaTemplate<String, UserProfileEvent> kafkaTemplate, SequenceGenerator sequenceGenerator) {
        this.kafkaTemplate = kafkaTemplate;
        this.sequenceGenerator = sequenceGenerator;
    }

    public void sendProfileUpdate(UserProfileEvent event) {
        // 生成全局递增序列号
        event.setSequenceId(sequenceGenerator.nextId());
        // 哈希路由到固定Partition，保证同一用户事件进入同一分区
        int partition = Math.abs(event.getUserId().hashCode()) % 3;
        kafkaTemplate.send("user-profile-topic", partition, event.getUserId(), event);
    }
}

// 事件实体
public record UserProfileEvent(Long userId, String name, Long sequenceId) {}

消费者实现（Spring Boot） #

通过批量监听+定时窗口排序处理：

@Service
public class UserProfileConsumer {
    private final ConcurrentHashMap<Long, List<UserProfileEvent>> userEventBuffer = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    public UserProfileConsumer() {
        // 每5秒触发一次窗口内消息排序处理
        scheduler.scheduleAtFixedRate(this::processBufferedEvents, 0, 5, TimeUnit.SECONDS);
    }

    @KafkaListener(topics = "user-profile-topic", containerFactory = "batchListenerContainerFactory")
    public void listen(List<ConsumerRecord<String, UserProfileEvent>> records) {
        for (ConsumerRecord<String, UserProfileEvent> record : records) {
            UserProfileEvent event = record.value();
            userEventBuffer.computeIfAbsent(event.userId(), k -> new ArrayList<>()).add(event);
        }
    }

    private void processBufferedEvents() {
        for (Map.Entry<Long, List<UserProfileEvent>> entry : userEventBuffer.entrySet()) {
            List<UserProfileEvent> events = entry.getValue();
            // 按序列号排序
            events.sort(Comparator.comparing(UserProfileEvent::sequenceId));
            // 处理排序后的消息
            events.forEach(this::handleProfileEvent);
            // 保留最近10条消息应对延迟到达情况
            entry.setValue(events.subList(Math.max(0, events.size() - 10), events.size()));
        }
    }

    private void handleProfileEvent(UserProfileEvent event) {
        // 执行业务逻辑：更新用户资料等
        System.out.printf("处理用户%d的资料修改：%s（序列号：%d）%n", event.userId(), event.name(), event.sequenceId());
    }
}

2. 请求路由到单一生产者实例 #

通过请求路由策略，将同一用户的所有修改请求转发到固定的生产者实例，确保同一业务流的消息由单个生产者串行发送，从源头避免跨实例乱序。

Spring Cloud Gateway路由配置示例 #

spring:
  cloud:
    gateway:
      routes:
        - id: user-producer-route
          uri: lb://user-profile-producer
          predicates:
            - Path=/api/profile/update/**
          filters:
            - name: UserIdRouteFilter
              args:
                producerUris: ["http://producer-1:8080", "http://producer-2:8080", "http://producer-3:8080"]

自定义路由过滤器实现 #

@Component
public class UserIdRouteFilter implements GatewayFilter, Ordered {
    private final List<String> producerUris;

    public UserIdRouteFilter(@Value("${producerUris}") List<String> producerUris) {
        this.producerUris = producerUris;
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String userId = exchange.getRequest().getHeaders().getFirst("X-User-Id");
        if (userId != null) {
            // 根据用户ID哈希到固定的生产者实例
            int index = Math.abs(userId.hashCode()) % producerUris.size();
            URI uri = URI.create(producerUris.get(index) + exchange.getRequest().getPath());
            exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, uri);
        }
        return chain.filter(exchange);
    }

    @Override
    public int getOrder() {
        return 0;
    }
}

3. 分布式锁+单实例串行处理 #

利用分布式锁，确保同一用户的修改请求在任意时刻只能被一个生产者实例处理，保证消息发送的串行性。

生产者结合Redis分布式锁实现 #

@Service
public class LockedUserProfileProducer {
    private final KafkaTemplate<String, UserProfileEvent> kafkaTemplate;
    private final StringRedisTemplate redisTemplate;
    private static final String LOCK_KEY_PREFIX = "user:profile:lock:";
    private static final long LOCK_EXPIRE = 60; // 锁过期时间60秒

    public LockedUserProfileProducer(KafkaTemplate<String, UserProfileEvent> kafkaTemplate, StringRedisTemplate redisTemplate) {
        this.kafkaTemplate = kafkaTemplate;
        this.redisTemplate = redisTemplate;
    }

    public void sendProfileUpdate(UserProfileEvent event) {
        String lockKey = LOCK_KEY_PREFIX + event.userId();
        // 获取分布式锁
        Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "locked", LOCK_EXPIRE, TimeUnit.SECONDS);
        if (Boolean.TRUE.equals(locked)) {
            try {
                // 串行处理并同步等待消息发送确认
                kafkaTemplate.send("user-profile-topic", event.getUserId().hashCode() % 3, event.getUserId(), event).get();
            } catch (InterruptedException | ExecutionException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("消息发送失败", e);
            } finally {
                // 释放锁
                redisTemplate.delete(lockKey);
            }
        } else {
            // 锁被占用，根据业务场景选择重试或返回提示
            throw new RuntimeException("当前用户资料修改正在处理中，请稍后重试");
        }
    }
}

4. 版本号覆盖策略（最终一致性优先） #

如果业务允许最终一致，可给每个用户的修改事件添加版本号，消费者处理时仅保留最新版本的消息，忽略旧版本。

生产者实现 #

@Service
public class VersionedUserProfileProducer {
    private final KafkaTemplate<String, UserProfileEvent> kafkaTemplate;
    private final UserProfileRepository profileRepository; // 数据库查询当前版本

    public VersionedUserProfileProducer(KafkaTemplate<String, UserProfileEvent> kafkaTemplate, UserProfileRepository profileRepository) {
        this.kafkaTemplate = kafkaTemplate;
        this.profileRepository = profileRepository;
    }

    public void sendProfileUpdate(UserProfileEvent event) {
        // 查询当前用户的最新版本号，新事件版本号自增
        Long currentVersion = profileRepository.findVersionByUserId(event.userId());
        event.setVersion(currentVersion == null ? 1 : currentVersion + 1);
        kafkaTemplate.send("user-profile-topic", event.getUserId().hashCode() % 3, event.getUserId(), event);
    }
}

消费者实现 #

@Service
public class VersionedUserProfileConsumer {
    private final UserProfileRepository profileRepository;

    public VersionedUserProfileConsumer(UserProfileRepository profileRepository) {
        this.profileRepository = profileRepository;
    }

    @KafkaListener(topics = "user-profile-topic")
    public void listen(UserProfileEvent event) {
        // 对比数据库中的当前版本，仅处理版本更高的消息
        Long currentVersion = profileRepository.findVersionByUserId(event.userId());
        if (currentVersion == null || event.version() > currentVersion) {
            // 更新用户资料并保存版本号
            profileRepository.updateProfile(event.userId(), event.name(), event.version());
        } else {
            // 旧版本消息，直接忽略
            System.out.printf("忽略用户%d的旧版本消息：%s（当前版本：%d，消息版本：%d）%n", event.userId(), event.name(), currentVersion, event.version());
        }
    }
}

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