目录
Kafka集群中有一个broker会被选举为Controller，负责管理集群broker的上下线、所有topic的分区副本分配和leader的选举等工作。Controller的工作管理是依赖于zookeeper的。
Partition的Leader的选举过程
拦截器原理
Producer拦截器interceptor是在Kafka0.10版本引入的，主要用于Clients端的定制化控制逻辑。对于Producer而言，interceptor使得用户在消息发送之前以及Producer回调逻辑之前有机会对消息做一些定制化需求，比如修改消息的展示样式等，同时Producer允许用户指定多个interceptor按序作用于同一条消息从而形成一个拦截链interceptor chain，Interceptor实现的接口为ProducerInterceptor，主要有四个方法：
Inteceptor可能被运行到多个线程中，在具体使用时需要自行确保线程安全，另外倘若指定了多个interceptor，则producer将按照指定顺序调用它们，并紧紧是捕获每个interceptor可能抛出的异常记录到错误日志中而非向上传递。
自定义加入时间戳拦截器
/**
* @author caoduanxi
* @Date 2021/1/13 14:15
* @Motto Keep thinking, keep coding!
*/
public class TimeInterceptor implements ProducerInterceptor<String, String> {
@Override
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
return new ProducerRecord(record.topic(), record.partition(), record.timestamp(), record.key(),
"TimeInterceptor:" + System.currentTimeMillis() + "," + record.value());
}
// 其余方法省略
}
自定义消息发送统计拦截器
/**
* @author caoduanxi
* @Date 2021/1/13 14:18
* @Motto Keep thinking, keep coding!
*/
public class CounterInterceptor implements ProducerInterceptor<String, String> {
private int errorCounter = 0;
private int successCounter = 0;
@Override
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
return record;
}
@Override
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
if (exception == null) {
successCounter++;
} else {
errorCounter++;
}
}
@Override
public void close() {
// 输出结果，结束输出
System.out.println("Sent successful:" + successCounter);
System.out.println("Sent failed:" + errorCounter);
}
}
在CustomProducer中加入拦截器
// 加入拦截器
List<Object> interceptors = new ArrayList<>();
interceptors.add(TimeInterceptor.class);
interceptors.add(CounterInterceptor.class);
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
注意：拦截器的close()方法是收尾的，一定要调用Producer.close()方法，否则拦截器的close()方法不会被调用。
Eagle
Eagle是开源的额可视化和管理软件，允许查询、可视化、提醒和探索存储在任何地方的指标，简而言之，Eagle为您提供了将Kafka集群数据转换为漂亮的图形和可视化的工具。
实质: 一个运行在tomcat上的web应用。
kafka目前默认可支持1h以内的延迟消费。
使用方式：consumer启动参数增加 --delay-time-seconds n 设置消费延迟时间，单位秒，默认不延迟消费。仅能拉取到消费延迟时间之前的消息。
注意：此参数默认限制最大值为3600s，超过限制可能导致consumer启动失败。如有调整最大延迟时间的需求，请联系李锦涛（KIM：lijintao）
注意：消息拉取可能有分钟级别的误差。
注意：由于目前每4kb数据构建一次时间索引，如果最后一批数据的size不够4kb，可能导致这些数据不能被延迟消费到。
相关推荐文章：