Kafka Producer 发送数据

女爷i 2022-04-18 02:25 277阅读 0赞

# Kafka Producer 发送数据 #

## （1）生产者概览 ##

（1）不同的应用场景对消息有不同的需求，即是否允许消息丢失、重复、延迟以及吞吐量的要求。不同场景对Kafka生产者的API使用和配置会有直接的影响。  
（2）Kafka发送消息的主要步骤  
消息格式：每个消息是一个ProducerRecord对象，必须指定消息所属的Topic和消息值Value，此外还可以指定消息所属的Partition以及消息的Key。

1：序列化ProducerRecord有多个构造器，这里使用了三个参数的，topic、key、value

2：如果ProducerRecord中指定了Partition，则Partitioner不做任何事情；否则，Partitioner根据消息的key得到一个Partition。这是生产者就知道向哪个Topic下的哪个Partition发送这条消息。

3：消息被添加到相应的batch中，独立的线程将这些batch发送到Broker上

4：broker收到消息会返回一个响应。如果消息成功写入Kafka，则返回RecordMetaData对象，该对象包含了Topic信息、Patition信息、消息在Partition中的Offset信息；若失败，返回一个错误![在这里插入图片描述][overview.png]  
（3）Kafka的顺序保证。Kafka保证同一个partition中的消息是有序的，即如果生产者按照一定的顺序发送消息，broker就会按照这个顺序把他们写入partition，消费者也会按照相同的顺序读取他们。

## （2）创建Kafka生产者 ##

> Properties props = new Properties();  
> props.put(“bootstrap.servers”, “broker:port”);  
> props.put(“acks”, “all”);// acks=0 配置适用于实现非常高的吞吐量 , acks=all 这是最安全的模式  
> //发送到同一个partition的消息会被先存储在batch中，该参数指定一个batch可以使用的内存大小，单位是 byte。不一定需要等到batch被填满才能发送  
> props.put(“batch.size”, 16384);// 默认16384=16KB  
> //生产者在发送消息前等待linger.ms，从而等待更多的消息加入到batch中。如果batch被填满或者linger.ms达到上限，就把batch中的消息发送出去  
> props.put(“[linger.ms][]”, 1);// [当linger.ms][linger.ms 1]>0时，延时性会增加，但会提高吞吐量，因为会减少消息发送频率  
> // props.put(“[client.id][]”, 1);//用于标识发送消息的客户端，通常用于日志和性能指标以及配额  
> props.put(“buffer.memory”, 33554432);// 32MB  
> // Snappy压缩技术是Google开发的，它可以在提供较好的压缩比的同时，减少对CPU的使用率并保证好的性能，所以建议在同时考虑性能和带宽的情况下使用。  
> // Gzip压缩技术通常会使用更多的CPU和时间，但会产生更好的压缩比，所以建议在网络带宽更受限制的情况下使用  
> props.put(“compression.type”, “Gzip”);// 默认不压缩，该参数可以设置成snappy、gzip或lz4对发送给broker的消息进行压缩  
> // 默认值为0，当设置为大于零的值，客户端会重新发送任何发送失败的消息。  
> // 注意，此重试与客户端收到错误时重新发送消息是没有区别的。  
> // 在配置max.in.flight.requests.per.connection不等于1的情况下，允许重试可能会改变消息的顺序  
> // 因为如果两个批次的消息被发送到同一个分区，第一批消息发送失败但第二批成功，而第一批消息会被重新发送，则第二批消息会先被写入。  
> props.put(“retries”, 1);  
> props.put(“max.in.flight.requests.per.connection”, 2);// 生产者在收到服务器响应之前可以发送的消息个数  
> props.put(“key.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);  
> props.put(“value.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);  
> Producer<String, String> producer = new KafkaProducer<String, String>(props);

Kafka的生产者有如下三个必选的属性：

（1）bootstrap.servers，指定broker的地址清单

（2）key.serializer必须是一个实现org.apache.kafka.common.serialization.Serializer接口的类，将key序列化成字节数组。注意：key.serializer必须被设置，即使消息中没有指定key。

（3）value.serializer，将value序列化成字节数组

## （3）发送消息到Kafka ##

1）同步发送消息

> ProducerRecord<String, String> record = new ProducerRecord<>(“Kafka”, “Kafka\_Products”, “测试”);//Topic Key Value  
> try\{  
> Future future = producer.send(record);  
> future.get();//不关心是否发送成功，则不需要这行。  
> \} catch(Exception e) \{  
> e.printStackTrace();//连接错误、No Leader错误都可以通过重试解决；消息太大这类错误kafkaProducer不会进行任何重试，直接抛出异常  
> \}  
> 或者  
> for (int i = 0; i < 100; i++) \{  
> //ProducerRecord有多个构造器，这里使用了三个参数的，topic、key、value。  
> producer.send(new ProducerRecord<String, String>(“topic”, Integer.toString(i), Integer.toString(i)));  
> \}  
> producer.close();

