如何为Kafka Producer实现熔断器模式及消息管控机制？

由于Spring Kafka并未提供类似KafkaListenerEndpointRegistry的Producer容器管理API，我们需要通过自定义发送管控层+熔断器状态监听+消息降级存储的组合方案，实现Producer端的熔断控制，同时保证消息不丢失。

核心思路 #

1. 统一所有Producer发送请求的入口，用熔断器包裹发送逻辑
2. 监听熔断器状态转换，控制是否允许直接发送消息到Kafka
3. 当熔断器打开或发送失败时，将消息降级存储到内存/数据库/文件等介质
4. 熔断器恢复时，触发存储消息的重发逻辑

完整实现代码 #

1. 自定义管控型Producer服务 #

封装发送逻辑，绑定熔断器并监听状态变化，处理降级存储与重发：

@Service
public class ManagedKafkaProducer {
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final CircuitBreaker circuitBreaker;
    private final FallbackMessageStorage fallbackStorage;
    private volatile boolean isProductionPaused = false;

    public ManagedKafkaProducer(KafkaTemplate<String, Object> kafkaTemplate,
                               CircuitBreakerRegistry circuitBreakerRegistry,
                               FallbackMessageStorage fallbackStorage) {
        this.kafkaTemplate = kafkaTemplate;
        this.circuitBreaker = circuitBreakerRegistry.circuitBreaker("kafkaProducerCB");
        this.fallbackStorage = fallbackStorage;
        initCircuitBreakerListener();
    }

    /**
     * 监听熔断器状态转换，控制生产开关
     */
    private void initCircuitBreakerListener() {
        circuitBreaker.getEventPublisher().onStateTransition(event -> {
            switch (event.getStateTransition()) {
                case CLOSED_TO_OPEN:
                case CLOSED_TO_FORCED_OPEN:
                case HALF_OPEN_TO_OPEN:
                    isProductionPaused = true;
                    break;
                case OPEN_TO_HALF_OPEN:
                case HALF_OPEN_TO_CLOSED:
                case FORCED_OPEN_TO_CLOSED:
                case FORCED_OPEN_TO_HALF_OPEN:
                    isProductionPaused = false;
                    // 状态恢复时触发存储消息重发
                    fallbackStorage.retryStoredMessages(this::sendToKafka);
                    break;
                default:
                    throw new IllegalStateException("Unknown transition state: " + event.getStateTransition());
            }
        });
    }

    /**
     * 对外暴露的发送接口
     */
    public void sendMessage(String topic, Object payload) {
        if (isProductionPaused) {
            // 熔断器打开，直接降级存储
            fallbackStorage.storeMessage(topic, payload);
            return;
        }

        // 用熔断器包裹发送逻辑，捕获失败后降级存储
        Try.run(() -> sendToKafka(topic, payload))
                .recover(throwable -> {
                    fallbackStorage.storeMessage(topic, payload);
                    throw throwable;
                })
                .get();
    }

    /**
     * 实际发送到Kafka的逻辑
     */
    private void sendToKafka(String topic, Object payload) throws InterruptedException, ExecutionException {
        // 同步发送便于熔断器捕获异常，可根据需求改为异步并处理回调异常
        kafkaTemplate.send(topic, payload).get();
    }
}

2. 消息降级存储接口与实现 #

内存存储实现（适合临时缓存，重启会丢失）

public interface FallbackMessageStorage {
    void storeMessage(String topic, Object payload);
    void retryStoredMessages(Consumer<StoredMessage> sender);
}

@Component
public class InMemoryFallbackStorage implements FallbackMessageStorage {
    private final Queue<StoredMessage> messageQueue = new ConcurrentLinkedQueue<>();

    @Override
    public void storeMessage(String topic, Object payload) {
        messageQueue.add(new StoredMessage(topic, payload, LocalDateTime.now()));
    }

    @Override
    public void retryStoredMessages(Consumer<StoredMessage> sender) {
        while (!messageQueue.isEmpty()) {
            StoredMessage message = messageQueue.peek();
            try {
                sender.accept(message);
                messageQueue.poll();
            } catch (Exception e) {
                // 重发失败，停止重试等待下一次状态恢复
                break;
            }
        }
    }

    public static class StoredMessage {
        private final String topic;
        private final Object payload;
        private final LocalDateTime storeTime;

        public StoredMessage(String topic, Object payload, LocalDateTime storeTime) {
            this.topic = topic;
            this.payload = payload;
            this.storeTime = storeTime;
        }

        public String getTopic() { return topic; }
        public Object getPayload() { return payload; }
    }
}

数据库存储实现（持久化，适合关键消息）

@Component
public class JdbcFallbackStorage implements FallbackMessageStorage {
    private final JdbcTemplate jdbcTemplate;
    private final ObjectMapper objectMapper;

    public JdbcFallbackStorage(JdbcTemplate jdbcTemplate, ObjectMapper objectMapper) {
        this.jdbcTemplate = jdbcTemplate;
        this.objectMapper = objectMapper;
    }

    @Override
    public void storeMessage(String topic, Object payload) throws JsonProcessingException {
        String payloadJson = objectMapper.writeValueAsString(payload);
        jdbcTemplate.update(
                "INSERT INTO kafka_fallback_messages(topic, payload, store_time) VALUES (?, ?, ?)",
                topic, payloadJson, LocalDateTime.now()
        );
    }

    @Override
    public void retryStoredMessages(Consumer<StoredMessage> sender) throws JsonProcessingException {
        List<StoredMessage> messages = jdbcTemplate.query(
                "SELECT id, topic, payload FROM kafka_fallback_messages ORDER BY store_time ASC LIMIT 100",
                (rs, rowNum) -> new StoredMessage(
                        rs.getInt("id"),
                        rs.getString("topic"),
                        objectMapper.readValue(rs.getString("payload"), Object.class)
                )
        );

        for (StoredMessage message : messages) {
            try {
                sender.accept(message);
                // 重发成功后删除记录
                jdbcTemplate.update("DELETE FROM kafka_fallback_messages WHERE id = ?", message.getId());
            } catch (Exception e) {
                // 重发失败，后续可通过定时任务重试
                break;
            }
        }
    }

    public static class StoredMessage {
        private final Integer id;
        private final String topic;
        private final Object payload;

        public StoredMessage(Integer id, String topic, Object payload) {
            this.id = id;
            this.topic = topic;
            this.payload = payload;
        }

        public String getTopic() { return topic; }
        public Object getPayload() { return payload; }
    }
}

3. 熔断器配置（Resilience4j） #

@Configuration
public class CircuitBreakerConfig {
    @Bean
    public CircuitBreakerRegistry circuitBreakerRegistry() {
        CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
                .failureRateThreshold(50) // 失败率达50%时打开熔断器
                .waitDurationInOpenState(Duration.ofMinutes(1)) // 打开状态持续1分钟
                .permittedNumberOfCallsInHalfOpenState(5) // 半开状态允许5次试探调用
                .slidingWindowSize(10) // 滑动窗口大小10
                .build();

        return CircuitBreakerRegistry.of(circuitBreakerConfig);
    }
}

关键说明 #

• 生产控制：通过isProductionPaused变量标记生产状态，熔断器状态切换时自动更新
• 消息可靠性：发送失败或熔断打开时，消息会被存储到备选介质，避免丢失
• 重发机制：熔断器恢复时自动触发重发，也可结合定时任务定期重试未发送成功的消息
• 异步适配：若需异步发送，可在sendToKafka中使用addCallback处理成功/失败回调，将异常传递给熔断器

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