Kafka 0.11.0.0 实现 producer的Exactly-once 语义(官方DEMO)
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.11.0.2</version> </dependency>
public class KafkaProducer<K,V> extends java.lang.Object implements Producer<K,V>A Kafka client that publishes records to the Kafka cluster.
The producer is thread safe and sharing a single producer instance across threads will generally be faster than having multiple instances.
Here is a simple example of using the producer to send records with strings containing sequential numbers as the key/value pairs.
Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("acks", "all"); props.put("retries", 0); props.put("batch.size", 16384); props.put("linger.ms", 1); 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<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();
The producer consists of a pool of buffer space that holds records that haven't yet been transmitted to the server as well as a background I/O thread that is responsible for turning these records into requests and transmitting them to the cluster. Failure to close the producer after use will leak these resources.
The send()
method is asynchronous. When called it adds the record to a buffer of pending record sends and immediately returns. This allows the producer to batch together individual records for efficiency.
The
acksconfig controls the criteria under which requests are considered complete. The "all" setting we have specified will result in blocking on the full commit of the record, the slowest but most durable setting.
If the request fails, the producer can automatically retry, though since we have specified
retriesas 0 it won't. Enabling retries also opens up the possibility of duplicates (see the documentation on message delivery semantics for details).
The producer maintains buffers of unsent records for each partition. These buffers are of a size specified by the
batch.sizeconfig. Making this larger can result in more batching, but requires more memory (since we will generally have one of these buffers for each active partition).
By default a buffer is available to send immediately even if there is additional unused space in the buffer. However if you want to reduce the number of requests you can set
linger.msto something greater than 0. This will instruct the producer to wait up to that number of milliseconds before sending a request in hope that more records will arrive to fill up the same batch. This is analogous to Nagle's algorithm in TCP. For example, in the code snippet above, likely all 100 records would be sent in a single request since we set our linger time to 1 millisecond. However this setting would add 1 millisecond of latency to our request waiting for more records to arrive if we didn't fill up the buffer. Note that records that arrive close together in time will generally batch together even with
linger.ms=0so under heavy load batching will occur regardless of the linger configuration; however setting this to something larger than 0 can lead to fewer, more efficient requests when not under maximal load at the cost of a small amount of latency.
The
buffer.memorycontrols the total amount of memory available to the producer for buffering. If records are sent faster than they can be transmitted to the server then this buffer space will be exhausted. When the buffer space is exhausted additional send calls will block. The threshold for time to block is determined by
max.block.msafter which it throws a TimeoutException.
The
key.serializerand
value.serializerinstruct how to turn the key and value objects the user provides with their
ProducerRecordinto bytes. You can use the included
ByteArraySerializeror
StringSerializerfor simple string or byte types.
From Kafka 0.11, the KafkaProducer supports two additional modes: the idempotent producer and the transactional producer. The idempotent producer strengthens Kafka's delivery semantics from at least once to exactly once delivery. In particular producer retries will no longer introduce duplicates. The transactional producer allows an application to send messages to multiple partitions (and topics!) atomically.
To enable idempotence, the
enable.idempotenceconfiguration must be set to true. If set, the
retriesconfig will be defaulted to
Integer.MAX_VALUE, the
max.in.flight.requests.per.connectionconfig will be defaulted to
1, and
acksconfig will be defaulted to
all. There are no API changes for the idempotent producer, so existing applications will not need to be modified to take advantage of this feature.
To take advantage of the idempotent producer, it is imperative to avoid application level re-sends since these cannot be de-duplicated. As such, if an application enables idempotence, it is recommended to leave the
retriesconfig unset, as it will be defaulted to
Integer.MAX_VALUE. Additionally, if a
send(ProducerRecord)returns an error even with infinite retries (for instance if the message expires in the buffer before being sent), then it is recommended to shut down the producer and check the contents of the last produced message to ensure that it is not duplicated. Finally, the producer can only guarantee idempotence for messages sent within a single session.
To use the transactional producer and the attendant APIs, you must set the
transactional.idconfiguration property. If the
transactional.idis set, idempotence is automatically enabled along with the producer configs which idempotence depends on. Further, topics which are included in transactions should be configured for durability. In particular, the
replication.factorshould be at least
3, and the
min.insync.replicasfor these topics should be set to 2. Finally, in order for transactional guarantees to be realized from end-to-end, the consumers must be configured to read only committed messages as well.
The purpose of the
transactional.idis to enable transaction recovery across multiple sessions of a single producer instance. It would typically be derived from the shard identifier in a partitioned, stateful, application. As such, it should be unique to each producer instance running within a partitioned application.
All the new transactional APIs are blocking and will throw exceptions on failure. The example below illustrates how the new APIs are meant to be used. It is similar to the example above, except that all 100 messages are part of a single transaction.
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();
As is hinted at in the example, there can be only one open transaction per producer. All messages sent between the beginTransaction()
and commitTransaction()
calls will be part of a single transaction. When the
transactional.idis specified, all messages sent by the producer must be part of a transaction.
The transactional producer uses exceptions to communicate error states. In particular, it is not required to specify callbacks for
producer.send()or to call
.get()on the returned Future: a
KafkaExceptionwould be thrown if any of the
producer.send()or transactional calls hit an irrecoverable error during a transaction. See the
send(ProducerRecord)documentation for more details about detecting errors from a transactional send.
By calling
producer.abortTransaction()upon receiving a
KafkaExceptionwe can ensure that any successful writes are marked as aborted, hence keeping the transactional guarantees.
This client can communicate with brokers that are version 0.10.0 or newer. Older or newer brokers may not support certain client features. For instance, the transactional APIs need broker versions 0.11.0 or later. You will receive an
UnsupportedVersionExceptionwhen invoking an API that is not available in the running broker version.
- Kafka 0.11.0.0 实现 producer的Exactly-once 语义(英文)
- Kafka 0.11.0.0 实现 producer的Exactly-once 语义(中文)
- Kafka 0.11.0.0 是如何实现 Exactly-once 语义的
- kafka exactly once 的实现原理解析
- Spark Streaming 中如何实现 Exactly-Once 语义
- SparkStreaming实现Exactly-Once语义
- SparkStreaming实现Exactly-Once语义
- BaiduMap---百度地图官方Demo之调用百度地图(介绍如何调启百度地图实现自身业务功能)
- 高德地图的Js API 简单Demo,可以快速实现路点描线(摘自官方接口及自行修改测试)
- 基于Confluent.Kafka实现的KafkaConsumer消费者类和KafkaProducer消息生产者类型
- 【译】Flink + Kafka 0.11端到端精确一次处理语义的实现
- 【安卓SDK学习之anyChatSDK】 1_1 实现用户登录和房间进出(根据官方HelloAnyChat的demo整理的简要的开发流程)
- Kafka Java Producer代码实现
- Kafka java 实现消息队列demo
- kafka producer java API的实现
- kafka消息demo的java实现
- kafka_2.11-0.10.0.1生产者producer的Java实现
- kafka 0.10.0 producer java代码实现
- Kafka官方文档翻译——实现
- Trident exactly once实现原理