从数据库向Kafka发送消息：确保不丢失、保持顺序与提升性能的策略

引言：数据库到Kafka数据同步的挑战

在许多企业级应用中，将数据库中的数据可靠地同步到kafka是一个常见的需求。这通常涉及几个关键挑战：

1. 消息不丢失（At-Least-Once Delivery）：确保数据库中的每一条记录都能成功发送到Kafka。这通常需要Kafka生产者配置acks=all和集群配置min.insync.replicas等参数。
2. 严格的消息顺序性：消息在数据库中的写入顺序，应严格保持在Kafka中的消费顺序。这对于某些业务场景（如事件溯源、交易日志）至关重要。
3. 数据原子性操作：消息成功发送到Kafka后，应从数据库中删除。这个过程需要与Kafka发送操作协同，确保要么都成功，要么都失败，避免数据重复发送或丢失。
4. 性能与吞吐量：在满足上述可靠性和顺序性要求的同时，系统需要具备足够的性能来处理大量数据，尤其是在定时调度任务中。

这些要求之间往往存在权衡。为了实现严格的顺序性和不丢失，通常需要引入同步机制，但这可能导致性能瓶颈。本文将深入探讨两种常见的解决方案及其各自的优缺点。

同步阻塞式发送：严格顺序与可靠性的保障

为了确保消息的严格顺序性，一种直观的方法是采用同步阻塞式的发送机制。这种方法的核心思想是：在发送下一条消息之前，必须确认前一条消息已成功送达Kafka Broker。

工作原理

生产者在发送每条消息后，会阻塞等待Kafka Broker的确认。只有当ListenableFuture返回成功结果后，才认为当前消息已成功发送。如果发送失败，则停止后续消息的发送，并记录已成功发送的消息ID。

