巧用 Kafka 消费者组，实现精准消费控制

Kafka 消费者组基础概念

Kafka 作为一款高性能的分布式消息系统，消费者组（Consumer Group）是其实现多消费者高效协作消费的核心概念。消费者组由多个消费者实例组成，这些实例共同承担一组主题（Topic）的消费任务。

在 Kafka 中，每个分区（Partition）在同一时间只能被消费者组内的一个消费者实例消费。这确保了消息的有序性以及避免重复消费。当消费者组内新增或移除消费者实例时，Kafka 会自动进行分区的重新分配，这个过程被称为再均衡（Rebalance）。

例如，假设有一个主题 my_topic 包含 3 个分区 P0, P1, P2，同时有一个消费者组 my_group 包含 2 个消费者实例 C0 和 C1。在初始状态下，可能 C0 负责消费 P0 和 P1，C1 负责消费 P2。如果此时 C0 因为某些原因下线，Kafka 会触发再均衡，将 P0 和 P1 重新分配给 C1，从而保证所有分区都能被消费。

消费者组的优势

1. 并行消费：通过多个消费者实例共同消费，可以显著提高消费的吞吐量。对于高流量的主题，单个消费者可能无法及时处理所有消息，而消费者组可以将负载分散到多个实例上。
2. 容错性：当消费者组内某个消费者实例出现故障时，Kafka 的再均衡机制会将该实例负责的分区重新分配给其他正常的实例，确保消息不会丢失且消费能够继续进行。
3. 灵活的消费控制：不同的消费者组可以独立地消费同一个主题，每个消费者组可以根据自身需求设置不同的消费策略，如从最早的消息开始消费还是从最新的消息开始消费。

精准消费控制的需求场景

在实际的后端开发中，精准消费控制有着诸多重要的应用场景。

按业务规则分配消费

在电商系统中，订单相关的消息可能需要按照不同的业务逻辑进行处理。例如，高价值订单（金额大于一定阈值）的消息可能需要分配给专门的消费者实例进行处理，这些实例可能具备更强大的计算资源或更复杂的业务逻辑来处理高价值订单的特殊需求，如更严格的风险评估。

保证消息顺序性

对于某些业务场景，消息的顺序至关重要。例如在金融交易系统中，账户的转账操作消息必须按照先后顺序进行处理，否则可能导致账户余额出现错误。通过合理配置消费者组，可以确保特定分区的消息由固定的消费者实例按顺序消费，从而保证消息的顺序性。

数据一致性与幂等性处理

在分布式系统中，由于网络等原因，消息可能会被重复消费。精准消费控制可以帮助实现幂等性处理，即确保相同的消息无论被消费多少次，对系统产生的影响都是一致的。通过消费者组可以控制消费的频率和方式，结合业务逻辑实现数据的一致性。

Kafka 消费者组实现精准消费控制的关键机制

分区分配策略

Kafka 提供了多种分区分配策略，通过选择合适的策略可以实现精准的消费控制。

1. Range 策略：这是 Kafka 默认的分区分配策略。它将主题的分区按照顺序排列，然后平均分配给消费者组内的消费者实例。例如，对于主题 my_topic 有 6 个分区，消费者组内有 2 个消费者实例 C0 和 C1，Range 策略会将分区 P0, P1, P2 分配给 C0，P3, P4, P5 分配给 C1。这种策略在分区数量能够被消费者实例数量整除时效果较好，但如果不能整除，可能会导致某个消费者实例分配到过多的分区，从而出现负载不均衡的情况。
2. Round Robin 策略：该策略将主题的所有分区以轮询的方式依次分配给消费者组内的消费者实例。还是以 6 个分区和 2 个消费者实例为例，Round Robin 策略会将 P0, P2, P4 分配给 C0，P1, P3, P5 分配给 C1。这种策略在一定程度上可以避免负载不均衡的问题，特别是当多个主题共享消费者组时，能更均匀地分配负载。
3. Sticky 策略：Sticky 策略是 Kafka 0.11.0.0 版本引入的一种分区分配策略。它在尽量保持分区分配均匀的同时，还会尽量维持原有的分配方案，减少不必要的再均衡。例如，在再均衡时，如果某个消费者实例已经处理了一部分分区的消息，Sticky 策略会尽量让该实例继续处理这些分区，而不是重新分配给其他实例，这样可以减少因为再均衡导致的消息处理中断。

