Kafka Java客户端开发指南：从基础到高级特性

1. Kafka Java客户端开发环境准备在开始编写Kafka Java客户端代码之前我们需要先完成开发环境的搭建。这里我以kafka_2.11-0.8.2.2版本为例分享实际项目中的配置经验。1.1 依赖配置对于Maven项目需要在pom.xml中添加以下依赖dependency groupIdorg.apache.kafka/groupId artifactIdkafka_2.11/artifactId version0.8.2.2/version /dependency dependency groupIdorg.apache.kafka/groupId artifactIdkafka-clients/artifactId version0.8.2.2/version /dependency注意0.8.2.2版本是较早期的Kafka版本其API与新版本有较大差异。如果项目没有特殊要求建议使用更新的稳定版本。1.2 基础配置类创建配置文件KafkaConfig.java包含生产者和消费者的通用配置import java.util.Properties; public class KafkaConfig { public static Properties getProducerProps() { Properties props new Properties(); props.put(metadata.broker.list, localhost:9092); props.put(serializer.class, kafka.serializer.StringEncoder); props.put(request.required.acks, 1); return props; } public static Properties getConsumerProps(String groupId) { Properties props new Properties(); props.put(zookeeper.connect, localhost:2181); props.put(group.id, groupId); props.put(zookeeper.session.timeout.ms, 400); props.put(zookeeper.sync.time.ms, 200); props.put(auto.commit.interval.ms, 1000); return props; } }2. 生产者实现详解2.1 基础生产者示例下面是一个完整的Kafka生产者实现import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig; public class SimpleProducer { private final ProducerString, String producer; public SimpleProducer() { ProducerConfig config new ProducerConfig(KafkaConfig.getProducerProps()); this.producer new Producer(config); } public void send(String topic, String message) { KeyedMessageString, String data new KeyedMessage(topic, message); producer.send(data); } public void close() { producer.close(); } public static void main(String[] args) { SimpleProducer producer new SimpleProducer(); try { for(int i 0; i 100; i) { producer.send(test-topic, Message i); Thread.sleep(1000); } } catch (InterruptedException e) { e.printStackTrace(); } finally { producer.close(); } } }2.2 生产者关键参数解析metadata.broker.list指定Kafka broker地址列表多个地址用逗号分隔serializer.class消息序列化类这里使用StringEncoderrequest.required.acks消息确认机制0不等待确认1等待leader确认-1等待所有ISR副本确认实际项目中建议将生产者设计为单例模式避免频繁创建销毁带来的性能开销。3. 消费者实现详解3.1 基础消费者示例import kafka.consumer.Consumer; import kafka.consumer.ConsumerConfig; import kafka.consumer.ConsumerIterator; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import java.util.HashMap; import java.util.List; import java.util.Map; public class SimpleConsumer { private final ConsumerConnector consumer; private final String topic; public SimpleConsumer(String topic, String groupId) { this.consumer Consumer.createJavaConsumerConnector( new ConsumerConfig(KafkaConfig.getConsumerProps(groupId))); this.topic topic; } public void consume() { MapString, Integer topicCountMap new HashMap(); topicCountMap.put(topic, 1); MapString, ListKafkaStreambyte[], byte[] consumerMap consumer.createMessageStreams(topicCountMap); ListKafkaStreambyte[], byte[] streams consumerMap.get(topic); for (final KafkaStreambyte[], byte[] stream : streams) { ConsumerIteratorbyte[], byte[] it stream.iterator(); while (it.hasNext()) { System.out.println(Received: new String(it.next().message())); } } } public void close() { consumer.shutdown(); } public static void main(String[] args) { SimpleConsumer consumer new SimpleConsumer(test-topic, test-group); try { consumer.consume(); } finally { consumer.close(); } } }3.2 消费者关键参数解析zookeeper.connectZooKeeper连接地址group.id消费者组ID相同组内的消费者共享消息auto.commit.interval.ms自动提交offset的时间间隔4. 高级特性实现4.1 多线程消费者public class MultiThreadConsumer { private final ExecutorService executor; private final ConsumerConnector consumer; private final String topic; public MultiThreadConsumer(String topic, String groupId, int threadNum) { this.consumer Consumer.createJavaConsumerConnector( new ConsumerConfig(KafkaConfig.getConsumerProps(groupId))); this.topic topic; this.executor Executors.newFixedThreadPool(threadNum); } public void run() { MapString, Integer topicCountMap new HashMap(); topicCountMap.put(topic, threadNum); MapString, ListKafkaStreambyte[], byte[] consumerMap consumer.createMessageStreams(topicCountMap); ListKafkaStreambyte[], byte[] streams consumerMap.get(topic); for (final KafkaStreambyte[], byte[] stream : streams) { executor.submit(() - { ConsumerIteratorbyte[], byte[] it stream.iterator(); while (it.hasNext()) { System.out.println(Thread.currentThread().getName() : new String(it.next().message())); } }); } } public void shutdown() { if (consumer ! null) consumer.shutdown(); if (executor ! null) executor.shutdown(); } }4.2 自定义分区策略public class CustomPartitioner implements Partitioner { Override public int partition(Object key, int numPartitions) { // 自定义分区逻辑 if (key instanceof String) { return ((String) key).length() % numPartitions; } return Math.abs(key.hashCode()) % numPartitions; } } // 在生产者配置中添加 props.put(partitioner.class, com.example.CustomPartitioner);5. 常见问题与解决方案5.1 消息丢失问题现象生产者发送消息后消费者没有收到解决方案检查request.required.acks配置生产环境建议设置为-1增加重试机制props.put(message.send.max.retries, 3); props.put(retry.backoff.ms, 100);5.2 消费者重复消费现象同一条消息被消费多次解决方案确保消费者逻辑是幂等的手动管理offsetprops.put(auto.commit.enable, false); // 处理完消息后手动提交 consumer.commitOffsets();5.3 性能优化建议生产者端批量发送props.put(batch.num.messages, 200);压缩消息props.put(compression.codec, 1); // gzip压缩消费者端增加fetch大小props.put(fetch.message.max.bytes, 1048576);调整线程数根据分区数合理设置消费者线程数6. 版本兼容性说明kafka_2.11-0.8.2.2版本与新版本的主要差异API包结构不同新版使用org.apache.kafka.clients包新版使用KafkaClient代替ZooKeeper进行协调新版提供了更丰富的监控指标新版改进了消息格式和压缩算法如果考虑升级需要注意逐步迁移先升级客户端再升级服务端测试所有业务场景监控系统指标变化