IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Consumer案例

wen java案例 3

本文目录导读：

Consumer案例

  1. 基础消费者案例
  2. 手动提交偏移量
  3. 精准一次性消费模式
  4. 多线程消费者案例
  5. 指定分区消费
  6. 消费者组动态分配监控
  7. 实战：带重试和死信队列的消费者
  8. 性能优化案例
  9. 关键配置说明

我来为你详细解析Kafka Consumer的完整案例，包括基础消费、高级特性和实战场景。

基础消费者案例

最简单的消费者

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 SimpleConsumer {
    public static void main(String[] args) {
        // 1. 配置消费者
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 2. 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 3. 订阅主题
        consumer.subscribe(Collections.singletonList("test-topic"));
        try {
            // 4. 消费消息
            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, partition = %d%n",
                        record.offset(), record.key(), record.value(), record.partition());
                }
            }
        } finally {
            consumer.close();
        }
    }
}

手动提交偏移量

精确控制提交策略

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;
public class ManualCommitConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "manual-commit-group");
        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");
        // 设置批量大小
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                // 处理所有消息
                for (ConsumerRecord<String, String> record : records) {
                    processRecord(record);
                }
                // 手动同步提交
                consumer.commitSync();
                // 或者手动异步提交 with callback
                // consumer.commitAsync((offsets, exception) -> {
                //     if (exception != null) {
                //         System.err.println("Commit failed for offsets " + offsets);
                //     }
                // });
            }
        } finally {
            try {
                // 最后同步提交，确保提交成功
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }
    }
    private static void processRecord(ConsumerRecord<String, String> record) {
        // 业务处理逻辑
        System.out.println("Processing: " + record.value());
    }
}

精准一次性消费模式

结合事务的消费者

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.time.Duration;
import java.util.*;
public class ExactlyOnceConsumer {
    private static KafkaProducer<String, String> createProducer() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "consumer-transaction-1");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.initTransactions();
        return producer;
    }
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "exactly-once-group");
        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");
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("input-topic"));
        KafkaProducer<String, String> producer = createProducer();
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                // 开启事务
                producer.beginTransaction();
                try {
                    // 处理消息并发送到输出主题
                    for (ConsumerRecord<String, String> record : records) {
                        String processedValue = process(record.value());
                        producer.send(new ProducerRecord<>("output-topic", 
                            record.key(), processedValue));
                    }
                    // 提交消费偏移量到事务
                    Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                    for (ConsumerRecord<String, String> record : records) {
                        offsets.put(new TopicPartition(record.topic(), record.partition()),
                            new OffsetAndMetadata(record.offset() + 1));
                    }
                    producer.sendOffsetsToTransaction(offsets, "exactly-once-group");
                    // 提交事务
                    producer.commitTransaction();
                } catch (Exception e) {
                    // 回滚事务
                    producer.abortTransaction();
                    throw e;
                }
            }
        } finally {
            consumer.close();
            producer.close();
        }
    }
    private static String process(String value) {
        // 业务处理
        return "processed: " + value;
    }
}

多线程消费者案例

使用线程池并发处理

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
public class MultiThreadConsumer {
    private static final int THREAD_COUNT = 4;
    private static final ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
    private static final AtomicLong totalProcessed = new AtomicLong(0);
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "multi-thread-group");
        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");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); // 5分钟
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (!records.isEmpty()) {
                    // 提交给线程池处理
                    List<Future<?>> futures = new ArrayList<>();
                    for (ConsumerRecord<String, String> record : records) {
                        Future<?> future = executor.submit(() -> {
                            try {
                                processRecord(record);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        });
                        futures.add(future);
                    }
                    // 等待所有任务完成
                    for (Future<?> future : futures) {
                        try {
                            future.get();
                        } catch (InterruptedException | ExecutionException e) {
                            e.printStackTrace();
                        }
                    }
                    // 同步提交
                    consumer.commitSync();
                    System.out.println("Total processed: " + totalProcessed.incrementAndGet());
                }
            }
        } finally {
            consumer.close();
            executor.shutdown();
        }
    }
    private static void processRecord(ConsumerRecord<String, String> record) {
        // 模拟耗时处理
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("Thread: " + Thread.currentThread().getName() + 
            " processing: " + record.value());
    }
}

指定分区消费

手动分配分区

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 PartitionAssignConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "partition-group");
        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);
        // 手动分配特定分区的特定偏移量
        TopicPartition partition0 = new TopicPartition("test-topic", 0);
        TopicPartition partition1 = new TopicPartition("test-topic", 1);
        consumer.assign(Arrays.asList(partition0, partition1));
        // 从最新偏移量开始消费
        // consumer.seekToBeginning(Arrays.asList(partition0, partition1));
        // 或从指定偏移量开始消费
        consumer.seek(partition0, 100);  // 从分区0的第100条消息开始
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Partition: %d, Offset: %d, Key: %s, Value: %s%n",
                        record.partition(), record.offset(), record.key(), record.value());
                }
                // 异步提交
                consumer.commitAsync((offsets, exception) -> {
                    if (exception != null) {
                        System.err.println("Commit failed: " + exception.getMessage());
                    }
                });
            }
        } finally {
            consumer.close();
        }
    }
}