2）异步发送消息

> ProducerRecord<String, String> record = new ProducerRecord<>(“Kafka”, “Kafka\_Products”, “测试”);//Topic Key Value  
> producer.send(record, new DemoProducerCallback());//发送消息时，传递一个回调对象，该回调对象必须实现org.apahce.kafka.clients.producer.Callback接口  
> private class DemoProducerCallback implements Callback \{  
> @Override  
> public void onCompletion(RecordMetadata recordMetadata, Exception e) \{  
> if (e != null) \{//如果Kafka返回一个错误，onCompletion方法抛出一个non null异常。  
> e.printStackTrace();//对异常进行一些处理，这里只是简单打印出来  
> \}  
> \}  
> \}

## （4）生产者的配置 ##

**上面我们只配置了bootstrap.servers和序列化类，其实生产者还有很多配置，上面只是使用了默认值。下面来看下这些配置参数。**

**acks**

> acks控制多少个副本必须写入消息后生产者才能认为写入成功，这个参数对消息丢失可能性有很大影响。这个参数有三种取值：

> acks=0：生产者把消息发送到broker即认为成功，不等待broker的处理结果。这种方式的吞吐最高，但也是最容易丢失消息的。  
> acks=1：生产者会在该分区的群首（leader）写入消息并返回成功后，认为消息发送成功。如果群首写入消息失败，生产者会收到错误响应并进行重试。这种方式能够一定程度避免消息丢失，但如果群首宕机时该消息没有复制到其他副本，那么该消息还是会丢失。另外，如果我们使用同步方式来发送，延迟会比前一种方式大大增加（至少增加一个网络往返时间）；如果使用异步方式，应用感知不到延迟，吞吐量则会受异步正在发送中的数量限制。  
> acks=all：生产者会等待所有副本成功写入该消息，这种方式是最安全的，能够保证消息不丢失，但是延迟也是最大的。

**buffer.memory**

> 这个参数设置生产者缓冲发送的消息的内存大小，如果应用调用send方法的速度大于生产者发送的速度，那么调用会阻塞或者抛出异常，具体行为取决于block.on.buffer.full（这个参数在0.9.0.0版本被max.block.ms代替，允许抛出异常前等待一定时间）参数。

**compresstion.type**

> 默认情况下消息是不压缩的，这个参数可以指定使用消息压缩，参数可以取值为snappy、gzip或者lz4。snappy压缩算法由Google研发，这种算法在性能和压缩比取得比较好的平衡；相比之下，gzip消耗更多的CPU资源，但是压缩效果也是最好的。通过使用压缩，我们可以节省网络带宽和Kafka存储成本。

**retries**

> 当生产者发送消息收到一个可恢复异常时，会进行重试，这个参数指定了重试的次数。在实际情况中，[这个参数需要结合retry.backoff.ms][retry.backoff.ms]（重试等待间隔）来使用，建议总的重试时间比集群重新选举群首的时间长，这样可以避免生产者过早结束重试导致失败。

**batch.size**

> 当多条消息发送到一个分区时，生产者会进行批量发送，这个参数指定了批量消息的大小上限（以字节为单位）。当批量消息达到这个大小时，生产者会一起发送到broker；但即使没有达到这个大小，生产者也会有定时机制来发送消息，避免消息延迟过大。

**[linger.ms][]**

> 这个参数指定生产者在发送批量消息前等待的时间，当设置此参数后，即便没有达到批量消息的指定大小，到达时间后生产者也会发送批量消息到broker。默认情况下，生产者的发送消息线程只要空闲了就会发送消息，即便只有一条消息。设置这个参数后，发送线程会等待一定的时间，这样可以批量发送消息增加吞吐量，但同时也会增加延迟。

**[client.id][]**

> 这个参数可以是任意字符串，它是broker用来识别消息是来自哪个客户端的。在broker进行打印日志、衡量指标或者配额限制时会用到。

**max.in.flight.requests.per.connection**

> 这个参数指定生产者可以发送多少消息到broker并且等待响应，设置此参数较高的值可以提高吞吐量，但同时也会增加内存消耗。另外，如果设置过高反而会降低吞吐量，因为批量消息效率降低。设置为1，可以保证发送到broker的顺序和调用send方法顺序一致，即便出现失败重试的情况也是如此。

