Kafka学习（四）-------- Kafka核心之Producer

通过https://www.cnblogs.com/tree1123/p/11243668.html 已经对consumer有了一定的了解。producer比consumer要简单一些。

一、旧版本producer

0.9.0.0版本以前，是由scala编写的旧版本producer。

入口类：kafka.producer.producer

代码示例：

properties properties = new properties();
        properties.put("metadata.broker.list", "kafka01:9092,kafka02:9092");
        properties.put("serializer.class", "kafka.serializer.stringencoder");
        properties.put("request.requird.acks", "1");
        producerconfig config = new producerconfig(properties);
        producer<string, string> producer = new producer<string, string>(config);
        keyedmessage<string,string> msg = new keyedmessage<string,string>("topic","hello");
        producer.send(msg);

旧版本是同步机制，等待响应。吞吐性很差。在0.9.0.0版本以后，正式下架了。

旧版本的方法：

send   发送
close   关闭
sync   异步发送  有丢失消息的可能性

二、新版本producer

旧版本producer由scala编写，0.9.0.0版本以后，新版本producer由java编写。

新版本主要入口类是：org.apache.kafka.clients.producer.kafkaproducer

常用方法：

send  实现消息发送主逻辑
close  关闭producer   
metrics  获取producer的实时监控指标数据 比如发送消息的速率

kafka producer要比consumer设计简单一些，主要就是向某个topic的某个分区发送一条消息。partitioner决定向哪个分区发送消息。用户指定key，默认的分区器会根据key的哈希值来选择分区，如果没有指定key就以轮询的方式选择分区。也可以自定义分区策略。

确定分区后，producer寻找到分区的leader，也就是该leader所在的broker，然后发送消息，leader会进行副本同步isr。

producer会启两个线程，主线程封装producerrecord类，序列化后发给partitioner，然后发送到内存缓冲区。

另一个i/o线程，提取消息分batch统一发送给对应的broker。

示例代码：

properties properties = new properties();
        properties.put("bootstrap.servers", "kafka01:9092,kafka02:9092");
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.stringserializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.stringserializer");
        kafkaproducer<string, string> kafkaproducer = new kafkaproducer<string, string>(properties);
        for (int i = 1; i <= 600; i++) {
            kafkaproducer.send(new producerrecord<string, string>("z_test_20190430", "testkafka0613"+i));
            system.out.println("testkafka"+i);
        }
        kafkaproducer.close();

1、构造properties对象，bootstrap.servers key.serializer value.serializer是必须指定的。

2、使用properties构造kafkaproducer对象。

3、构造producerrecord 指定topic 分区 key value。

4、kafkaproducer的send方法发送。

5、关闭kafkaproducer。

properties主要参数：

bootstrap.servers 和consumer一样，指定部分broker即可。而且broker端如果没有配ip地址，要写成主机名。

key.serializer value.serializer 序列化参数 一定要全类名 没有key也必须设置。

acks 三个值

​ 0： producer完全不管broker的处理结果 回调也就没有用了 并不能保证消息成功发送 但是这种吞吐量最高

​ all或者-1： leader broker会等消息写入 并且isr都写入后 才会响应，这种只要isr有副本存活就肯定不会丢失，但吞 吐量最低。

​ 1： 默认的值 leader broker自己写入后就响应，不会等待isr其他的副本写入，只要leader broker存活就不会丢失，即保证了不丢失，也保证了吞吐量。

buffer.memory 缓冲区大小 字节 默认是33554432 就是发送消息的内存缓冲区大小 过小的话会影响吞吐量

compression.type 设置是否压缩消息 默认值是none 压缩后可以降低io开销提高吞吐，但是会增大cpu开销。

​ 支持三种： gzip snappy lz4 性能 lz4 > snappy > gzip

retries 发送消息重试的次数 默认0 不重试 重试可能造成重复发送 可能造成乱序

​ retry.backoff.ms 设置重试间隔 默认100毫秒

batch.size 调优重要的参数 batch小 吞吐量也会小 batch大 内存压力会大 默认值是16384 16kb

