首頁手記 Kafka重復消費學習：確保消息處理的準確性

Kafka重復消費學習：確保消息處理的準確性

標簽：

大數據 Hadoop Spark

概述

本文深入探讨了Kafka重复消费学习的相关内容，包括系统异常和消费者组变更导致的消息重复问题，以及由此引发的数据不一致和业务逻辑执行错误。文章还详细介绍了通过数据库事务保障、序列号或时间戳策略以及消费端幂等性实现等方法来解决Kafka重复消费的问题。Kafka重复消费学习对于确保消息处理的准确性至关重要。

Kafka简介

Kafka的基本概念

Apache Kafka 是一个分布式流处理平台，最初由 LinkedIn 开发，用于构建实时数据管道和流式应用程序。Kafka 可以被看作一个分布式的发布订阅系统，它提供了一个可扩展且容错的数据流处理框架。Kafka 由一系列分布式的服务组成，这些服务可以在一个集群中运行，提供高效且持久的数据流服务。

Kafka 的核心组件是主题（Topic），主题是一个分类的名字，生产者（Producer）通过这个名字将消息发布到 Kafka 集群中。消费者（Consumer）通过订阅一个或多个主题来获取消息。Kafka 中的数据以消息（Message）的形式存在，每条消息都有一个键（Key）、值（Value）和时间戳（Timestamp）。

消息模型

Kafka 中的消息模型如下所示：

public class KafkaMessage {
    private String key;
    private String value;
    private long timestamp;

    public KafkaMessage(String key, String value, long timestamp) {
        this.key = key;
        this.value = value;
        this.timestamp = timestamp;
    }

    public String getKey() {
        return key;
    }

    public String getValue() {
        return value;
    }

    public long getTimestamp() {
        return timestamp;
    }
}

这个模型展示了消息的基本组成部分：键、值和时间戳。这些信息对于消息的生产和消费都非常重要。

Kafka的特点和应用场景

Kafka 的设计具有以下特点：

高吞吐量：Kafka 能够处理每秒数千条消息，吞吐量可以轻松扩展到每秒数百万条消息。
持久性：Kafka 把消息持久化到磁盘上，确保消息不会因为程序挂掉或机器重启而丢失。
分布式：Kafka 可以在多个服务器上部署，具有良好的扩展能力。
容错性：Kafka 能够在集群中容忍部分节点故障，保证消息不丢失。
实时处理：Kafka 能够支持实时数据流处理，可以与其他流处理系统集成。

Kafka 的应用场景包括：

日志聚合：收集来自不同服务器的日志文件，并将它们存储在 Kafka 中，然后进行实时分析或批量处理。
流处理：将实时数据流处理任务集成到业务流程中，实现数据的实时分析。
数据集成：连接多个数据源和数据仓库，实现数据的无缝传输。
消息代理：作为中间件，连接各种服务或应用，实现异步通信。

示例代码

以下是一个简单的 Java 示例，展示如何创建一个 Kafka 生产者和消费者：

生产者代码

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        // 配置生产者
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // 发送消息
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("test-topic", "key-" + i, "value-" + i));
        }

        // 关闭生产者
        producer.close();
    }
}

消费者代码

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 配置消费者
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"));

        // 消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }

        // 关闭消费者
        consumer.close();
    }
}

Kafka消息消费机制

消息生产和消费的基本流程

在 Kafka 中，消息的生产和消费遵循一定的流程：

生产者（Producer）：生产者负责将消息发布到 Kafka 主题（Topic）中。生产者通常向 Kafka 集群中的一个或多个 Broker 发送消息。
主题（Topic）：生产者发送的消息会被分配到一个或多个分区（Partition）中。每个分区是一个线性、有序的数据流。
消费组（Consumer Group）：消费者通过订阅主题来获取消息。每个消费者属于一个消费组，消费组中的消费者可以并行消费消息。
消费者（Consumer）：消费者从 Kafka 中读取消息，并进行相应的业务处理。每个消费者可以订阅多个主题。

生产者与消费者的交互

生产者和消费者的交互流程如下：

生产者发送消息：生产者将消息发送到 Kafka 主题，每个消息会包含一个键（Key）和一个值（Value）。
消息分发到分区：根据键的哈希值，消息会被分发到不同的分区中。每个分区中的消息是有序的。
消费者组管理：Kafka 使用消费组来管理消费者。每个消费者组中的消费者会均匀地消费分区中的消息。
消费者读取消息：消费者从 Kafka 中读取消息，并进行业务处理。

