本文共 6974 字，大约阅读时间需要 23 分钟。

Kafka生产者基础

Kafka生产者是将数据发布到Kafka topics的核心组件。通过合理配置Kafka生产者的方式，能够实现高效、可靠的数据发布。本文将详细介绍Kafka生产者的工作模式、代码实现以及负载均衡方案。

1.1 Producer发送模式

Kafka生产者在发送数据时，主要有以下三种发送模式：

1.2代码片段

以下是一个典型的Kafka生产者代码示例，展示了异步发送模式和异步回调发送模式的实现：

package com.item.kafka.producer;import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.clients.producer.Producer;import java.util.Properties;import java.util.concurrent.ExecutionException;public class ProducerSample {    final static String TOPIC_NAME = "demo_topic1";    public static void main(String[] args) throws ExecutionException, InterruptedException {        // 异步发送        producerSend();        // 异步阻塞（同步）发送        producerSendBlock();        // 异步发送带回调函数        producerSendWithCallback();        // 异步发送带回调函数和Partition负载均衡        producerSendWithCallbackAndPartition();    }    /**      * 异步发送     */    public static void producerSend() {        Properties properties = new Properties();        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");        properties.put(ProducerConfig.ACKS_CONFIG, "all");        properties.put(ProducerConfig.RETRIES_CONFIG, "0");        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");        Producer
   
     producer = new KafkaProducer<>(properties);        for (int i = 0; i < 10; i++) {            ProducerRecord
    
      stringStringProducerRecord = new ProducerRecord<>(TOPIC_NAME, "key" + i, "value" + i);            producer.send(stringStringProducerRecord);        }        producer.close();    }    /**      * 异步阻塞（同步）发送     */    public static void producerSendBlock() throws ExecutionException, InterruptedException {        Properties properties = new Properties();        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");        properties.put(ProducerConfig.ACKS_CONFIG, "all");        properties.put(ProducerConfig.RETRIES_CONFIG, "0");        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");        Producer
     
       producer = new KafkaProducer<>(properties);        for (int i = 0; i < 10; i++) {            String key = "key-" + i;            ProducerRecord
      
        stringStringProducerRecord = new ProducerRecord<>(TOPIC_NAME, key, "value" + i);            Future
       
         send = producer.send(stringStringProducerRecord);            RecordMetadata recordMetadata = send.get();            System.out.println(key + ":" + recordMetadata.partition() + " , offset : " + recordMetadata.offset());        }        producer.close();    }    /**      * 异步发送带回调函数     */    public static void producerSendWithCallback() {        Properties properties = new Properties();        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");        properties.put(ProducerConfig.ACKS_CONFIG, "all");        properties.put(ProducerConfig.RETRIES_CONFIG, "0");        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");        Producer
        
          producer = new KafkaProducer<>(properties); for (int i = 0; i < 10; i++) { ProducerRecord
         
           record = new ProducerRecord<>(TOPIC_NAME, "key-" + i, "value-" + i); producer.send(record, new Callback() { @Override public void onCompletion(RecordMetadata recordMetadata, Exception e) { if (e != null) { System.out.println("发送失败：" + e.getMessage()); } else { System.out.println("partition : " + recordMetadata.partition() + " , offset : " + recordMetadata.offset()); } } }); } producer.close(); } /** * 异步发送带回调函数和Partition负载均衡 */ public static void producerSendWithCallbackAndPartition() { Properties properties = new Properties(); properties.put(ProducerConfig.ACKS_CONFIG, "all"); properties.put(ProducerConfig.RETRIES_CONFIG, "0"); properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384"); properties.put(ProducerConfig.LINGER_MS_CONFIG, "1"); properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432"); properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.item.kafka.producer.SamplePartition"); Producer
          
            producer = new KafkaProducer<>(properties); for (int i = 0; i < 10; i++) { ProducerRecord
           
             record = new ProducerRecord<>(TOPIC_NAME, "key-" + i, "value-" + i); producer.send(record, new Callback() { @Override public void onCompletion(RecordMetadata recordMetadata, Exception e) { System.out.println("partition : " + recordMetadata.partition() + " , offset : " + recordMetadata.offset()); } }); } producer.close(); }}
           
          
         
        
       
      
     
    
   

自定义负载均衡方案

通过自定义负载均衡器，可以根据业务需求实现Topic的数据分区策略。在本文中，提供一个简单的负载均衡器实现：

package com.item.kafka.producer;import org.apache.kafka.clients.producer.Partitioner;import org.apache.kafka.common.Cluster;import java.util.Map;public class SamplePartition implements Partitioner {    @Override    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {        String keyStr = key + "";        String keyInt = keyStr.substring(4);        try {            int i = Integer.parseInt(keyInt);            return i % 2;        } catch (NumberFormatException e) {            return 0;        }    }    @Override    public void close() {    }    @Override    public void configure(Map
   
     configs) {    }}
   

该负载均衡器会对Topic的数据按照Key的特定规则进行分区，支持多种负载均衡策略的灵活配置。

通过上述内容，可以对Kafka生产者的工作原理、配置方式以及负载均衡方案有清晰的了解。这对于优化实际业务中的Kafka生产者配置具有重要参考价值。