linger.ms 发送延时 默认是0 0的话不用等batch满就发送 延时的话可以提高吞吐 看具体情况进行调整

max.request.size producer能够发送最大消息的大小 默认1048576字节 如果消息很大 需要修改它

request.timeout.ms 发送请求后broker在规定时间返回 默认30秒 超过就是超时了。

send方法

fire and forget 就是上边的示例

properties properties = new properties();
        properties.put("bootstrap.servers", "kafka01:9092,kafka02:9092");
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.stringserializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.stringserializer");
        kafkaproducer<string, string> kafkaproducer = new kafkaproducer<string, string>(properties);
        for (int i = 1; i <= 600; i++) {
            kafkaproducer.send(new producerrecord<string, string>("z_test_20190430", "testkafka0613"+i));
            system.out.println("testkafka"+i);
        }
        kafkaproducer.close();

异步回调 不阻塞

properties properties = new properties();
        properties.put("bootstrap.servers", "kafka01:9092,kafka02:9092");
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.stringserializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.stringserializer");
        kafkaproducer<string, string> kafkaproducer = new kafkaproducer<string, string>(properties);
        for (int i = 1; i <= 600; i++) {
            kafkaproducer.send(new producerrecord<string, string>("z_test_20190430", "testkafka0613"+i),new callback(){
              public void oncompletion(recordmetadata metadata, exception e) {
                         if(e != null) {
                            e.printstacktrace();
                         } else {
                            system.out.println("the offset of the record we just sent is: " +       metadata.offset());
                         }
                     }           
            });
            system.out.println("testkafka"+i);
        }
        kafkaproducer.close();

同步发送 无限等待返回

producer.send（record).get()

重试机制

如果需要自定义重试机制，就要在回调里对不同异常区别对待，常见的几种如下：

可重试异常

leadernotavailableexception :分区的leader副本不可用，这可能是换届选举导致的瞬时的异常，重试几次就可以恢复
notcontrollerexception:controller主要是用来选择分区副本和每一个分区leader的副本信息，主要负责统一管理分区信息等，也可能是选举所致。

networkerexception :瞬时网络故障异常所致。

不可重试异常

serializationexception:序列化失败异常

recordtoollargeexception:消息尺寸过大导致。

示例代码：

 producer.send(myrecord,
                   new callback() {
                       public void oncompletion(recordmetadata metadata, exception e) {
                           if(e ==null){
                               //正常处理逻辑
                               system.out.println("the offset of the record we just sent is: " + metadata.offset()); 
                               
                           }else{
                                   
                                 if(e instanceof retriableexception) {
                                    //处理可重试异常
                                    ......
                                 } else {
                                    //处理不可重试异常
                                    ......
                                 }
                           }
                       }
                   });

分区机制

partitioner决定向哪个分区发送消息。用户指定key，默认的分区器会根据key的哈希值来选择分区，如果没有指定key就以轮询的方式选择分区。也可以自定义分区策略。

对于有key的消息，java版本的producer自带的partitioner会根据murmur2算法计算消息key的哈希值。然后对总分区数求模得到消息要被发送到的目标分区号。

自定义分区策略：

创建一个类，实现org.apache.kafka.clients.producer.partitioner接口

主要分区逻辑在partitioner.partition中实现：通过topic key value 一同确定分区

在构造kafkaproducer得properties中设置partitioner.class 为自定义类 注意是全类名

序列化机制

常用的serializer

bytearrayserializer.class

bytebufferserializer.class

bytesserializer.class

doubleserializer.class

integerserializer.class

longserializer.class

stringserializer.class

但是其他一些复杂的就需要自定义序列化：

1、定义数据格式

2、创建自定义序列化类，实现org.apache.kafka.common.serialization.serializer接口

3、在kafkaproducer的properties中设置key.serializer value.serializer为自定义类

以上均为单线程的情况，但producer是线程安全的，单线程适合分区较少的情况，分区较多可以多线程但对内存损耗较大。