消费者组的概念

消费者组（Consumer Group）是 Kafka 中用来管理多个消费者的机制。每个消费者组由一组消费者组成，这些消费者可以并行消费同一个主题（Topic）中的消息。消费者组的主要特点包括：

负载均衡：消费者组可以实现负载均衡，每个消费者可以均匀地消费分区中的消息。
容错性：如果一个消费者失败，其他消费者会自动接管该消费者的分区。
消息重复消费：消费者组的变更可能导致消息的重复消费。

示例代码

以下是一个简单的 Java 示例，展示如何创建和配置消费者组：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class ConsumerGroupExample {
    public static void main(String[] args) {
        // 配置消费者
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"));

        // 消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }

        // 关闭消费者
        consumer.close();
    }
}

消费者组的变更

消费者组的变更可能导致以下行为：

分区再均衡：当消费者组中的消费者发生变化时，Kafka 会重新分配分区。这可能导致消费者接收到重复的消息。
消息偏移量：Kafka 使用偏移量（Offset）来跟踪每个消费者已经消费的消息。如果消费者组发生变化，可能会导致偏移量的不一致，从而引起重复消费。

示例代码

以下是一个简单的 Java 示例，展示消费者变化时发生的消息重复：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class ConsumerChangesExample {
    public static void main(String[] args) {
        // 配置消费者
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"));

        // 消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }

            // 模拟消费者变化
            if (records.count() > 0) {
                Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                for (TopicPartition partition : records.partitions()) {
                    offsets.put(partition, new OffsetAndMetadata(records.offsets().get(partition).endOffset()));
                }

                // 提交偏移量
                consumer.commitSync(offsets);

                // 假设这里模拟消费者组变化，可能会导致偏移量不一致
                // 消费者变化后重新分配分区
            }
        }

        // 关闭消费者
        consumer.close();
    }
}

重复消费问题的成因

系统异常导致的消息重复

系统异常是导致消息重复消费的常见原因。例如：

网络故障：网络不稳定可能导致消息在网络传输过程中丢失或重复。
硬件故障：硬件故障可能导致消息在传输或存储过程中丢失或重复。
软件异常：软件异常可能导致消费者在处理消息时失败，从而导致消息被重新发送。

网络故障示例

假设网络突然中断，生产者发送的消息在网络传输过程中丢失，导致消费者接收到重复的消息。

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class NetworkFaultExample {
    public static void main(String[] args) {
        // 配置生产者
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // 发送消息
        producer.send(new ProducerRecord<>("test-topic", "key-1", "value-1"));

        // 模拟网络故障
        // 这里可以插入代码模拟网络故障，例如关闭网络连接

        // 重新发送消息
        producer.send(new ProducerRecord<>("test-topic", "key-1", "value-1"));

        // 关闭生产者
        producer.close();
    }
}

消费者组变更引发的重复

消费者组变更也是导致消息重复消费的常见原因。例如：

消费者增加：当消费者组中的消费者增加时，Kafka 会重新分配分区，可能导致消费者接收到重复的消息。
消费者减少：当消费者组中的消费者减少时，Kafka 会重新分配分区，可能导致消费者接收到重复的消息。
消费者失败：当消费者失败时，Kafka 会重新分配分区，可能导致消费者接收到重复的消息。

消费者增加示例

假设消费者组中的消费者增加，Kafka 会重新分配分区，导致消费者接收到重复的消息。

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class ConsumerIncreaseExample {
    public static void main(String[] args) {
        // 配置消费者
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"));

        // 消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }

            // 模拟消费者增加
            // 这里可以插入代码模拟增加消费者，Kafka 会重新分配分区
        }

        // 关闭消费者
        consumer.close();
    }
}

重复消费问题的影响

数据不一致的问题

重复消费可能导致数据不一致的问题。例如：

计数器：如果一个消息被重复消费，可能导致计数器的值增加过多，导致数据不一致。
账户余额：如果一个消息被重复消费，可能导致账户余额的增加过多，导致数据不一致。
订单状态：如果一个消息被重复消费，可能导致订单状态的更新错误，导致数据不一致。

计数器示例

假设一个消息被重复消费，导致计数器的值增加过多。

public class Counter {
    private int count;

    public void increment() {
        count++;
    }

    public int getCount() {
        return count;
    }
}

