Kafka（二、Kafka生产者 - 向kafka写入数据）

末蓝、 2023-06-12 04:10 4阅读 0赞

# Kakfa 生产者配置 #

由于配置太多，在此不一一列举，只显示使用率最高的几个配置，其他配置请参考官网。  
[http://kafka.apache.org/documentation/\#producerconfigs][http_kafka.apache.org_documentation_producerconfigs]

<table> 
 <thead> 
  <tr> 
   <th>NAME</th> 
   <th>DESCRIPTION</th> 
   <th>TYPE</th> 
   <th>DEFAULT</th> 
   <th>VALID VALUES</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>key.serializer</td> 
   <td>key序列化，可以使用Kafka默认的序列化和自定义序列化器</td> 
   <td>class</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>value.serializer</td> 
   <td>value序列化，可以使用Kafka默认的序列化和自定义序列化器</td> 
   <td></td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>acks</td> 
   <td>生产者需要leader确认请求完成之前接收的应答数。此配置控制了发送记录的持久性。允许以下设置：<strong>acks=0</strong> 如果设置为0，那么生产者将不等待任何消息确认。消息将立刻添加到socket缓冲区并考虑发送。在这种情况下不能保障消息被服务器接收到。并且重试机制不会生效（因为客户端不知道故障了没有）。每个消息返回的offset始终设置为-1。<strong>acks=1</strong>，这意味着leader写入消息到本地日志就立即响应，而不等待所有follower应答。在这种情况下，如果响应消息之后但follower还未复制之前leader立即故障，那么消息将会丢失。<strong>acks=all</strong> 这意味着leader将等待所有副本同步后应答消息。此配置保障消息不会丢失（只要至少有一个同步的副本）。这是最强壮的可用性保障。等价于acks=-1。</td> 
   <td>string</td> 
   <td>1</td> 
   <td>[all, -1, 0, 1]</td> 
  </tr> 
  <tr> 
   <td>bootstrap.servers</td> 
   <td>host/port列表，用于初始化建立和Kafka集群的连接。列表格式为host1:port1,host2:port2,…，无需添加所有的集群地址，kafka会根据提供的地址发现其他的地址（你可以多提供几个，以防提供的服务器关闭）</td> 
   <td>list</td> 
   <td>“”</td> 
   <td>non-null string</td> 
  </tr> 
  <tr> 
   <td>buffer.memory</td> 
   <td>生产者用来缓存等待发送到服务器的消息的内存总字节数。如果消息发送比可传递到服务器的快，生产者将阻塞max.block.ms之后，抛出异常。此设置应该大致的对应生产者将要使用的总内存，但不是硬约束，因为生产者所使用的所有内存都用于缓冲。一些额外的内存将用于压缩（如果启动压缩），以及用于保持发送中的请求。一个合理的内存大小。</td> 
   <td>long</td> 
   <td>33554432</td> 
   <td>[0,…]</td> 
  </tr> 
  <tr> 
   <td>compression.type</td> 
   <td>生产者生成的所有数据的压缩类型。默认值为无（即不压缩）。有效值为<strong>none</strong>，<strong>gzip</strong>，<strong>snappy</strong>，<strong>lz4</strong>，或<strong>zstd</strong>。压缩是指全部数据的压缩，因此，批处理的效率也会影响压缩率（批处理越多意味着压缩越好）。</td> 
   <td>string</td> 
   <td>none</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>retries</td> 
   <td>设置一个比零大的值，客户端如果发送失败则会重新发送。注意，这个重试功能和客户端在接到错误之后重新发送没什么不同。如果max.in.flight.requests.per.connection没有设置为1，有可能改变消息发送的顺序，因为如果2个批次发送到一个分区中，并第一个失败了并重试，但是第二个成功了，那么第二个批次将超过第一个。“retries”和“retries.backoff.ms”决定了重试机制，也就是如果一个请求失败了可以重试几次，每次重试的间隔是多少毫秒。这个大家适当设置几次重试的机会，给一定的重试间隔即可，比如给100ms的重试间隔。 int</td> 
   <td>2147483647</td> 
   <td>[0,…,2147483647]</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>batch.size</td> 
   <td>当多个消息要发送到相同分区的时，生产者尝试将消息批量打包在一起，以减少请求交互。这样有助于客户端和服务端的性能提升。该配置的默认批次大小（以字节为单位）：不会打包大于此配置大小的消息。发送到broker的请求将包含多个批次，每个分区一个，用于发送数据。</td> 
   <td>int</td> 
   <td>16384</td> 
   <td>[0,…]</td> 
  </tr> 
  <tr> 
   <td>ssl.key.password</td> 
   <td>密钥存储文件中私钥的密码。这对于客户端是可选的。</td> 
   <td>password</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.keystore.location</td> 
   <td>密钥存储文件的位置这对于客户端是可选的，可以用于客户端的双向身份验证。</td> 
   <td>string</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.keystore.password</td> 
   <td>密钥存储文件的存储密码。这对于客户端是可选的，并且仅在配置了ssl.keystore.location时才需要。</td> 
   <td>password</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.truststore.location</td> 
   <td>信任存储和文件的位置。</td> 
   <td>string</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.truststore.password</td> 
   <td>信任存储文件的密码。如果未设置密码，则仍然可以访问信任库，但是将禁用完整性检查。</td> 
   <td>password</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

# 代码示例 #

这里没有使用最后几个ssl配置，内网默认都是网络安全的。

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    public class ProducerDemo { 
    
        private final KafkaProducer<String, String> producer;
    
        public final static String TOPIC = "test5";
    
        private ProducerDemo() { 
            Properties props = new Properties();
    
            props.put("bootstrap.servers", "127.0.0.1:9092");
            // 所有follower都响应了才认为消息提交成功，即"committed"
            props.put("acks", "all");
            // retries = MAX 无限重试，直到你意识到出现了问题:)
            props.put("retries", 0);
            // producer将试图批处理消息记录，以减少请求次数.默认的批量处理消息字节数
            props.put("batch.size", 16384);
            // batch.size当批量的数据大小达到设定值后，就会立即发送，不顾下面的linger.ms,
            // 延迟1ms发送，这项设置将通过增加小的延迟来完成--即，不是立即发送一条记录，producer将会等待给定的延迟时间以允许其他消息记录发送，这些消息记录可以批量处理
            props.put("linger.ms", 1);
            // producer可以用来缓存数据的内存大小。
            props.put("buffer.memory", 33554432);
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            producer = new KafkaProducer<>(props);
        }
    
        public void produce() { 
            int messageNo = 1;
            final int COUNT = 5;
    
            while (messageNo < COUNT) { 
    
                String key = String.valueOf(messageNo);
                String data = String.format("hello KafkaProducer message %s from hubo 06291018 ", key);
    
                try { 
                    // 返回值为Future，现在为不关心是否发送成功。还有同步发送，异步发送两种。
                    producer.send(new ProducerRecord<String, String>(TOPIC,",", data));
                } catch (Exception e) { 
                    e.printStackTrace();
                }
    
                messageNo++;
            }
    
            producer.close();
        }
    
        public static void main(String[] args) { 
            new ProducerDemo().produce();
        }
    }

**自定义分区器**  
首先自定义个分区器，一定要实现`Partitioner`接口。

import org.apache.kafka.clients.producer.Partitioner;
    import org.apache.kafka.common.Cluster;
    import org.apache.kafka.common.PartitionInfo;
    import org.apache.kafka.common.record.InvalidRecordException;
    import org.apache.kafka.common.utils.Utils;
    
    import java.util.List;
    import java.util.Map;
    
    public class CustomPartitioner implements Partitioner { 
    
        String keyName;
    
        @Override
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { 
            List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic);
            int numPartitions = partitionInfos.size();
    
            if ((keyBytes == null) || (!(key instanceof String))) { 
                throw new InvalidRecordException("We expect all messages to have customer name as key");
            }
    
            if (((String) key).equalsIgnoreCase(keyName)) { 
                return numPartitions;
            }
    
            return (Math.abs(Utils.murmur2(keyBytes))) % (numPartitions - 1);
    
        }
    
        @Override
        public void close() { 
    
        }
    
        @Override
        public void configure(Map<String, ?> configs) { 
            keyName = (String) configs.get("");
        }
    }

然后将自定义分区器添加到配置中：

//设置自定义分区
            props.put("partitioner.class", "com.study.Kafka.common.CustomPartitioner");

[http_kafka.apache.org_documentation_producerconfigs]: http://kafka.apache.org/documentation/#producerconfigs