Kafka：生产者参数配置

清疚 2024-03-24 11:13 102阅读 0赞

**maven配置**

// 生产者
    Properties properties = new Properties();
    // 连接集群 bootstrap.servers
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.25.129:9092,192.168.25.128:9092");
    // 指定对应的key和value的序列化类型 key.serializer
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

**springboot配置**

@Configuration
    public class KafkaConfig {
     
        // 配置全局admin
        @Bean
        public KafkaAdmin kafkaAdmin() {
            Map<String， Object> configs = new HashMap<>();
            configs.put("bootstrap.servers"，"192.168.25.129:9092");
            KafkaAdmin admin = new KafkaAdmin(configs);
            return admin;
        }
    }

**配置文件**

# 用于建立初始连接的broker地址
    spring.kafka.bootstrap-servers=192.168.25.129:9092
    # producer用到的key和value的序列化类
    spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.IntegerSerializer
    spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
    # 默认的批处理记录数 16k
    spring.kafka.producer.batch-size=16384
    # 32MB的总发送缓存
    spring.kafka.producer.buffer-memory=33554432
    # consumer用到的key和value的反序列化类
    spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
    spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    # consumer的消费组id
    spring.kafka.consumer.group-id=spring-kafka-consumer-group
    # 是否自动提交消费者偏移量
    spring.kafka.consumer.enable-auto-commit=true
    # 每隔100ms向broker提交一次偏移量
    spring.kafka.consumer.auto-commit-interval=100
    # 如果该消费者的偏移量不存在，则自动设置为最早的偏移量
    spring.kafka.consumer.auto-offset-reset=earliest

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>属性</td> 
   <td>说明</td> 
  </tr> 
  <tr> 
   <td>bootstrap.servers</td> 
   <td>生产者客户端与broker集群建立初始连接需要的broker地址列表，由 该初始连接发现Kafka集群中其他的所有broker。该地址列表不需要 写全部的Kafka集群中broker的地址，但也不要写一个，以防该节点 宕机的时候不可用。形式为：host1:port1,host2:port2,... .</td> 
  </tr> 
  <tr> 
   <td>key.serializer</td> 
   <td>实现了接口 org.apache.kafka.common.serialization.Serializer的key序 列化类。</td> 
  </tr> 
  <tr> 
   <td>value.serializer</td> 
   <td>实现了接口 org.apache.kafka.common.serialization.Serializer的value 序列化类。</td> 
  </tr> 
  <tr> 
   <td>acks</td> 
   <td>该选项控制着已发送消息的持久性。 acks=0：生产者不等待broker的任何消息确认。只要将消息放到了 socket的缓冲区，就认为消息已发送。不能保证服务器是否收到该消 息，retries设置也不起作用，因为客户端不关心消息是否发送失 败。客户端收到的消息偏移量永远是-1。 acks=1：leader将记录写到它本地日志，就响应客户端确认消息， 而不等待follower副本的确认。如果leader确认了消息就宕机，则可 能会丢失消息，因为follower副本可能还没来得及同步该消息。 acks=all：leader等待所有同步的副本确认该消息。保证了只要有 一个同步副本存在，消息就不会丢失。这是最强的可用性保证。等价 于acks=-1。默认值为1，字符串。可选值：[all, -1, 0, 1]</td> 
  </tr> 
  <tr> 
   <td>compression.type</td> 
   <td>生产者生成数据的压缩格式。默认是none（没有压缩）。允许的 值：none，gzip，snappy和lz4。压缩是对整个消息批次来讲 的。消息批的效率也影响压缩的比例。消息批越大，压缩效率越好。 字符串类型的值。默认是none。</td> 
  </tr> 
  <tr> 
   <td>retry.backoff.ms</td> 
   <td>在向一个指定的主题分区重发消息的时候，重试之间的等待时间。 比如3次重试，每次重试之后等待该时间长度，再接着重试。在一些失败的场 景，避免了密集循环的重新发送请求。 long型值，默认100。可选值：[0,...]</td> 
  </tr> 
  <tr> 
   <td>retries</td> 
   <td>设置该属性为一个大于1的值，将在消息发送失败的时候重新发送消 息。该重试与客户端收到异常重新发送并无二至。允许重试但是不设 置max.in.flight.requests.per.connection为1，存在消息乱序 的可能，因为如果两个批次发送到同一个分区，第一个失败了重试， 第二个成功了，则第一个消息批在第二个消息批后。int类型的值，默 认：0，可选值：[0,...,2147483647]</td> 
  </tr> 
  <tr> 
   <td>request.timeout.ms</td> 
   <td>客户端等待请求响应的最大时长。如果服务端响应超时，则会重发请求，除非 达到重试次数。该设置应该比 replica.lag.time.max.ms (a broker configuration)要大，以免在服务器延迟时间内重发消息。int类型值，默认： 30000，可选值：[0,...]</td> 
  </tr> 
  <tr> 
   <td>interceptor.classes</td> 
   <td>在生产者接收到该消息，向Kafka集群传输之前，由序列化器处理之前，可以 通过拦截器对消息进行处理。 要求拦截器类必须实现 org.apache.kafka.clients.producer.ProducerInterceptor 接口。 默认没有拦截器。 Map configs中通过List集合配置多个拦截器类名。</td> 
  </tr> 
  <tr> 
   <td>batch.size</td> 
   <td>当多个消息发送到同一个分区的时候，生产者尝试将多个记录作为一个批来处 理。批处理提高了客户端和服务器的处理效率。 该配置项以字节为单位控制默认批的大小。 所有的批小于等于该值。 发送给broker的请求将包含多个批次，每个分区一个，并包含可发送的数 据。 如果该值设置的比较小，会限制吞吐量（设置为0会完全禁用批处理）。如果 设置的很大，又有一点浪费内存，因为Kafka会永远分配这么大的内存来参与 到消息的批整合中。</td> 
  </tr> 
  <tr> 
   <td>client.id</td> 
   <td>生产者发送请求的时候传递给broker的id字符串。 用于在broker的请求日志中追踪什么应用发送了什么消息。 一般该id是跟业务有关的字符串。</td> 
  </tr> 
  <tr> 
   <td>send.buffer.bytes</td> 
   <td>TCP发送数据的时候使用的缓冲区（SO_SNDBUF）大小。如果设置为0，则使 用操作系统默认的。</td> 
  </tr> 
  <tr> 
   <td>buffer.memory</td> 
   <td>生产者可以用来缓存等待发送到服务器的记录的总内存字节。如果记录的发送 速度超过了将记录发送到服务器的速度，则生产者将阻塞 max.block.ms 的时 间，此后它将引发异常。此设置应大致对应于生产者将使用的总内存，但并非 生产者使用的所有内存都用于缓冲。一些额外的内存将用于压缩（如果启用了 压缩）以及维护运行中的请求。long型数据。默认值：33554432，可选值： [0,...]</td> 
  </tr> 
  <tr> 
   <td>connections.max.idle.ms</td> 
   <td>当连接空闲时间达到这个值，就关闭连接。long型数据，默认：540000</td> 
  </tr> 
  <tr> 
   <td>linger.ms</td> 
   <td>生产者在发送请求传输间隔会对需要发送的消息进行累积，然后作为一个批次 发送。一般情况是消息的发送的速度比消息累积的速度慢。有时客户端需要减 少请求的次数，即使是在发送负载不大的情况下。该配置设置了一个延迟，生 产者不会立即将消息发送到broker，而是等待这么一段时间以累积消息，然 后将这段时间之内的消息作为一个批次发送。该设置是批处理的另一个上限： 一旦批消息达到了 batch.size 指定的值，消息批会立即发送，如果积累的消 息字节数达不到 batch.size 的值，可以设置该毫秒值，等待这么长时间之 后，也会发送消息批。该属性默认值是0（没有延迟）。如果设置 linger.ms=5 ，则在一个请求发送之前先等待5ms。long型值，默认：0，可 选值：[0,...]</td> 
  </tr> 
  <tr> 
   <td>max.block.ms</td> 
   <td>控制 KafkaProducer.send() 和 KafkaProducer.partitionsFor() 阻塞的 时长。当缓存满了或元数据不可用的时候，这些方法阻塞。在用户提供的序列 化器和分区器的阻塞时间不计入。long型值，默认：60000，可选值：[0,...]</td> 
  </tr> 
  <tr> 
   <td>max.request.size</td> 
   <td>单个请求的最大字节数。该设置会限制单个请求中消息批的消息个数，以免单 个请求发送太多的数据。服务器有自己的限制批大小的设置，与该配置可能不 一样。int类型值，默认1048576，可选值：[0,...]</td> 
  </tr> 
  <tr> 
   <td>partitioner.class</td> 
   <td>实现了接口 org.apache.kafka.clients.producer.Partitioner 的分区 器实现类。默认值为：org.apache.kafka. clients.producer.internals.DefaultPartitioner</td> 
  </tr> 
  <tr> 
   <td>receive.buffer.bytes</td> 
   <td>TCP接收缓存（SO_RCVBUF），如果设置为-1，则使用操作系统默认的值。 int类型值，默认32768，可选值：[-1,...]</td> 
  </tr> 
  <tr> 
   <td>security.protocol</td> 
   <td>跟broker通信的协议：PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL. string类型值，默认：PLAINTEXT</td> 
  </tr> 
  <tr> 
   <td>max.in.flight.requests.per .connection</td> 
   <td>单个连接上未确认请求的最大数量。达到这个数量，客户端阻塞。如果该值大 于1，且存在失败的请求，在重试的时候消息顺序不能保证。 int类型值，默认5。可选值：[1,...]</td> 
  </tr> 
  <tr> 
   <td>reconnect.backoff.max.ms</td> 
   <td>对于每个连续的连接失败，每台主机的退避将成倍增加，直至达到此最大值。 在计算退避增量之后，添加20％的随机抖动以避免连接风暴。 long型值，默认1000，可选值：[0,...]</td> 
  </tr> 
  <tr> 
   <td>reconnect.backoff.ms</td> 
   <td>尝试重连指定主机的基础等待时间。避免了到该主机的密集重连。该退避时间 应用于该客户端到broker的所有连接。 long型值，默认50。可选值：[0,...]</td> 
  </tr> 
 </tbody> 
</table>