**[timeout.ms][], [request.timeout.ms][], [metadata.fetch.timeout.ms][]**

> 这些参数控制生产者等待broker的响应时间。request.timeout.ms指定发送数据的等待响应时间，metadata.fetch.timeout.ms指定获取元数据（例如获取分区的群首信息）的等待响应时间。timeout.ms则指定broker在返回结果前等待其他副本（与acks参数相关）响应的时间，如果时间到了但其他副本没有响应结果，则返回消息写入失败。

**[max.block.ms][]**

> 这个参数指定应用调用send方法或者获取元数据方法（例如partitionFor）时的阻塞时间，超过此时间则抛出timeout异常。

**max.request.size**

> 这个参数限制生产者发送数据包的大小，数据包的大小与消息的大小、消息数相关。如果我们指定了最大数据包大小为1M，那么最大的消息大小为1M，或者能够最多批量发送1000条消息大小为1K的消息。另外，broker也有message.max.bytes参数来控制接收的数据包大小。在实际中，建议这些参数值是匹配的，避免生产者发送了超过broker限定的数据大小。

**receive.buffer.bytes, send.buffer.bytes**

> 这两个参数设置用来发送/接收数据的TCP连接的缓冲区，如果设置为-1则使用操作系统自身的默认值。如果生产者与broker在不同的数据中心，建议提高这个值，因为不同数据中心往往延迟比较大。

> 最后讨论下顺序保证。Kafka保证分区的顺序，也就是说，如果生产者以一定的顺序发送消息到Kafka的某个分区，那么Kafka在分区内部保持此顺序，而且消费者也按照同样的顺序消费。但是，应用调用send方法的顺序和实际发送消息的顺序不一定是一致的。举个例子，如果retries参数不为0，而max.in.flights.requests.per.session参数大于1，那么有可能第一个批量消息写入失败，但是第二个批量消息写入成功，然后第一个批量消息重试写入成功，那么这个顺序乱序的。因此，如果需要保证消息顺序，建议设置max.in.flights.requests.per.session为1，这样可以在第一个批量消息发送失败重试时，第二个批量消息需要等待。

## （5）序列化器 ##

上面提到了Kafka自带的序列化类，现在来看下如何使用其他的序列化策略

**自定义序列化**

带尺寸的  
如果我们发送的消息不是整数或者字符串时，我们需要自定义序列化策略或者使用通用的Avro、Thrift或者Protobuf这些序列化方案。下面来看下如何使用自定义的序列化方案，以及存在的问题。

假如我们要发送的消息对象是这么一个Customer：

class Custom { 
        private int customID;
        private String customerName;
        
        public Custom(int customID, String customerName) { 
            super();
            this.customID = customID;
            this.customerName = customerName;
        }
    
        public int getCustomID() { 
            return customID;
        }
    
        public String getCustomerName() { 
            return customerName;
        }
    }

那么，自定义的序列化类实现样例如下：

import org.apache.kafka.common.errors.SerializationException;
    import java.nio.ByteBuffer;
    import java.util.Map;
    
    public class CustomerSerializer implements Serializer<Customer> { 
        @Override
        public void configure(Map configs, boolean isKey) { 
            // nothing to configure
        }
    
        @Override
        /** We are serializing Customer as: 4 byte int representing customerId 4 byte int representing length of customerName in UTF-8 bytes (0 if name is Null) N bytes representing customerName in UTF-8 */
        public byte[] serialize(String topic, Customer data) { 
            try { 
                byte[] serializedName;
                int stringSize;
                if (data == null)
                    return null;
                else { 
                    if (data.getName() != null) { 
                        serializeName = data.getName().getBytes("UTF-8");
                        stringSize = serializedName.length;
                    } else { 
                        serializedName = new byte[0];
                        stringSize = 0;
                    }
                }
                ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + stringSize);
                buffer.putInt(data.getID());
                buffer.putInt(stringSize);
                buffer.put(serializedName);
                return buffer.array();
            } catch (Exception e) { 
                throw new SerializationException("Error when serializing Customer to byte[] " + e);
            } 
        }
    
        @Override
        public void close() { 
            // nothing to close
        } 
    }

我们将Customer的ID和名字进行了序列化，通过这个序列化对象，我们可以发送Customer的对象消息。但这样的序列化存在很多问题，比如想要将ID升级为Long型或者增加新的Customer域时，我们需要兼容新老消息。尤其是公司内多个团队同时消费Customer数据时，他们需要同时修改代码进行兼容。

