kafka生产者知识整理

清疚 2023-07-24 11:55 38阅读 0赞

kafka的生产用来生产消息，发送到kafka。接下来，我们介绍一下kafka的生产者。

**生产者**  
kafka的生产者有3个必须的参数  
**bootstrap.servers**：该参数填写kafka服务端的地址，可以填写多个格式为：host1:port,host2:port,host3:port。填写1个就够了，kafka服务端可以根据你填写的这一个节点信息找到其他节点的信息。但是，一般配置2个，防止你填写的节点出现了故障，导致整个生产者不能用。  
**key.serializer**：该参数填写key的序列化器，这两个参数无默认值，一般可以填写kakfa的序列化器  
org.apache.kafka.common.serialization.StringSerializer  
**value.serializer**：该参数填写value的序列化器，一般可以填写kakfa的序列化器  
有了以上3个值，我们就可以连接kafka，并向kafka发送一条消息了  
示例：

public static void main(String[] args) { 
            Properties properties = new Properties();
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
            properties.put(ProducerConfig.RETRIES_CONFIG, 3);
            KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
            ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("test-topic","key","value");        
    		producer.send(producerRecord)
    		producer.close();
    }

kafka的生产者是线程安全的，可以在多个线程中共享kafka的生产者实例。

**kafka的发送模式**  
kafka一共有三种发送模式  
1).发后即忘(fire-and-forget)  
意思就是，只是将消息发送出去了，至于消息是否发送成功，不关心。显然，这种方式的性能肯定是非常高。但是在大多数的场景中，我们都会关心消息是否发送成功。  
2).同步(sync)  
同步方式，kafka发送消息时，线程会阻塞，等待kafka服务端回复。  
kafka的send方法，返回值是Future。我们可以使用Future的get方法实现等待的效果。当然，我们可以给get方法添加超时参数

Future<RecordMetadata> future = producer.send(record);
    RecordMetadata metadata = future.get();

RecordMetadata对象中封装了消息发往的主题、分区号、偏移量、时间戳等  
kakfa的消息发送不一定会成功，当消息发送出现异常时，我们可以选择是否重试，但不是所有的异常都可以重试。  
可以重试的异常：  
NetWorkException、NotCoordinatorException、LeaderNotAvailableException、UnknownTopicOrPartitionException、NotEnoughReplicasException等。  
不可重试异常：  
RecordTooLargeException。消息过大异常，这种异常是不可以重试的。  
同步发送方式非常的靠谱，线程会阻塞在这里等待服务端回复发送消息的结果，但是性能上太差了。  
3).异步(async)  
异步方式，kafka发送消息时，线程不会阻塞。此种方式，我们提供一个回调函数，让kafka在回调函数中告知我们消息是否发送成功。  
如下：

producer.send(producerRecord, new Callback() { 
                        @Override
                        public void onCompletion(RecordMetadata recordMetadata, Exception exception) { 
                            if (exception != null) { 
                                exception.printStackTrace();
                            } else { 
                                System.out.println("key发送到了["+recordMetadata.partition()+"]分区");
                            }
                        }
                    });

如果没有异常，exception为空，否则exception有值。

**分区器**  
默认情况下，如果我们在发送消息过程中，没有指定分区。key不为null的话，kafka会对key作哈希运算，得到一个分区值，相同key的消息会发往同一个分区，如果key为null，消息默认会轮询发送到主题的所有可用分区中。这里，kafka提供了分区器，可以让我们控制消息发往的分区。  
kafka提供的默认分区器是org.apache.kafka.clients.producer.internals.DefaultPartitioner.它实现了org.apache.kafka.clients.producer.Partitioner接口，接口中有2个方法

public interface Partitioner extends Configurable, Closeable { 
        int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluter);
    
        void close();
    }

partition方法，就是我们控制消息发往哪个分区的方法。返回值为int类型，即：分区号。  
close方法，用于在关闭分区器的时候，回收一些资源  
Partitioner还有一个父接口Configurable，父接口中有一个configure方法，主要用来获取配置信息及初始化数据。