消费者组动态分配监控

带消费者组信息的客户端

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
public class ConsumerGroupInfoExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "monitoring-group");
        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(Arrays.asList("test-topic"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.println("Partitions revoked: " + partitions);
                // 在重平衡前保存偏移量
                consumer.commitSync();
            }
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.println("Partitions assigned: " + partitions);
                // 可以在这里从外部存储恢复偏移量
            }
        });
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                // 获取当前消费者组的分配信息
                if (!records.isEmpty()) {
                    System.out.println("Group info: " + consumer.groupMetadata());
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.printf("Received: topic=%s partition=%d offset=%d key=%s value=%s%n",
                            record.topic(), record.partition(), record.offset(), 
                            record.key(), record.value());
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }
}

实战：带重试和死信队列的消费者

企业级消息处理模式

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.time.Duration;
import java.util.*;
public class EnterpriseConsumer {
    private static final int MAX_RETRIES = 3;
    private static final String DLQ_TOPIC = "test-topic-dlq";
    public static void main(String[] args) {
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "enterprise-group");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
        consumer.subscribe(Arrays.asList("test-topic"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        processWithRetry(record);
                    } catch (Exception e) {
                        System.err.println("Processing failed after retries: " + e.getMessage());
                        // 发送到死信队列
                        producer.send(new ProducerRecord<>(DLQ_TOPIC, record.key(), record.value()));
                        producer.flush();
                    }
                }
                // 确认所有消息已处理
                consumer.commitSync();
            }
        } finally {
            consumer.close();
            producer.close();
        }
    }
    private static void processWithRetry(ConsumerRecord<String, String> record) throws Exception {
        Exception lastException = null;
        for (int retry = 0; retry < MAX_RETRIES; retry++) {
            try {
                // 模拟业务处理
                System.out.println("Attempt " + (retry + 1) + ": processing " + record.value());
                processMessage(record.value());
                return; // 成功则返回
            } catch (Exception e) {
                lastException = e;
                System.err.println("Attempt " + (retry + 1) + " failed: " + e.getMessage());
                // 退避等待
                try {
                    Thread.sleep(1000 * (retry + 1));
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException("Interrupted during retry", ie);
                }
            }
        }
        throw lastException;
    }
    private static void processMessage(String message) throws Exception {
        // 模拟可能导致失败的业务逻辑
        if (message.contains("error")) {
            throw new RuntimeException("Business logic error");
        }
        System.out.println("Successfully processed: " + message);
    }
}

性能优化案例

批量处理和高吞吐配置

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;
public class OptimizedConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "optimized-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 性能优化配置
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1000");
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1024");
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "500");
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "1048576");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // 禁用消费者组协调
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
            "org.apache.kafka.clients.consumer.RoundRobinAssignor");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("high-volume-topic"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) {
                    continue;
                }
                // 批量处理
                Map<Integer, List<ConsumerRecord<String, String>>> byPartition = 
                    records.records(new TopicPartition("high-volume-topic", 0));
                // 按分区批量处理
                records.partitions().forEach(partition -> {
                    List<ConsumerRecord<String, String>> partitionRecords = 
                        records.records(partition);
                    // 批量处理逻辑
                    List<String> batchValues = partitionRecords.stream()
                        .map(ConsumerRecord::value)
                        .collect(Collectors.toList());
                    // 批量保存到数据库等
                    batchSave(batchValues);
                });
                // 自动提交由Kafka处理
            }
        } finally {
            consumer.close();
        }
    }
    private static void batchSave(List<String> values) {
        // 批量保存到数据库
        System.out.println("Batch saving " + values.size() + " records");
    }
}

关键配置说明

配置项 默认值 说明
bootstrap.servers Kafka服务器地址
group.id 消费者组ID
enable.auto.commit true 是否自动提交偏移量
auto.commit.interval.ms 5000 自动提交间隔
auto.offset.reset latest 偏移量重置策略
max.poll.records 500 每次poll最大记录数
max.poll.interval.ms 300000 最大poll间隔
session.timeout.ms 10000 会话超时时间
isolation.level read_uncommitted 事务隔离级别
  1. 使用异步提交时：确保最终能提交成功
  2. 重平衡处理：在重平衡前保存偏移量
  3. 异常处理：为不可恢复的异常准备DLQ
  4. 性能优化：合理设置批量大小和拉取间隔
  5. 监控：记录消费者组的lag等信息

这些案例覆盖了Kafka Consumer的主要使用场景，你可以根据实际需求选择合适的模式，需要我详细解释某个特定场景吗？

上一篇Supplier案例

下一篇Java函数式接口案例

相关文章

抱歉，评论功能暂时关闭!