位移管理

消费者在消费消息的过程中，需要记录自己消费到的位置，这个位置被称为位移（Offset）。Kafka 提供了自动和手动两种位移管理方式。

1. 自动位移管理：在自动位移管理模式下，Kafka 消费者会定期将自己的消费位移提交到 Kafka 集群。这种方式简单方便，但可能会导致消息的重复消费或丢失。例如，如果消费者在处理完消息但还未提交位移时发生故障，重启后会从上次提交的位移处重新消费，可能会导致部分消息被重复处理。
2. 手动位移管理：手动位移管理允许开发者在代码中精确控制位移的提交时机。开发者可以在消息处理完成并且确保业务逻辑执行无误后，再手动提交位移。这样可以避免消息的重复消费和丢失，实现更精准的消费控制。但手动位移管理需要开发者更加小心，确保位移提交的正确性，否则可能会导致消费出现混乱。

代码示例：巧用 Kafka 消费者组实现精准消费控制

以下是使用 Java 语言结合 Kafka 客户端实现精准消费控制的代码示例。

引入依赖

首先，在项目的 pom.xml 文件中引入 Kafka 客户端依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

自动位移管理示例

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;

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

public class AutoOffsetConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my_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());
            }
        }
    }
}

在上述代码中，AUTO_OFFSET_RESET_CONFIG 设置为 earliest 表示从主题的最早消息开始消费。ENABLE_AUTO_COMMIT_CONFIG 设置为 true 开启自动位移提交，AUTO_COMMIT_INTERVAL_MS_CONFIG 设置为 5000 表示每 5 秒自动提交一次位移。

手动位移管理示例

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;

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

public class ManualOffsetConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my_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.commitSync();
            }
        }
    }
}

在手动位移管理示例中，ENABLE_AUTO_COMMIT_CONFIG 设置为 false 关闭自动位移提交。在处理完消息后，通过 consumer.commitSync() 方法手动提交位移。commitSync() 方法是同步提交，会等待 Kafka 集群确认位移提交成功，确保位移提交的可靠性。

自定义分区分配策略示例

要实现自定义分区分配策略，需要继承 AbstractPartitionAssignor 类并重写相关方法。

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

public class CustomPartitionAssignor extends AbstractPartitionAssignor {
    @Override
    public String name() {
        return "custom_assignor";
    }

    @Override
    public Map<String, List<TopicPartition>> assign(ConsumerGroupMetadata consumerGroupMetadata, Map<String, Integer> topicPartitionCounts) {
        Map<String, List<TopicPartition>> assignment = new HashMap<>();
        List<ConsumerGroupMember> members = consumerGroupMetadata.members();
        int numMembers = members.size();
        if (numMembers == 0) {
            return assignment;
        }

        for (String topic : topicPartitionCounts.keySet()) {
            int numPartitions = topicPartitionCounts.get(topic);
            List<TopicPartition> partitions = new ArrayList<>();
            for (int i = 0; i < numPartitions; i++) {
                partitions.add(new TopicPartition(topic, i));
            }

            List<List<TopicPartition>> partitionChunks = new ArrayList<>();
            int partitionChunkSize = (numPartitions + numMembers - 1) / numMembers;
            for (int i = 0; i < numMembers; i++) {
                int start = i * partitionChunkSize;
                int end = Math.min((i + 1) * partitionChunkSize, numPartitions);
                partitionChunks.add(partitions.subList(start, end));
            }

            for (int i = 0; i < numMembers; i++) {
                String memberId = members.get(i).memberId();
                if (!assignment.containsKey(memberId)) {
                    assignment.put(memberId, new ArrayList<>());
                }
                assignment.get(memberId).addAll(partitionChunks.get(i));
            }
        }
        return assignment;
    }
}

然后在消费者端配置使用自定义的分区分配策略：

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;

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

public class CustomAssignorConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CustomPartitionAssignor.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my_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());
            }
        }
    }
}

在上述代码中，通过 props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CustomPartitionAssignor.class.getName()); 配置使用自定义的分区分配策略 CustomPartitionAssignor。

再均衡监听器的使用

