Java实现Kafka消费者批量计数器：避免重复计数最优方案

一、写库时的锁、消费暂停与映射处理 #

不需要停止消费，但必须保证本地计数映射的线程安全，具体处理如下：

• 线程安全保障：用ConcurrentHashMap<String, AtomicInteger>这类线程安全结构维护计数，或在读写关键段加synchronized/ReentrantLock锁。消费线程负责累加计数，定时写库线程负责快照和清空，两者并发时需保证数据一致性。
• 无需停止消费：Kafka消费线程与定时写库线程相互独立，只要计数映射的读写操作是线程安全的，消费可以持续进行，不会干扰写库流程。
• 提交后清空映射：需在写库成功并提交Kafka offset后，原子性清空本地映射。注意要先对当前计数做快照，用快照执行写库，再清空原映射——这样消费线程可继续往原映射累加新数据，不会被写库操作阻塞。

二、避免重复计数的核心实现 #

从Kafka Offset提交、本地计数快照、数据库幂等性三个层面构建保障：

1. 基于Offset的精确提交 #

只有当计数成功写入数据库后，才提交对应的Kafka offset。即使消费者重启或崩溃，也只会从上次成功提交的offset处开始消费，不会重复处理已计数的消息。

2. 本地计数的原子快照 #

写库前先复制一份当前计数的快照，用快照执行写库操作，而非直接操作原映射。既避免消费线程被阻塞，也防止写库失败导致的数据丢失。

3. 数据库层面的幂等性 #

针对极端情况（如写库成功但offset提交失败，导致消息被重新消费），数据库端要保证计数更新是幂等的。例如MySQL可使用INSERT ... ON DUPLICATE KEY UPDATE语法，天然实现累加操作的幂等性。

三、Java代码示例 #

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

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public class HashtagCounterConsumer {
    private final ConcurrentHashMap<String, AtomicInteger> hashtagCountMap = new ConcurrentHashMap<>();
    private final KafkaConsumer<String, Event> kafkaConsumer;
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    private final Object lock = new Object();

    public HashtagCounterConsumer(KafkaConsumer<String, Event> consumer) {
        this.kafkaConsumer = consumer;
        // 启动每分钟一次的写库任务
        scheduler.scheduleAtFixedRate(this::flushCountsToDatabase, 1, 1, TimeUnit.MINUTES);
    }

    // 消费主循环
    public void startConsuming() {
        while (true) {
            ConsumerRecords<String, Event> records = kafkaConsumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, Event> record : records) {
                Event event = record.value();
                // 线程安全累加hashtag计数
                event.getHashtags().forEach(hashtag -> {
                    hashtagCountMap.computeIfAbsent(hashtag, k -> new AtomicInteger()).incrementAndGet();
                });
            }
        }
    }

    // 写库逻辑
    private void flushCountsToDatabase() {
        Map<String, Integer> countSnapshot;

        // 原子性获取快照并清空原映射
        synchronized (lock) {
            countSnapshot = hashtagCountMap.entrySet().stream()
                    .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().get()));
            hashtagCountMap.clear();
        }

        if (countSnapshot.isEmpty()) {
            return;
        }

        try (Connection conn = getDatabaseConnection()) {
            conn.setAutoCommit(false);
            // 幂等性SQL：存在则累加，不存在则插入
            String sql = "INSERT INTO hashtag_counts (hashtag, count) VALUES (?, ?) " +
                    "ON DUPLICATE KEY UPDATE count = count + VALUES(count)";

            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                for (Map.Entry<String, Integer> entry : countSnapshot.entrySet()) {
                    stmt.setString(1, entry.getKey());
                    stmt.setInt(2, entry.getValue());
                    stmt.addBatch();
                }
                stmt.executeBatch();
                conn.commit();
            }

            // 写库成功后同步提交Kafka offset
            kafkaConsumer.commitSync();
        } catch (SQLException e) {
            // 写库失败，将快照数据放回计数映射等待重试
            synchronized (lock) {
                countSnapshot.forEach((hashtag, count) -> {
                    hashtagCountMap.computeIfAbsent(hashtag, k -> new AtomicInteger()).addAndGet(count);
                });
            }
            // 实际场景替换为日志告警
            e.printStackTrace();
        }
    }

    // 数据库连接获取（根据实际数据源实现）
    private Connection getDatabaseConnection() throws SQLException {
        // 示例：替换为你的数据库连接逻辑
        return null;
    }

    // 事件实体类
    static class Event {
        private String photoId;
        private Iterable<String> hashtags;

        public String getPhotoId() {
            return photoId;
        }

        public Iterable<String> getHashtags() {
            return hashtags;
        }
    }
}

四、关键注意事项 #

• 线程安全细节：ConcurrentHashMap配合AtomicInteger保证单个hashtag计数的原子性，synchronized块保证快照和清空操作的原子性，避免并发冲突。
• Offset提交策略：必须使用commitSync()同步提交offset，确保offset提交与写库操作的强一致性，避免异步提交可能出现的写库成功但offset丢失的情况。
• 异常重试机制：写库失败时将快照数据放回计数映射，避免数据丢失，可额外增加重试次数限制或告警机制。
• 集群负载均衡：多消费者集群需保证Kafka主题分区数不少于消费者数，让每个消费者处理独立分区，避免重复消费同一分区消息。

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