public class DuplicateMessageExample {
    public static void main(String[] args) {
        Counter counter = new Counter();

        // 发送消息
        counter.increment();
        counter.increment(); // 重复消费

        System.out.println("Counter: " + counter.getCount());
    }
}

业务逻辑执行的错误

重复消费还可能导致业务逻辑执行的错误。例如：

订单处理：如果一个消息被重复消费，可能导致订单处理的错误，例如重复发货。
库存更新：如果一个消息被重复消费，可能导致库存更新的错误，例如重复减少库存。
用户信息更新：如果一个消息被重复消费，可能导致用户信息更新的错误，例如重复增加用户的积分。

订单处理示例

假设一个消息被重复消费，导致订单处理的错误，例如重复发货。

public class Order {
    private String orderId;
    private int quantity;

    public Order(String orderId, int quantity) {
        this.orderId = orderId;
        this.quantity = quantity;
    }

    public void shipOrder() {
        // 处理订单，例如发货
        System.out.println("Shipping order: " + orderId + ", quantity: " + quantity);
    }
}

public class DuplicateMessageExample {
    public static void main(String[] args) {
        Order order = new Order("12345", 1);

        // 发送消息
        order.shipOrder();
        order.shipOrder(); // 重复消费

        System.out.println("Order shipped.");
    }
}

解决重复消费的方法

使用数据库事务保障

使用数据库事务保障是一种常见的方法，可以确保消息只被消费一次。具体做法包括：

检查消息是否已被消费：在消费消息之前，检查消息是否已被消费。如果已经被消费，则跳过该消息。
标记消息已被消费：在消费消息之后，标记消息已被消费。这可以防止消息被重复消费。

检查消息是否已被消费

以下是一个简单的示例，展示了如何在消费消息之前检查消息是否已被消费。

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class MessageConsumer {
    private Connection connection;

    public void consumeMessage(String message) {
        // 检查消息是否已被消费
        String query = "SELECT * FROM messages WHERE message = ?";
        try (PreparedStatement stmt = connection.prepareStatement(query)) {
            stmt.setString(1, message);
            ResultSet rs = stmt.executeQuery();
            if (rs.next()) {
                // 消息已被消费，跳过该消息
                return;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 处理消息
        processMessage(message);

        // 标记消息已被消费
        String insertQuery = "INSERT INTO messages (message) VALUES (?)";
        try (PreparedStatement insertStmt = connection.prepareStatement(insertQuery)) {
            insertStmt.setString(1, message);
            insertStmt.executeUpdate();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void processMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Processing message: " + message);
    }
}

序列号或时间戳策略

序列号或时间戳策略也是一种常见的方法，可以确保消息只被消费一次。具体做法包括：

使用序列号：为每个消息分配一个唯一的序列号。在消费消息之前，检查消息的序列号是否已被消费。
使用时间戳：为每个消息分配一个时间戳。在消费消息之前，检查消息的时间戳是否已被消费。

使用序列号

以下是一个简单的示例，展示了如何使用序列号来确保消息只被消费一次。

import java.util.HashMap;
import java.util.Map;

public class MessageConsumer {
    private Map<Long, Boolean> consumedMessages = new HashMap<>();

    public void consumeMessage(long sequenceNumber, String message) {
        // 检查消息是否已被消费
        if (consumedMessages.get(sequenceNumber)) {
            // 消息已被消费，跳过该消息
            return;
        }

        // 处理消息
        processMessage(message);

        // 标记消息已被消费
        consumedMessages.put(sequenceNumber, true);
    }

    private void processMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Processing message: " + message);
    }
}

使用时间戳

以下是一个简单的示例，展示了如何使用时间戳来确保消息只被消费一次。

import java.util.Map;
import java.util.TreeMap;

public class MessageConsumer {
    private TreeMap<Long, Boolean> consumedMessages = new TreeMap<>();

    public void consumeMessage(long timestamp, String message) {
        // 检查消息是否已被消费
        if (consumedMessages.containsKey(timestamp)) {
            // 消息已被消费，跳过该消息
            return;
        }

        // 处理消息
        processMessage(message);

        // 标记消息已被消费
        consumedMessages.put(timestamp, true);
    }

    private void processMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Processing message: " + message);
    }
}

消费端幂等性实现

消费端幂等性实现是一种高级的方法，可以确保消息只被消费一次。具体做法包括：

唯一标识消息：为每个消息分配一个唯一标识符。在消费消息之前，检查消息的唯一标识符是否已被消费。
处理幂等性逻辑：在处理消息时，确保幂等性逻辑能够处理重复的消息。

唯一标识消息

以下是一个简单的示例，展示了如何使用唯一标识符来确保消息只被消费一次。

import java.util.HashMap;
import java.util.Map;

public class MessageConsumer {
    private Map<String, Boolean> consumedMessages = new HashMap<>();

    public void consumeMessage(String uniqueId, String message) {
        // 检查消息是否已被消费
        if (consumedMessages.get(uniqueId)) {
            // 消息已被消费，跳过该消息
            return;
        }

        // 处理消息
        processMessage(message);

        // 标记消息已被消费
        consumedMessages.put(uniqueId, true);
    }

    private void processMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Processing message: " + message);
    }
}