**重要的生产者参数**  
**1 .acks**  
我们知道，kafka的生产者生产的消息是发往leader副本的。消息发往leader副本之后，可以通过该参数限制消息是否发送成功  
该参数的取值有3个。  
acks = 0  
acks = 1  
acks = all 或者 -1  
**当取值为0时**，生产者将消息发送出去就可以了，不需要等待服务端回复。这种配置，当kafka发生故障时，这种配置很容易引起消息丢失。但是，在某些不严格的场景，配置该值，可以让kafka的性能达到最大。  
**当取值为1时**，生产者发送的消息，leader副本成功写入就算发送成功，然后follower副本再从leader副本处拉取消息即可。这种方式还是会造成消息丢失，原因是：当leader副本成功写入消息后，follower副本还未及时同步消息，此时leader副本崩溃，新的leader副本并没有此条消息。  
**当取值为all或者-1时**，此时，leader副本需要写入消息，同时所有follower副本也要全部拉取到消息才算消息发送成功。这种配置，可以最大程度的保证消息不丢失。但也不是绝对的。因为很有可能，ISR列表中只有一个leader副本，此时这种方案就会退化为acks=0。所以，要保证消息发送的可靠性，还要配合min.insync.replicas等参数的联动。但是kafka的性能也会下降很多。  
**2 .max.request.size**  
生产者可以发送的消息的最大值。注意，此值要和kafka服务端的message.max.bytes一致。max.request.size的默认值是1M。  
**3 .retries和retry.backoff.ms**  
retries参数用来配置生产者重试的次数。当生产者发送消息失败，客户端可以进行重试以避免直接给应用程序抛出异常。重试的次数通过retries参数进行配置，retry.backoff.ms用来配置2次重试之间的时间间隔，此值的设定要估算服务端恢复正常所需的时间time，然后retry.backoff.ms要大于time的值，防止生产者客户端过早的放弃  
**4 .max.in.flight.requests.per.connection**  
kafka发送出去但是还未响应的消息会放到一个InFlightRequests中，该对象默认可以缓存5条这样未响应的消息，如果超过5条，生产者就不能向这个分区中再发送消息了。kafka的分区消息是默认保证有序的，但凡事无绝对。当前这个参数的设置，就是这个例外。比如：我们向某个主题中的某个分区发送了2个批次的消息，第一批次的消息响应较慢，进入InFlightRequests对象中，第二批次的消息响应成功。此时第一批次的消息经过重试发送成功，此时就会造成同一个分区中的分区消息乱序。如果我们要保证同一个分区中的消息绝对有序，可以将max.in.flight.requests.per.connectioin的值设定为1，当有1条消息未响应，此时就停止发送消息。  
网络环境很复杂，什么样的情况都可能出现，难免存在某些原因导致消息发送失败。如果要保证消息绝对有序，此值的设定很有必要  
**5 .commpression.type**  
默认值为：none，也就是不要说。可选值有：gzip，snappy，lz4.对消息进行压缩，可以减少网络传输，降低网络I/O。启用压缩，是一种时间换空间的方案，如果我们对消息发送的时延有一定的要求 ，不要设定此值。另外，涉及到压缩，就一定会消耗CPU，所以，如果我们机器的CPU资源已经非常紧张了，此时就不要启用压缩了。  
**6 .connections.max.idle.ms**  
这个参数用来设定多久之后关闭闲置的连接，默认是9分钟  
**7 .linger.ms**  
默认值为0。kafka发送消息，并不是一条一条的发送，这样会频繁走网络请求。kafka采用的方式为：缓存同一批次的消息，然后一次性发送。缓存的大小可以通过设定batch.size，该值默认为16KB。当消息总量超过16KB，或者时间超过linger.ms后，kakfa就开始发送消息。调整该值为非0值，可以提升一定的吞吐量，但是时延就会增大。  
**8 .receive.buffer.bytes**  
设定Socket接收消息缓冲区的大小，默认为32KB，如果设定为-1，则默认采用操作系统的值。如果我们的Producer和kafka位于不同的机房，可以适当的调大此值  
**9 .send.buffer.bytes**  
设定Socket发送消息缓冲区的大小，默认为128KB，如果设定为-1，则默认采用操作系统的值。  
**10 .replica.lag.time.max.ms**  
当超过该时间，follower还未向leader发起fetch请求  
**11 .request.timeout.ms**  
Producer等待响应的最长时间，默认值为30s。超过此值，kafka会开始重试。此配置的大小要大于replica.lag.time.max.ms的值，防止消息重复