原文地址：https://kafka.apache.org/0101/documentation.html#apidesign
Producer APIs
Producer API封装了底层两个Producer：
- kafka.producer.SyncProducer
- kafka.producer.async.AsyncProducer
class Producer {
/* Sends the data, partitioned by key to the topic using either the */
/* synchronous or the asynchronous producer */
public void send(kafka.javaapi.producer.ProducerData<K,V> producerData);
/* Sends a list of data, partitioned by key to the topic using either */
/* the synchronous or the asynchronous producer */
public void send(java.util.List<kafka.javaapi.producer.ProducerData<K,V>> producerData);
/* Closes the producer and cleans up */
public void close();
}
这么做的目的是通过一个简答的API暴露把所有Producer的功能暴露给Client。Kafka的Producer可以：
- 排队/缓存多个发送请求并且异步的批量分发出去：
kafka.producer.Producer提供批量化多个发送请求(producer.type=async)，之后进行序列化并发送的分区的能力。批大小可以通过一些配置参数进行设置。将事件加入到queue中，他们会被缓冲在queue中，直到满足queue.time或batch.size达到配置的值。后台线程（kafka.producer.async.ProducerSendThread）从queue中获取数据并使用kafka.producer.EventHandler对数据进行序列化并发送到合适的分区。可以通过event.handler参数以插件的形式添加自定义的event handler程序。在producer queue pipeline处理的各个阶段可以注入回调，用于自定义的日志/跟踪代码或者监控逻辑。这可以通过实现kafka.producer.async.CallbackHandler接口并设置callback.handler参数来实现。 - 使用用户指定的Encoder来序列化数据：
interface Encoder<T> {
public Message toMessage(T data);
}
默认使用kafka.serializer.DefaultEncoder。
- 通过用户可选的Partitioner来实现乱负载均衡：
Partition的路由由kafka.producer.Partitioner决定。
interface Partitioner<T> {
int partition(T key, int numPartitions);
}
分区选择API使用key和分区总数来选择最终的partition（返回选择的partition id）。id用于从排序的partition列表中选择最终的一个分区去发送数据。默认的分区策略是hash(key)%numPartitions。如果key是null，会随机选择一个分区。可以通过partitioner.class参数来配置特定的分区选择策略。
Consumer APIs
我们有两个级别的Consumer API。低级别的“简单的”API和单个Broker之间保持链接并且和发送到服务端的网络请求有紧密的对应关系。这个API是无状态的，每个请求都包含offset信息，允许用户维护这个元数据。
高级别的API在Consumer端隐藏了Broker的细节，并且允许从集群消费数据而不关心底层的拓扑结构。同样维持了“哪些数据已经被消费过”的状态。高级别的API还提供了通过表达式订阅的Topic的功能（例如通过白名单或者黑名单的方式订阅）。
Low-level API
class SimpleConsumer {
/* Send fetch request to a broker and get back a set of messages. */
public ByteBufferMessageSet fetch(FetchRequest request);
/* Send a list of fetch requests to a broker and get back a response set. */
public MultiFetchResponse multifetch(List<FetchRequest> fetches);
/**
* Get a list of valid offsets (up to maxSize) before the given time.
* The result is a list of offsets, in descending order.
* @param time: time in millisecs,
* if set to OffsetRequest$.MODULE$.LATEST_TIME(), get from the latest offset available.
* if set to OffsetRequest$.MODULE$.EARLIEST_TIME(), get from the earliest offset available.
*/
public long[] getOffsetsBefore(String topic, int partition, long time, int maxNumOffsets);
}
低级别的API用于实现高级别的API，也被直接使用在一些在状态上有特殊需求的“离线”Consumer。
High-level API
/* create a connection to the cluster */
ConsumerConnector connector = Consumer.create(consumerConfig);
interface ConsumerConnector {
/**
* This method is used to get a list of KafkaStreams, which are iterators over
* MessageAndMetadata objects from which you can obtain messages and their
* associated metadata (currently only topic).
* Input: a map of <topic, #streams>
* Output: a map of <topic, list of message streams>
*/
public Map<String,List<KafkaStream>> createMessageStreams(Map<String,Int> topicCountMap);
/**
* You can also obtain a list of KafkaStreams, that iterate over messages
* from topics that match a TopicFilter. (A TopicFilter encapsulates a
* whitelist or a blacklist which is a standard Java regex.)
*/
public List<KafkaStream> createMessageStreamsByFilter(
TopicFilter topicFilter, int numStreams);
/* Commit the offsets of all messages consumed so far. */
public commitOffsets()
/* Shut down the connector */
public shutdown()
}
这个API围绕迭代器，通过KafkaStream类实现。一个KafkaStream表示了一个或多个分区（可以分布在不同的Broker上）组成的消息流。每个Stream被单个线程处理，客户端可以在创建流时提供需要的个数。这样，一个流背后可以是多个分区，但是一个分区只会属于一个流。
createMessageStreams调用会把Consumer注册到Topic，促使Consumer/Broker的重新分配。API鼓励在单次调用中创建多个Stream以减少充分配的次数。createMessageStreamsByFilter方法的调用（另外的）用于注册watcher去发现匹配过滤规则的topic。createMessageStreamsByFilter返回的迭代器可以迭代来此多个Topic的消息（如果多个Topic都符合过滤规则）。
网友评论