幂等性逻辑

以下是一个简单的示例，展示了如何实现幂等性逻辑来处理重复的消息。

public class MessageConsumer {
    public void consumeMessage(String message) {
        // 检查消息是否已被处理
        if (isMessageProcessed(message)) {
            // 消息已被处理，跳过该消息
            return;
        }

        // 处理消息
        processMessage(message);

        // 标记消息已被处理
        markMessageAsProcessed(message);
    }

    private boolean isMessageProcessed(String message) {
        // 实现检查消息是否已被处理的逻辑
        return false;
    }

    private void processMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Processing message: " + message);
    }

    private void markMessageAsProcessed(String message) {
        // 实现标记消息已被处理的逻辑
    }
}

实践案例与代码示例

如何在实际开发中避免重复消费

在实际开发中，避免重复消费的方法有很多，具体选择哪种方法取决于具体的业务场景和需求。以下是一些常见的实践案例：

使用数据库事务保障：通过数据库事务保障可以确保消息只被消费一次。这适用于需要持久化存储的场景。
使用序列号或时间戳策略：通过序列号或时间戳策略可以确保消息只被消费一次。这适用于不需要持久化存储的场景。
实现消费端幂等性：通过实现消费端幂等性可以确保消息只被消费一次。这适用于需要幂等性处理的场景。

使用数据库事务保障的示例

以下是一个简单的示例，展示了如何使用数据库事务保障来避免重复消费。

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;

public class DatabaseTransactionExample {
    private Connection connection;

    public void consumeMessage(String message) {
        // 检查消息是否已被消费
        String query = "SELECT * FROM messages WHERE message = ?";
        try (PreparedStatement stmt = connection.prepareStatement(query)) {
            stmt.setString(1, message);
            ResultSet rs = stmt.executeQuery();
            if (rs.next()) {
                // 消息已被消费，跳过该消息
                return;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 处理消息
        processMessage(message);

        // 标记消息已被消费
        String insertQuery = "INSERT INTO messages (message) VALUES (?)";
        try (PreparedStatement insertStmt = connection.prepareStatement(insertQuery)) {
            insertStmt.setString(1, message);
            insertStmt.executeUpdate();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void processMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Processing message: " + message);
    }
}

使用序列号的示例

以下是一个简单的示例，展示了如何使用序列号来避免重复消费。

import java.util.HashMap;
import java.util.Map;

public class SequenceNumberExample {
    private Map<Long, Boolean> consumedMessages = new HashMap<>();

    public void consumeMessage(long sequenceNumber, String message) {
        // 检查消息是否已被消费
        if (consumedMessages.get(sequenceNumber)) {
            // 消息已被消费，跳过该消息
            return;
        }

        // 处理消息
        processMessage(message);

        // 标记消息已被消费
        consumedMessages.put(sequenceNumber, true);
    }

    private void processMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Processing message: " + message);
    }
}

使用时间戳的示例

以下是一个简单的示例，展示了如何使用时间戳来避免重复消费。

import java.util.Map;
import java.util.TreeMap;

public class TimestampExample {
    private TreeMap<Long, Boolean> consumedMessages = new TreeMap<>();

    public void consumeMessage(long timestamp, String message) {
        // 检查消息是否已被消费
        if (consumedMessages.containsKey(timestamp)) {
            // 消息已被消费，跳过该消息
            return;
        }

        // 处理消息
        processMessage(message);

        // 标记消息已被消费
        consumedMessages.put(timestamp, true);
    }

    private void processMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Processing message: " + message);
    }
}

实现消费端幂等性的示例

以下是一个简单的示例，展示了如何实现消费端幂等性来避免重复消费。

public class IdempotentConsumer {
    public void consumeMessage(String uniqueId, String message) {
        // 检查消息是否已被消费
        if (isMessageProcessed(uniqueId)) {
            // 消息已被消费，跳过该消息
            return;
        }

        // 处理消息
        processMessage(message);

        // 标记消息已被消费
        markMessageAsProcessed(uniqueId);
    }