在 Kafka 消费者组中，再均衡过程可能会对业务逻辑产生影响。例如，在再均衡开始时，需要暂停一些正在进行的任务，在再均衡结束后，需要重新初始化相关资源。通过再均衡监听器（ConsumerRebalanceListener）可以实现对再均衡过程的精确控制。

再均衡监听器代码示例

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

public class RebalanceListenerConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Collections.singletonList("my_topic"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.println("Partitions revoked: " + partitions);
                // 这里可以暂停相关任务
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.println("Partitions assigned: " + partitions);
                // 这里可以重新初始化资源
            }
        });

        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.subscribe(Collections.singletonList("my_topic"), new ConsumerRebalanceListener() {... }) 为消费者注册了再均衡监听器。onPartitionsRevoked 方法在分区被撤销时调用，onPartitionsAssigned 方法在分区被分配时调用。在这两个方法中，可以根据业务需求编写相应的逻辑，如暂停或恢复任务、初始化或清理资源等。

基于消费者组的消息过滤与处理

在实际应用中，可能需要对消费到的消息进行过滤和特定处理。Kafka 消费者组可以结合业务逻辑实现这一需求。

消息过滤示例

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;

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

public class MessageFilterConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my_topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                if (record.value().contains("specific_keyword")) {
                    System.out.printf("Filtered Message - Offset: %d, Key: %s, Value: %s%n", record.offset(), record.key(), record.value());
                    // 这里可以进行进一步的处理
                }
            }
        }
    }
}

在上述代码中，通过 if (record.value().contains("specific_keyword")) 对消息进行过滤，只有包含特定关键字的消息才会被处理和输出。

复杂业务逻辑处理示例

假设我们有一个电商订单处理的场景，订单消息包含订单金额、商品信息等。我们需要根据订单金额进行不同的处理。

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;

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

public class OrderProcessingConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("order_topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                String[] parts = record.value().split(",");
                double orderAmount = Double.parseDouble(parts[0]);
                if (orderAmount > 100) {
                    System.out.println("High - value order, perform complex processing...");
                    // 高价值订单的复杂处理逻辑
                } else {
                    System.out.println("Normal order, perform simple processing...");
                    // 普通订单的简单处理逻辑
                }
            }
        }
    }
}

在上述代码中，订单消息以逗号分隔，通过解析消息获取订单金额，然后根据金额进行不同的业务逻辑处理。

消费者组与多线程消费

在一些高性能需求的场景下，单线程消费者可能无法满足消息处理的速度要求。可以通过多线程结合消费者组来提高消费的并发能力。

多线程消费者示例

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MultiThreadedConsumer {
    private static final int THREADS = 3;

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(THREADS);
        for (int i = 0; i < THREADS; i++) {
            executorService.submit(new ConsumerThread());
        }
    }

    static class ConsumerThread implements Runnable {
        @Override
        public void run() {
            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(ConsumerConfig.GROUP_ID_CONFIG, "my_group");
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Collections.singletonList("my_topic"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Thread %s - Offset: %d, Key: %s, Value: %s%n", Thread.currentThread().getName(), record.offset(), record.key(), record.value());
                }
            }
        }
    }
}

在上述代码中，创建了一个固定大小为 3 的线程池，每个线程作为一个消费者实例加入到同一个消费者组 my_group 中。这样可以利用多线程的优势提高消息的消费速度。但需要注意的是，在多线程消费时，要处理好线程安全问题，特别是在涉及共享资源的操作时。

消费者组在分布式系统中的应用

在分布式系统中，Kafka 消费者组扮演着重要的角色。例如，在微服务架构中，不同的微服务可以作为不同的消费者组消费同一个主题的消息，实现业务解耦和并行处理。

微服务架构下的消费者组应用示例

假设我们有一个电商系统，包含订单服务、库存服务和物流服务。订单创建的消息发送到 order_topic 主题。

1. 订单服务消费者组：订单服务的消费者组 order_service_group 负责消费 order_topic 中的订单创建消息，进行订单的持久化、订单状态更新等操作。

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;

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

public class OrderServiceConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order_service_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("order_topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Order Service - Processing order: " + record.value());
                // 订单持久化、状态更新等业务逻辑
            }
        }
    }
}

