kafka原生producer API

转自https://blog.csdn.net/tianlan996/article/details/80495208

1. 类

public class KafkaProducer<K,V>

extends java.lang.Object

implements Producer<K,V>

2. producer是线程安全的（这点不同于consumer），多线程共享producer可以提高效率。

3. 使用示例：

 Properties props = new Properties();

 props.put("bootstrap.servers", "localhost:9092");

 props.put("acks", "all");

 props.put("retries", );---重试次数

 props.put("batch.size", );

 props.put("linger.ms", );

 props.put("buffer.memory", );

 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 = ; i < ; i++)

     producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));

 producer.close();

producer包含一个缓存空间来存放未发送的记录，一个后台i/O进程负责缓存中记录的请求和传送。使用后，如果不关闭producer，那么将存在资源泄漏。

send()方法是异步的。

batch.size：producer维护了每个分区未发送记录的缓存，该缓存的大小由batch.size设定。

linger.ms：一般情况下，记录会被立即发送出去，而不会等待缓存的填充。用户可以通过配置linger.ms来让producer等待一段时间再发送消息。

buffer.memory：缓存的大小。消息填满缓存后，后续的消息就会阻塞。阻塞超过max.block.ms设定的时间，就会抛出TimeoutException。

key.serializer and value.serializer：如何将key和value组合成对象，可以自定义类。使用 StringSerializer默认组合成字符串。

3. idempotent producer

enable.idempotence true

retries Integer.MAX_VALUE（不设置，默认即为此值）

acks all

通过上述配置开启idempotent，可以保证exactly语意。prouducer java api 不变。只有在同一个session中才能保证produder的idempotent。

4. transactional producer

transactional.id <one value>

replication.factor

min.insync.replicas

通过上述配置开启transactional。transactional.id被设置后，idempotent也会自动开启。

consumer端需要配置为只消费committed的消息。

在分区应用中，每个producer的transactional.id须唯一。

 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 = ; i < ; i++) //100条消息组成一个单独的事务

         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();

巴特西

kafka原生producer API

最新文章

热门文章