因此，建议使用JSON、Apache Avro、Thrift或者Protobuf这些成熟的序列化/反序列化方案。下面来看下如何使用Avro来进行序列化  
使用Avro序列化  
Apache Avro是一个语言无关的序列化方案，使用Avro的数据使用语言无关的结构来描述，例如JSON。Avro一般序列化成字节文件，当然也可以序列化成JSON形式。Kafka使用Avro的好处是，当写入消息需要升级协议时，消费者读取消息可以不受影响。

//例如，原始的协议可能是这样的：
    { 
      "namespace": "customerManagement.avro",
      "type": "record",
      "name": "Customer",
      "fields": [
          { "name": "id", "type": "int"},
          { "name": "name",  "type": "string""},
          { "name": "faxNumber", "type": ["null", "string"], "default": "null"}
      ] 
    }

生成Avro对象发送到Kafka

其中，  
我们使用KafkaAvroSerializer来序列化对象，注意它可以处理原子类型，上面代码中使用其来序列化消息的key。  
schema.registry.url参数指定了注册中心的地址，我们将数据的结构存储在该注册中心。  
Customer是生成的对象，生产者发送的消息类型也为Customer。  
我们创建一个包含Customer的记录并发送，序列化的工作由KafkaAvroSerializer来完成。  
当然，我们也可以使用通用的Avrod对象而不是生成的Avro对象，对于这种情况，我们需要提供数据的结构：

Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("key.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
    props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
    props.put("schema.registry.url", schemaUrl);
    
    String schemaString = "{ \"namespace\": \"customerManagement.avro\",
                            \"type\": \"record\", " +
                            "\"name\": \"Customer\"," +
                            "\"fields\": [" +
                                "{\"name\": \"id\", \"type\": \"int\"}," +
                                 "{\"name\": \"name\", \"type\": \"string\"}," +
                                 "{\"name\": \"email\", \"type\": [\"null\",\"string\"], \"default\":\"null\" }" +
                           "]}";
    
    Producer<String, GenericRecord> producer = new KafkaProducer<String, GenericRecord>(props);
    
    Schema.Parser parser = new Schema.Parser();
    Schema schema = parser.parse(schemaString);
    
    for (int nCustomers = 0; nCustomers < customers; nCustomers++) { 
        String name = "exampleCustomer" + nCustomers;
        String email = "example " + nCustomers + "@example.com";
        GenericRecord customer = new GenericData.Record(schema);
        customer.put("id", nCustomer);
        customer.put("name", name);
        customer.put("email", email);
       
        ProducerRecord<String, GenericRecord> data = new ProducerRecord<String,GenericRecord>("customerContacts",name, customer);
        
        producer.send(data);
    }

## 分区 ##

我们创建消息的时候，必须要提供主题和消息的内容，而消息的key是可选的，当不指定key时默认为null。消息的key有两个重要的作用：1）提供描述消息的额外信息；2）用来决定消息写入到哪个分区，所有具有相同key的消息会分配到同一个分区中。

如果key为null，那么生产者会使用默认的分配器，该分配器使用轮询（round-robin）算法来将消息均衡到所有分区。

如果key不为null而且使用的是默认的分配器，那么生产者会对key进行哈希并根据结果将消息分配到特定的分区。注意的是，在计算消息与分区的映射关系时，使用的是全部的分区数而不仅仅是可用的分区数。这也意味着，如果某个分区不可用（虽然使用复制方案的话这极少发生），而消息刚好被分配到该分区，那么将会写入失败。另外，如果需要增加额外的分区，那么消息与分区的映射关系将会发生改变，因此尽量避免这种情况。

## 自定义分配器 ##

现在来看下如何自定义一个分配器，下面将key为Banana的消息单独放在一个分区，与其他的消息进行分区隔离：

public class BananaPartitioner implements Partitioner { 
        public void configure(Map<String, ?> configs) { }
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { 
        
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if ((keyBytes == null) || (!(key instanceOf String)))
            throw new InvalidRecordException("We expect all messages to have customer name as key")
        if (((String) key).equals("Banana"))
            return numPartitions; // Banana will always go to last partition
       
         // Other records will get hashed to the rest of the partitions
        return (Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1))
        }
        
        public void close() { }
     
    }

[overview.png]: /images/20220418/9386bfe51b394269a3068d686fb1b935.png
[linger.ms]: http://linger.ms
[linger.ms 1]: http://xn--linger-9y2l.ms
[client.id]: http://client.id
[retry.backoff.ms]: http://xn--retry-mn1h874a9pah76j4b5cityay5pg3v.backoff.ms
[timeout.ms]: http://timeout.ms
[request.timeout.ms]: http://request.timeout.ms
[metadata.fetch.timeout.ms]: http://metadata.fetch.timeout.ms
[max.block.ms]: http://max.block.ms