    private boolean isMessageProcessed(String uniqueId) {
        // 实现检查消息是否已被消费的逻辑
        return false;
    }

    private void processMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Processing message: " + message);
    }

    private void markMessageAsProcessed(String uniqueId) {
        // 实现标记消息已被消费的逻辑
    }
}

使用Java或Python示例代码演示

使用Java演示

以下是一个使用 Java 实现的消息消费示例，展示了如何避免重复消费。

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 配置消费者
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"));

        // 消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 处理消息前，检查是否已处理过
                if (!isMessageProcessed(record)) {
                    processMessage(record);
                    // 标记消息已被处理
                    markMessageAsProcessed(record);
                }
            }
        }

        // 关闭消费者
        consumer.close();
    }

    private static boolean isMessageProcessed(ConsumerRecord<String, String> record) {
        // 实现检查消息是否已被处理的逻辑
        return false;
    }

    private static void processMessage(ConsumerRecord<String, String> record) {
        // 处理消息的逻辑
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }

    private static void markMessageAsProcessed(ConsumerRecord<String, String> record) {
        // 实现标记消息已被处理的逻辑
    }
}

使用Python演示

以下是一个使用 Python 实现的消息消费示例，展示了如何避免重复消费。

from kafka import KafkaConsumer

# 配置消费者
consumer = KafkaConsumer('test-topic', bootstrap_servers='localhost:9092')

# 消费消息
for message in consumer:
    # 处理消息前，检查是否已处理过
    if not is_message_processed(message):
        process_message(message)
        # 标记消息已被处理
        mark_message_as_processed(message)

def is_message_processed(message):
    # 实现检查消息是否已被处理的逻辑
    return False

def process_message(message):
    # 处理消息的逻辑
    print(f'offset = {message.offset()}, key = {message.key()}, value = {message.value}')

def mark_message_as_processed(message):
    # 实现标记消息已被处理的逻辑
    pass

通过这些示例代码，可以了解如何在实际开发中避免重复消费并确保消息处理的准确性。

點擊查看更多內容

為 TA 點贊

若覺得本文不錯，就分享一下吧！

評論

評論

共同學習，寫下你的評論

評論加載中...

展開查看更多評論

作者其他優質文章

正在加載中

慕容森

手記
篇

粉絲

183

獲贊與收藏

650

關注作者，訂閱最新文章

閱讀免費教程

后端通用面試教程

41個小節 32074 358

網絡編程入門教程

20個小節 13196 249

Pandas 入門教程

25個小節 19564 369

推薦

評論

收藏

共同學習，寫下你的評論



感謝您的支持，我會繼續努力的～

掃碼打賞，你說多少就多少

贊賞金額會直接到老師賬戶

支付方式

打開微信掃一掃，即可進行掃碼打賞哦

今天注冊有機會得

100積分直接送

付費專欄免費學

大額優惠券免費領

立即參與放棄機會

點擊
抽獎

慕課手記新用戶專享福利

恭喜你，你的運氣太好了，居然抽中了 100個積分！

恭喜你，抽中了價值元的專欄！

太棒了，直接落到你賬戶里！

積分商城里的羅技鼠標、機械鍵盤、
Kindle 閱讀器、小米平衡車
Apple iPad （10.2英寸）、大額優惠券
在等著你去兌換了噢

作者：

免費贈送

兌換碼：1111222211 復制

優惠券可用于購買實戰課、體系課
無門檻使用

先去看看，有什么好東西馬上兌換我愛學習，選課去


亚洲在线久爱草,狠狠天天香蕉网,天天搞日日干久草,伊人亚洲日本欧美

熱搜

最近搜索清空

Kafka重復消費學習：確保消息處理的準確性

消息模型

示例代码

生产者代码

消费者代码

生产者与消费者的交互

消费者组的概念

示例代码

示例代码

网络故障示例

消费者增加示例

计数器示例

订单处理示例

检查消息是否已被消费

使用序列号

使用时间戳

唯一标识消息

幂等性逻辑

使用数据库事务保障的示例

使用序列号的示例

使用时间戳的示例

实现消费端幂等性的示例

使用Java演示

使用Python演示

閱讀免費教程