Kafka-2.6.0之Producer

野性酷女 2022-12-05 04:57 156阅读 0赞

# 1. 简单请求，异步 #

Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("acks", "all");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
    Producer<String, String> producer = new KafkaProducer<>(props);
    for (int i = 0; i < 100; i++)
        producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));
    producer.close();

producer作为一个缓冲区，缓存所有未发送的消息记录，同时也持有一个后台I/O线程负责将记录发送给服务器。

<table> 
 <thead> 
  <tr> 
   <th>方法或者参数</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>send()</code></td> 
   <td align="left">异步方法，该方法会将消息记录发送到缓冲区，并且立即返回。producer会对这些消息进行批量处理。</td> 
  </tr> 
  <tr> 
   <td><code>acks</code></td> 
   <td align="left">指定了producer对消息的确认模式</td> 
  </tr> 
  <tr> 
   <td><code>retries</code></td> 
   <td align="left">如果消息发送失败，producer 会自动进行消息重发，<code>retries</code>设置重发次数，低版本的默认是0，0.10.2.1改为10次，需要检查配置</td> 
  </tr> 
  <tr> 
   <td><code>batch.size</code></td> 
   <td align="left">生产者为每个分区维护未发送的消息缓冲区，<code>batch.size</code>设置缓冲区大小。更大的缓冲区需要更大的内存，内存参数设置<code>buffer.memory</code>。</td> 
  </tr> 
  <tr> 
   <td><code>buffer.memory</code></td> 
   <td align="left">producer设置可用于缓冲的内存总量，当记录的速度快于发送到服务器的速度，缓冲区的内存将会被耗尽，继续记录将会阻塞，阻塞的时间由该参数<code>max.block.ms</code>设置，超过该时间将抛出TimeoutException。</td> 
  </tr> 
  <tr> 
   <td><code>linger.ms</code></td> 
   <td align="left">发送等待时间。默认情况下，及时缓冲区没有用完也会立即发送请求。为了充分利用缓冲区，可以将<code>linger.ms</code>设置大一些，也可以减少请求次数。但是如果在设置的时间内缓冲区还是有剩余，那就增加了请求等待时间。需要注意的是，即便是将<code>linger.ms</code>设置为0，及时到达的消息也会被批量发送。因此，在负载不大的情况加可以根据实际情况增加该值，在高负载的情况下应该根据时间情况调大该值。</td> 
  </tr> 
  <tr> 
   <td><code>key.serializer</code> <code>value.serializer</code></td> 
   <td align="left">将发送的消息转换为字节的方式，内置的转换器有<code>ByteArraySerializer</code>, <code>StringSerializer</code></td> 
  </tr> 
 </tbody> 
</table>

`acks`参数说明

> 1.  acks=0：producer不会等待服务器的任何确认，消息会马上加入到缓冲区，并且作为已发送对待。服务器的消息接收无法得到保证，由于客户端无法获取到消息是否发送成功，所以消息重试设置`retries`会失效。此种模式存在消息丢失。
> 2.  acks=1：只等待leader消息确认，不管followers。如果leader消息复制成功，有followers消息复制失败，消息会丢失。
> 3.  acks=-1或者all：leader会等待`ISR`（in-sync replicas）完全确认，只要有任意的`ISR`存活消息便不会丢失，这能保证消息的最强可用性。

# 2. 事务发送 #

从kafka0.11开始，KafkaProducer支持两种模式：`幂等（idempotence）生产者`和`事务（transactional ）生产者`。幂等生产者强调的是一次准确的发送。事务生产者允许应用程序将消息以原子性的方式发送到多个分区（和主题）。  
为了使用幂等性`enable.idempotence`必须设置为true，`retries`默认为`Integer.MAX_VALUE`，`acks`默认为`all`。为了使用幂等性，要避免使用程序级别的重试，否则会造成消息重复消费。如果一个发送在无限重试的情况下返回错误，建议停掉prooducer并检查最后一条消息，避免重复消费。  
要使用事务，producer必须设置参数`transactional.id`，  
使用事务发送无需定义callback函数，也无需检查返回结果。  
**示例**

Properties props = new Properties();
     props.put("bootstrap.servers", "localhost:9092");
     props.put("transactional.id", "my-transactional-id");
     Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
    
     producer.initTransactions();
    
     try { 
         producer.beginTransaction();
         for (int i = 0; i < 100; i++)
             producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
         producer.commitTransaction();
     } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) { 
         // We can't recover from these exceptions, so our only option is to close the producer and exit.
         producer.close();
     } catch (KafkaException e) { 
         // For all other exceptions, just abort the transaction and try again.
         producer.abortTransaction();
     }
     producer.close();

事务使用异常来传递错误状态，不需要指定生产者回调。方法`producer.abortTransaction()` 接收KafkaException来确保任何成功写入的消息进行终止。

# 3. 同步调用 #

使用`get()`方法进行阻塞

byte[] key = "key".getBytes();
     byte[] value = "value".getBytes();
     ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("my-topic", key, value)
     producer.send(record).get();

如果遇到`OutOfOrderSequenceException`，应该关闭生产者，重新创建一个生产者实例。

# 4. callback方法使用示例 #

ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("the-topic", key, value);
     producer.send(myRecord,
                   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());
                              //operation e.g. save to db...
                           }
                       }
                   });