示例代码

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SynchronousKafkaSender {

    private static final Logger log = LoggerFactory.getLogger(SynchronousKafkaSender.class);
    private final KafkaTemplate kafkaTemplate;

    public SynchronousKafkaSender(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 以同步阻塞方式发送消息，确保严格顺序和不丢失。
     *
     * @param topicName Kafka主题名称
     * @param data      待发送的数据列表，每个元素包含一个ID和消息体
     * @return 成功发送到Kafka并得到确认的消息ID列表
     */
    public List sendMessages(String topicName, List> data) {
        List successIds = new ArrayList<>();
        for (MessageWrapper messageWrapper : data) {
            // 发送消息，并阻塞等待结果
            ListenableFuture> listenableFuture = 
                kafkaTemplate.send(topicName, messageWrapper.getKey(), messageWrapper.getValue());
            try {
                listenableFuture.get(3, TimeUnit.SECONDS); // 设置超时时间
                successIds.add(messageWrapper.getId()); // 成功发送，记录ID
            } catch (Exception e) {
                log.warn("消息发送失败，ID: {}，原因: {}", messageWrapper.getId(), e.getMessage());
                // 发生错误时，停止后续消息发送，确保顺序性
                break; 
            }
        }
        return successIds;
    }

    // 假设有一个包装类来获取ID和Key
    public static class MessageWrapper {
        private String id;
        private String key; // 用于Kafka消息的key，通常是业务ID
        private T value;

        public MessageWrapper(String id, String key, T value) {
            this.id = id;
            this.key = key;
            this.value = value;
        }

        public String getId() { return id; }
        public String getKey() { return key; }
        public T getValue() { return value; }
    }
}

优点与缺点

• 优点：
  • 严格顺序性：如果消息N发送失败，消息N+1将不会被发送，从而保证了在当前批次中已发送消息的严格顺序。
  • 高可靠性：每条消息都等待Broker的确认，确保了消息不丢失。
• 缺点：
  • 性能瓶颈：由于每个消息发送后都需要阻塞等待，大大降低了系统的吞吐量，尤其是在网络延迟较高或消息量巨大的场景下。
  • 资源利用率低：生产者线程在等待确认期间处于空闲状态。

在上述代码中，successIds列表将只包含在发送失败点之前成功发送并得到确认的消息ID。例如，如果发送5条消息，第3条失败，那么successIds可能只包含第1和第2条消息的ID。未发送成功的消息（第3、4、5条）将在下一次调度运行时重新处理。

异步回调式发送：性能优化的实践与顺序性权衡

为了解决同步阻塞式发送的性能问题，可以采用异步发送结合回调机制。这种方法允许生产者在发送消息后立即返回，不等待Broker的确认，而是通过回调函数处理发送结果。

工作原理

生产者将消息发送到Kafka后，不会立即阻塞。它会注册一个回调函数，当Kafka Broker返回确认结果时，回调函数会被异步调用。为了确保所有异步发送的消息都能被处理，通常会在批次发送结束后调用kafkaTemplate.flush()方法，强制将所有待发送的消息立即发送出去。

示例代码

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class AsynchronousKafkaSender {

    private static final Logger log = LoggerFactory.getLogger(AsynchronousKafkaSender.class);
    private final KafkaTemplate kafkaTemplate;

    public AsynchronousKafkaSender(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 以异步回调方式发送消息，提升性能，但可能牺牲严格的全局顺序。
     *
     * @param topicName Kafka主题名称
     * @param data      待发送的数据列表，每个元素包含一个ID和消息体
     * @return 成功发送到Kafka并得到确认的消息ID列表
     */
    public List sendMessages(String topicName, List> data) {
        // 使用线程安全的列表，因为回调可能在不同线程中执行
        List successIds = Collections.synchronizedList(new ArrayList<>()); 

        data.forEach(messageWrapper ->
            kafkaTemplate.send(topicName, messageWrapper.getKey(), messageWrapper.getValue())
                    .addCallback(new ListenableFutureCallback<>() {
                        @Override
                        public void onSuccess(SendResult result) {
                            successIds.add(messageWrapper.getId());
                        }

                        @Override
                        public void onFailure(Throwable exception) {
                            log.warn("消息异步发送失败，ID: {}，原因: {}", messageWrapper.getId(), exception.getMessage());
                            // 可以在这里添加重试逻辑或错误处理
                        }
                    }));

        // 强制刷新所有待发送的消息
        kafkaTemplate.flush(); 
        return successIds;
    }
}

kafkaTemplate.flush() 的关键作用

在异步发送模式下，kafkaTemplate.send() 实际上只是将消息放入生产者的内部缓冲区。这些消息会由后台线程批量发送到Kafka Broker。kafkaTemplate.flush() 的作用是强制清空这个缓冲区，确保所有挂起的异步发送请求都被立即发送出去，并等待它们的回调完成（或超时）。

值得注意的是，如果KafkaTemplate配置了autoflush=true，它会在每次send操作后自动刷新。然而，实践表明，这种自动刷新可能导致性能下降，甚至比手动调用flush()更慢。这可能是因为autoflush=true会强制每个send操作都进行一次同步刷新，失去了批量发送的优势。因此，在需要高性能的场景下，通常推荐关闭autoflush并手动在批次发送结束后调用flush()。

WeShop唯象

WeShop唯象是国内首款AI商拍工具，专注电商产品图片的智能生成。

下载

优点与缺点

• 优点：
  • 显著提升性能：生产者可以连续发送消息，无需等待Broker确认，极大地提高了吞吐量。
  • 资源利用率高：生产者线程可以专注于消息的生产，不被I/O等待阻塞。
• 缺点：
  • 可能打破严格的全局顺序性：如果消息N发送失败，消息N+1、N+2可能已经成功发送并得到确认。当消息N在后续重试中成功发送时，它在Kafka中的实际顺序可能晚于N+1、N+2。
  • 错误处理复杂性：需要通过回调函数异步处理成功和失败，错误处理逻辑相对复杂。

在异步回调方案中，successIds列表会包含所有成功发送到Kafka的消息ID，即使中间有消息发送失败。例如，如果发送5条消息，第3条失败，但第4、5条成功，那么successIds可能包含{1, 2, 4, 5}。第3条消息需要在下一次调度中重新处理。

两种方案的对比与选择

下表总结了两种方案的关键特性：

如何选择

选择哪种方案取决于您的业务需求对消息顺序性的严格程度以及对性能的要求：

• 如果业务场景对消息的全局顺序性有绝对严格的要求，即任何消息N的失败都必须阻止N+1的发送，直到N成功，那么同步阻塞式方案是唯一的选择。但您必须接受由此带来的性能牺牲。
• 如果业务可以容忍局部消息顺序的微小调整，即允许失败消息在后续重试中成功发送，但其在Kafka中的最终位置可能晚于原始批次中后续的成功消息，那么异步回调式方案是更优的选择。这种方案能显著提升系统吞吐量，适用于大部分高性能数据同步场景。

在大多数实际应用中，Kafka通过分区键（messageWrapper.getKey()）来保证同一分区内的消息顺序。如果您的业务逻辑允许，可以将相关联的消息发送到同一个分区，并利用Kafka的分区顺序性。对于跨分区的全局严格顺序，Kafka本身并不直接提供，需要更复杂的分布式事务或外部协调机制。本文讨论的两种方案主要关注生产者客户端在单个批次发送时的顺序行为。

总结与建议

从数据库向Kafka发送消息，并在成功后从数据库中删除，这是一个典型的“At-Least-Once”数据同步模式。在实现过程中，性能、消息不丢失和顺序性是需要仔细权衡的关键因素。

• 对于极致的严格顺序性，同步阻塞的listenableFuture.get()方案是可行的，但会严重影响性能。
• 对于高性能和高吞吐量，异步回调结合kafkaTemplate.flush()是更优的选择，但可能需要对消息的全局严格顺序性做出一定的妥协。在很多业务场景下，这种妥协是可接受的，因为最终所有消息都会被传递，只是个别消息的相对顺序可能发生微调。

在实际部署前，建议：

1. 明确业务对顺序性的要求：这是选择方案的基础。
2. 进行性能测试：在您的实际环境中，对两种方案进行基准测试，以评估其性能表现是否满足需求。
3. 考虑幂等性：为了进一步增强可靠性，可以配置Kafka生产者为幂等性（enable.idempotence=true），这有助于防止生产者重试导致的重复消息，即便在“At-Least-Once”语义下也能更接近“Exactly-Once”的保证。
4. 完善错误处理：无论选择哪种方案，都应在回调函数或try-catch块中实现健壮的错误处理、日志记录和告警机制，以便及时发现和解决消息发送失败的问题。

最终，没有绝对“最优”的解决方案，只有最适合您特定业务场景的方案。理解不同方案的权衡，才能做出明智的决策。