Consumer 的源码解析主要来看 KafkaConsumer，KafkaConsumer 是 Consumer 接口的实现类。KafkaConsumer 提供了一套封装良好的 API，开发人员可以基于这套 API 轻松实现从 Kafka 服务端拉取消息的功能，这样开发人员根本不用关心与 Kafka 服务端之间网络连接的管理、心跳检测、请求超时重试等底层操作，也不必关心订阅 Topic 的分区数量、分区副本的网络拓扑以及 Consumer Group 的 Rebalance 等 Kafka 具体细节，KafkaConsumer 中还提供了自动提交 offset 的功能，使的开发人员更加关注业务逻辑，提高了开发效率。
下面我们来看一个 KafkaConsumer 的示例程序：
/**
* @author: 微信公众号【老周聊架构】
*/
public class KafkaConsumerTest {
public static void main(String[] args) {
Properties props = new Properties();
// kafka地址,列表格式为host1:port1,host2:port2,...，无需添加所有的集群地址，kafka会根据提供的地址发现其他的地址(建议多提供几个，以防提供的服务器关闭) 必须设置
props.put("bootstrap.servers", "localhost:9092");
// key序列化方式 必须设置
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// value序列化方式 必须设置
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("group.id", "consumer_riemann_test");
KafkaConsumer consumer = new KafkaConsumer<>(props);
// 可消费多个topic,组成一个list
String topic = "riemann_kafka_test";
consumer.subscribe(Arrays.asList(topic));
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord record : records) {
System.out.printf("offset = %d, key = %s, value = %s
", record.offset(), record.key(), record.value());
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
从示例中可以看出 KafkaConsumer 的核心方法是 poll()，它负责从 Kafka 服务端拉取消息。核心方法的具体细节我想放在下一篇再细讲，关乎消费侧的客户端与 Kafka 服务端的通信模型。这一篇我们主要从宏观的角度来剖析下 Consumer 消费端的源码。
我们先来看下 Consumer 接口，该接口定义了 KafkaConsumer 对外的 API，其核心方法可以分为以下六类：
我们先来看下 KafkaConsumer 的重要属性以及 UML 结构图。
ConsumerNetworkClient 在 NetworkClient 之上进行了封装，提供了更高级的功能和更易用的 API。
我们先来看下 ConsumerNetworkClient 的重要属性以及 UML 结构图。
ConsumerNetworkClient 的核心方法是 poll() 方法，poll() 方法有很多重载方法，最终会调用 poll(Timer timer, PollCondition pollCondition, boolean disableWakeup) 方法，这三个参数含义是：timer 表示定时器限制此方法可以阻塞多长时间；pollCondition 表示可空阻塞条件；disableWakeup 表示如果 true 禁用触发唤醒。
我们来简单回顾下 ConsumerNetworkClient 的功能：
循环处理 unsent 中缓存的请求，对每个 Node 节点，循环遍历其 ClientRequest 链表，每次循环都调用 NetworkClient.ready() 方法检测消费者与此节点之间的连接，以及发送请求的条件。若符合条件，则调用 NetworkClient.send() 方法将请求放入 InFlightRequest 中等待响应，也放入 KafkaChannel 中的 send 字段等待发送，并将消息从列表中删除。代码如下：
long trySend(long now) {
long pollDelayMs = maxPollTimeoutMs;
// send any requests that can be sent now
// 遍历 unsent 集合
for (Node node : unsent.nodes()) {
Iterator iterator = unsent.requestIterator(node);
if (iterator.hasNext())
pollDelayMs = Math.min(pollDelayMs, client.pollDelayMs(node, now));
while (iterator.hasNext()) {
ClientRequest request = iterator.next();
// 调用 NetworkClient.ready(）检查是否可以发送请求
if (client.ready(node, now)) {
// 调用 NetworkClient.send(）方法，等待发送请求。
client.send(request, now);
// 从 unsent 集合中删除此请求
iterator.remove();
} else {
// try next node when current node is not ready
break;
}
}
}
return pollDelayMs;
}
如果没有请求在进行中，则阻塞时间不要超过重试退避时间。
调用 checkDisconnects() 方法检测连接状态。检测消费者与每个 Node 之间的连接状态，当检测到连接断开的 Node 时，会将其在 unsent 集合中对应的全部 ClientRequest 对象清除掉，之后调用这些ClientRequest 的回调函数。
private void checkDisconnects(long now) {
// any disconnects affecting requests that have already been transmitted will be handled
// by NetworkClient, so we just need to check whether connections for any of the unsent
// requests have been disconnected; if they have, then we complete the corresponding future
// and set the disconnect flag in the ClientResponse
for (Node node : unsent.nodes()) {
// 检测消费者与每个 Node 之间的连接状态
if (client.connectionFailed(node)) {
// Remove entry before invoking request callback to avoid callbacks handling
// coordinator failures traversing the unsent list again.
// 在调用请求回调之前删除条目以避免回调处理再次遍历未发送列表的协调器故障。
Collection requests = unsent.remove(node);
for (ClientRequest request : requests) {
RequestFutureCompletionHandler handler = (RequestFutureCompletionHandler) request.callback();
AuthenticationException authenticationException = client.authenticationException(node);
// 调用 ClientRequest 的回调函数
handler.onComplete(new ClientResponse(request.makeHeader(request.requestBuilder().latestAllowedVersion()),
request.callback(), request.destination(), request.createdTimeMs(), now, true,
null, authenticationException, null));
}
}
}
}
检查 wakeupDisabled 和 wakeup，查看是否有其它线程中断。如果有中断请求，则抛出 WakeupException 异常，中断当前 ConsumerNetworkClient.poll() 方法。
public void maybeTriggerWakeup() {
// 通过 wakeupDisabled 检测是否在执行不可中断的方法，通过 wakeup 检测是否有中断请求。
if (!wakeupDisabled.get() && wakeup.get()) {
log.debug("Raising WakeupException in response to user wakeup");
// 重置中断标志
wakeup.set(false);
throw new WakeupException();
}
}
再次调用 trySend() 方法。在步骤 2.1.3 中调用了 NetworkClient.poll() 方法，在其中可能已经将 KafkaChannel.send 字段上的请求发送出去了，也可能已经新建了与某些 Node 的网络连接，所以这里再次尝试调用 trySend() 方法。
处理 unsent 中超时请求。它会循环遍历整个 unsent 集合，检测每个 ClientRequest 是否超时，将过期请求加入到 expiredRequests 集合，并将其从 unsent 集合中删除。调用超时 ClientRequest 的回调函数 onFailure()。
private void failExpiredRequests(long now) {
// clear all expired unsent requests and fail their corresponding futures
// 清除所有过期的未发送请求并使其相应的 futures 失败
Collection expiredRequests = unsent.removeExpiredRequests(now);
for (ClientRequest request : expiredRequests) {
RequestFutureCompletionHandler handler = (RequestFutureCompletionHandler) request.callback();
// 调用回调函数
handler.onFailure(new TimeoutException("Failed to send request after " + request.requestTimeoutMs() + " ms."));
}
}
private Collection removeExpiredRequests(long now) {
List expiredRequests = new ArrayList<>();
for (ConcurrentLinkedQueue requests : unsent.values()) {
Iterator requestIterator = requests.iterator();
while (requestIterator.hasNext()) {
ClientRequest request = requestIterator.next();
// 检查是否超时
long elapsedMs = Math.max(0, now - request.createdTimeMs());
if (elapsedMs > request.requestTimeoutMs()) {
// 将过期请求加入到 expiredRequests 集合
expiredRequests.add(request);
requestIterator.remove();
} else
break;
}
}
return expiredRequests;
}
说 RequestFutureCompletionHandler 之前，我们先来看下 ConsumerNetworkClient.send() 方法。里面的逻辑会将待发送的请求封装成 ClientRequest，然后保存到 unsent 集合中等待发送，代码如下：
public RequestFuture send(Node node,
AbstractRequest.Builder<?> requestBuilder,
int requestTimeoutMs) {
long now = time.milliseconds();
RequestFutureCompletionHandler completionHandler = new RequestFutureCompletionHandler();
ClientRequest clientRequest = client.newClientRequest(node.idString(), requestBuilder, now, true,
requestTimeoutMs, completionHandler);
// 创建 clientRequest 对象，并保存到 unsent 集合中。
unsent.put(node, clientRequest);
// wakeup the client in case it is blocking in poll so that we can send the queued request
// 唤醒客户端以防它在轮询中阻塞，以便我们可以发送排队的请求。
client.wakeup();
return completionHandler.future;
}
我们重点来关注一下 ConsumerNetworkClient 中使用的回调对象——RequestFutureCompletionHandler。其继承关系如下：
从 RequestFutureCompletionHandler 继承关系图我们可以知道，它不仅实现了 RequestCompletionHandler 接口，还组合了 RequestFuture 类，RequestFuture 是一个泛型类，其核心字段与方法如下：
我们之所以要分析源码，是因为源码中有很多设计模式可以借鉴，应用到你自己的工作中。RequestFuture 中有两处典型的设计模式的使用，我们来看一下：
/**
* 适配器
* Adapt from a request future of one type to another.
*
* @param Type to adapt from
* @param Type to adapt to
*/
public abstract class RequestFutureAdapter {
public abstract void onSuccess(F value, RequestFuture future);
public void onFailure(RuntimeException e, RequestFuture future) {
future.raise(e);
}
}
/**
* RequestFuture 适配成 RequestFuture
* Convert from a request future of one type to another type
* @param adapter The adapter which does the conversion
* @param The type of the future adapted to
* @return The new future
*/
public RequestFuture compose(final RequestFutureAdapter adapter) {
// 适配之后的结果
final RequestFuture adapted = new RequestFuture<>();
// 在当前 RequestFuture 上添加监听器
addListener(new RequestFutureListener() {
@Override
public void onSuccess(T value) {
adapter.onSuccess(value, adapted);
}
@Override
public void onFailure(RuntimeException e) {
adapter.onFailure(e, adapted);
}
});
return adapted;
}
使用 compose() 方法进行适配后，回调时的调用过程，也可以认为是请求完成的事件传播流程。当调用 RequestFuture 对象的对应方法。
chain() 方法与 compose() 方法类似，也是通过 RequestFutureListener 在多个 RequestFuture 之间传递事件。代码如下：
public void chain(final RequestFuture future) {
// 添加监听器
addListener(new RequestFutureListener() {
@Override
public void onSuccess(T value) {
// 通过监听器将 value 传递给下一个 RequestFuture 对象
future.complete(value);
}
@Override
public void onFailure(RuntimeException e) {
// 通过监听器将异常传递给下一个 RequestFuture 对象
future.raise(e);
}
});
}
好了，ConsumerNetworkClient 的源码分析告一段落了，希望文章对你有帮助，我们下期再见。
页面更新：2024-03-08
本站资料均由网友自行发布提供，仅用于学习交流。如有版权问题，请与我联系，QQ：4156828
© CopyRight 2020-2024 All Rights Reserved. Powered By 71396.com 闽ICP备11008920号-4
闽公网安备35020302034903号
