Kafka之Producer API详解

妖狐艹你老母 2022-11-29 12:28 211阅读 0赞

Kafka中如何创建生产者producer已经在前面给大家详细的讲解过，那么如何使用JAVA来创建并发送数据到topic中呢，今天就着重这一块说说。  
先创建一个topic，拥有一个副本和一个分区

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

示例：

public class MyProducer { 
        public static void main(String[] args) { 
            Properties prop = new Properties();
            prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.5.150:9092");
          //或者
            prop.put("bootstrap.servers","192.168.5.150:9092");
          
           //必须要组名
            prop.put(ProducerConfig.ACKS_CONFIG,"all");
            //或者
            prop.put(""acks,"all");
           
            prop.put(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG,0);
            //或者
            props.put("retries", 0);//重试次数
    
    	props.put("batch.size", 16384);
    	props.put("linger.ms", 1);
    	props.put("buffer.memory", 33554432);
    	
            //key的序列化器
            prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
           //或者
            prop.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
       
           //value的序列化器
            prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
            //或者
            prop.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
      
            KafkaProducer pro = new KafkaProducer(prop);
               for (int i = 0; i <10 ; i++) { //发送10条数据到topic中
                   pro.send(new ProducerRecord("mytest","test"+i,i+""));//mytest是topic名
                }
               pro.close();//关闭producer
         }
    
     }

检查是否已经写到topic

kafka-run-class.sh kafka.tools.GetOffsetShell --topic mytest \
    --time -1 --broker-list 192.168.5.150:9092 --partitions 0

或者启动消费者查看

kafka-console-consumer.sh --bootstrap-server 192.168.5.150:9092 --from-beginning --topic mytest

producer包含一个缓存空间来存放未发送的记录，一个后台i/O进程负责缓存中记录的请求和传送。使用后，如果不关闭producer，那么将存在资源泄漏。  
send()方法是异步的。调用时，它将记录添加到挂起记录发送的缓冲区，并立即返回。这样，生产者就可以为提高效率而对单个记录进行批处理。  
ack配置控制请求被认为是完整的条件。我们指定的“all”设置将导致阻塞记录的全部提交，这是最慢但最持久的设置。  
如果请求失败，生产者可以自动重试，但由于我们已经指定重试为0，它不会。启用重试还打开了复制的可能性(详细信息请参阅关于消息传递语义的文档)。  
默认情况下，即使缓冲区中还有未使用的空间，也可以立即发送缓冲区。但是，如果想减少可以设置linger的请求数。ms大于0次方。这将指示生成程序在发送请求之前等待该毫秒数，希望到达更多的记录以填充相同的批处理。这类似于Nagle在TCP中的算法。

如，在上面的代码片段中，由于我们将逗留时间设置为1毫秒，所以很可能所有100条记录都将在一个请求中发送。但是，如果我们不填充缓冲区，这个设置将为我们的请求增加1毫秒的延迟，等待更多的记录到达。注意，及时接近的记录通常会与linger一起批处理。ms=0，因此，在重载下，无论逗留配置如何，都将发生批处理;然而，将其设置为大于0的值会导致更少的、更有效的请求，而不是在最大负载下以少量延迟为代价。

**缓冲**：内存控制用于缓冲的内存的总量。如果记录的发送速度超过了传输到服务器的速度，那么这个缓冲区空间将被耗尽。当缓冲区空间耗尽时，将阻塞额外的发送调用。块的时间阈值由max.block确定。之后，它抛出一个TimeoutException。

由于发送调用是异步的，它将返回将分配给此记录的记录元数据的未来。在这个将来调用get()将会阻塞，直到相关请求完成，然后返回记录的元数据，或者抛出发送记录时发生的任何异常。

**参数解析：**

*  **bootstrap.servers**  
    producer中的的broker-list
 *  **acks**  
    确保生产者可靠性设置acks=0:不等待成功返回  
    acks=1:等Leader写成功返回  
    acks=all:等Leader和所有ISR中的Follower写成功返回,all也可以用-1代替
 *  **retries**  
    发送失败尝试重发次数
 *  **batch.size**  
    每个RecordBatch可以缓存的最大字节数
 *  **linger.ms**  
    一般情况下，记录会被立即发送出去，而不会等待缓存的填充。用户可以通过配置linger.ms来让producer等待一段时间再发送消息,可以设置每个RecordBatch的最长延迟发送时间
 *  **buffer.memory**  
    所有RecordBatch的总共最大字节数，表示缓存的大小。消息填满缓存后，后续的消息就会阻塞。阻塞超过max.block.ms设定的时间，就会抛出TimeoutException。
 *  **key.serializer 和 value.serializer**  
    如何将key和value组合成对象，可以自定义类。使用 StringSerializer默认组合成字符串。

Producer的内部实现，如下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center]  
1、客户端写程序，通过props中写的属性来连接broker集群，连接zookeeper集群，获取metadata信息，调用send()方法。

2、ProducerRecord对象携带者topic,partition，message等信息，在Serializer中被序列化

3、序列化过后的ProducerRecord对象进入Partitioner“中，按照Partitioning 策略决定这个消息将被分配到哪个Partition中。

4、确定partition的ProducerRecord进入一个缓冲区，通过减少IO来提升性能，在这个“车间”，消息被按照TopicPartition信息进行归类整理，相同Topic且相同parition的ProducerRecord被放在同一个RecordBatch中，等待被发送。什么时候发送？都在Producer的props中被指定了，有默认值，显然我们可以自己指定。  
一旦，当单个RecordBatch的linger.ms延迟到达或者batch.size达到上限，这个 RecordBatch会被立即发送。另外，如果所有RecordBatch作为一个整体，达到了buffer.memroy或者max.block.ms上限，所有的RecordBatch都会被发送。

5、ProducerRecord消息按照分配好的Partition发送到具体的broker中,broker接收保存消息，更新Metadata信息，同步给Zookeeper。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/42099da7e31d418493258e3493903a26.png