2. 库存服务消费者组：库存服务的消费者组 inventory_service_group 同样消费 order_topic 中的消息，根据订单中的商品信息进行库存扣减操作。

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;

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

public class InventoryServiceConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "inventory_service_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("order_topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Inventory Service - Reducing inventory for order: " + record.value());
                // 库存扣减业务逻辑
            }
        }
    }
}

3. 物流服务消费者组：物流服务的消费者组 logistics_service_group 消费 order_topic 消息，生成物流订单并安排发货。

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.common.serialization.StringDeserializer;

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

public class LogisticsServiceConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "logistics_service_group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("order_topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Logistics Service - Generating shipping order for: " + record.value());
                // 生成物流订单、安排发货等业务逻辑
            }
        }
    }
}

通过这种方式，不同的微服务作为不同的消费者组消费相同主题的消息，实现了业务的解耦和并行处理，提高了整个分布式系统的性能和可扩展性。

消费者组相关的常见问题与解决方法

在使用 Kafka 消费者组的过程中，可能会遇到一些常见问题，以下是这些问题及对应的解决方法。

再均衡频繁发生

1. 原因：
   • 消费者实例故障频繁，如网络不稳定导致消费者与 Kafka 集群的连接经常中断。
   • 分区分配策略不合理，导致每次消费者组内成员变化时都需要进行大规模的分区重新分配。
   • 消费者组内消费者实例数量动态变化过于频繁，例如在短时间内频繁添加或移除消费者实例。
2. 解决方法：
   • 检查网络环境，确保消费者与 Kafka 集群之间的网络稳定。可以增加网络监控和重试机制，当连接中断时，消费者能够快速恢复连接。
   • 评估并选择合适的分区分配策略。如果消费者组内消费者实例数量相对稳定，可以使用 Range 策略；如果需要更均匀的负载分配，特别是在多个主题共享消费者组时，可以考虑 Round Robin 策略；如果希望减少不必要的再均衡，可以使用 Sticky 策略。
   • 尽量保持消费者组内消费者实例数量的相对稳定。如果确实需要动态调整消费者实例数量，应该在系统负载较低的时间段进行，并且可以使用一些平滑的扩容或缩容策略，避免短时间内大量的实例变化。

消息重复消费

1. 原因：
   • 在自动位移管理模式下，消费者在处理完消息但还未提交位移时发生故障，重启后从上次提交的位移处重新消费，导致部分消息被重复处理。
   • 网络波动等原因导致 Kafka 集群没有正确接收到消费者的位移提交请求，而消费者认为位移已经提交成功，从而造成重复消费。
2. 解决方法：
   • 切换到手动位移管理模式，在消息处理完成并且确保业务逻辑执行无误后，再手动提交位移。例如在上述手动位移管理示例中，通过 consumer.commitSync() 方法确保位移提交的可靠性。
   • 实现幂等性处理逻辑。在业务层面，为每个消息添加唯一标识，当处理消息时，先检查该消息是否已经被处理过。如果已经处理过，则直接忽略，不进行重复处理。这样即使消息被重复消费，对系统状态也不会产生额外的影响。

消息丢失

1. 原因：
   • 消费者在处理消息时发生异常但没有进行适当的处理，导致消息没有被成功处理，而位移却被提交了。
   • Kafka 集群配置不当，如副本数量不足，当主副本发生故障时，消息可能丢失。
2. 解决方法：
   • 在消费者代码中，对消息处理过程进行异常捕获和处理。当发生异常时，不提交位移，并且可以将异常消息记录下来以便后续分析和处理。例如，可以将异常消息发送到专门的死信队列（Dead - Letter Queue）进行进一步处理。
   • 合理配置 Kafka 集群的副本数量和相关参数。确保每个分区有足够的副本，并且设置合适的 min.insync.replicas 参数，要求生产者在消息被一定数量的副本确认后才认为消息发送成功，从而提高消息的可靠性，减少消息丢失的风险。

通过对上述常见问题的分析和解决，可以更好地利用 Kafka 消费者组实现精准的消费控制，提高后端系统的稳定性和可靠性。在实际应用中，需要根据具体的业务场景和需求，灵活调整 Kafka 消费者组的配置和使用方式，以达到最佳的性能和功能效果。