首页 小编推荐正文

端午节是几月几日,技能架构师总结:Kafka多线程顾客,甲沟炎最佳治疗方法

  1. Kafka Java Consumer从Kaf韦昭尤风水视频完好版ka 0.10.1.0开端,KafkaConsumer变成了 双线程 规划,即 用户主线程心跳线程
  • 用户主线程:发动Consumer运用程序main办法的那个线程
  • 心跳线程:只担任定时给对应的Broker机器发送心跳恳求,以标识顾客运用的存活性
  1. 引进心跳线程的另一个意图
  • 将心跳频率和主线程调用KafkaConsumer.poll办法的频率分隔,解耦 实在的音讯处理逻辑消费组成员存活性办理
  1. 虽然有了心跳线程,但实践的音讯获取逻辑仍然是在用户主线程中完结
  • 因此在 消费音讯 的这个层面,仍然能够安全地以为Kafk二次元凶恶图片aConsumer是斗争在白垩纪 单线程的规划
  1. 老版别Consumer是 多线程 的架构
  • 每个Consumer实例在内部为 一切订阅的主题分区 创立对应的 音讯获取线程 ,即Fetcher线程
  1. 老版别Consumer一起也是 堵塞式 的,Consumer实例发动后,内部会创立许多堵塞式的音讯获取迭代器
  • 但在许多场景下,Consumer端有 非堵塞 需求,如在 流处理 运用中履行 过滤分组 等操作就不能是堵塞式的
  • 根据这个反剪原因,社区为新版别Consumer规划了 单线程+轮询 的机制,该机制能较好地完成非堵塞的音讯获取
  1. 单线程的规划 现场铁证第一部简化 了Consumer端的规划
  • Consumer获取到音讯后,处理音讯的逻辑是否选用多线程,完全由运用者决议
  1. 不管运用哪一种编程言语,单线程的规划都比较简略完成
  • 并不是一切的编程言语都能韩国瑜伽妹很好地支撑多线程,而单线程规划的Consumer更简略 移植 到其他言语上

多线程计划

  1. KafkaConsumer是 线程不安全
  2. 不能多线程同享一个KafkaConsumer实例,否则会抛出 ConcurrentModificationException
  3. 但KafkaConsumer.wakeup()是线程安全的

计划1

  1. 顾客程序发动多个线程, 每个线程保护专属的KafkaConsumer实例 ,担任完好的音讯获取、音讯处理流程
  2. 长处
  • 完成简略 ,比较契合现在运用Consumer API的习气
  • 多个线程之间 没有任何交互 ,省去了许多确保线程安全方面的开支
  • Ka吞天猿fka主题中的 每个分区 都能确保 只被一个线程处理 ,简略完成 分区内的音讯消费次序
  1. 缺陷
  • 每个线程都保护自己的KafkaConsumer实例,必然会占用 更多的系统资源 ,如内存、TCP衔接等
  • 能运用的线程数 受限 于Cons杜礼明umer 订阅主题剑巫纪的总分区数
  • 每个线程 完好 地履行音讯获取和音讯处理逻辑
  • 一旦音讯处理逻辑很重,音讯处理速度很慢,很简略端午节是几月几日,技术架构师总结:Kafka多线程顾客,甲沟炎最佳医治办法呈现 不用要的Rebalance ,引发整个顾客组的 消费阻滞
public class KafkaConsumerRunner implements Runnable {
private final AtomicBoolean clo插妈sed = new AtomicBooleshijijiayan(false);
private final KafkaConsumer consumer = new KafkaConsumer(new Properties());
@Override
public void run() {
try {
consumer.subscribe(Collections.singletonList("topic"));
wh福利相片ile (!closed.get()) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(10000));
// 履行音讯处理逻辑
}
} catch (WakeupException e) {
// Ignore exception i刑家军f closing
if (!closed.get()) {
throw e;端午节是几月几日,技术架构师总结:Kafka多线程顾客,甲沟炎最佳医治办法
}
} fi爱草nally {
consumer.close();
}
}
// Shutdown hook which can be called from a sep端午节是几月几日,技术架构师总结:Kafka多线程顾客,甲沟炎最佳医治办法arate thread
public void shutdown() {
closed.set(true);
consumer.wakeup();星际传说之人鱼清轻
}
}

计划2

  1. 顾客程序运用单个或多个线程获取音讯,一起创立多个消费线程执littlstar行音讯处理逻辑
  • 获取音讯端午节是几月几日,技术架构师总结:Kafka多线程顾客,甲沟炎最佳医治办法的线程能够是一个,也能够是多个, 每个线程保护专属的KafkaConsumer实例
  • 处理音讯则由 特定的线程池 来做,然后完成 音讯获取音讯处理 的 杨达与黄俊英一切相声真正解耦
  1. 长处
  • 把使命切分红 音讯获取音讯处理 两部分,别离由不同的线程来处理
  • 相对于计划1,计划2最大的优势是它的 高伸缩性
  • 能够独登时调理音讯获取的线程数,以及音讯处理的线程数,不用考虑两者之间是否相互影响
  1. 缺陷
  • 完成难度大 ,由于要别离办理两组线程
  • 音讯获取和音讯处了解耦, 无法确保分区内的消费次序
  • 两组线程,使得 整个音讯消费链路被拉长 ,终究导致 正确位移提交会变端午节是几月几日,技术架构师总结:Kafka多线程顾客,甲沟炎最佳医治办法得反常困难 ,可能会呈现音讯的 重复消费
private707特战营 final KafkaConsumer consumer;
privat雍正后宫e ExecutorService executors;
...
private int workerNum = 10;
executors = new ThreadPoolExecutor(
workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(1000),
new ThreadPoolExecutor.CallerR端午节是几月几日,技术架构师总结:Kafka多线程顾客,甲沟炎最佳医治办法unsPolicy());
...
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofSeconds(1));
for (final 端午节是几月几日,技术架构师总结:Kafka多线程顾客,甲沟炎最佳医治办法ConsumerRecord record : records) {
// 由专门的线程池担任处理详细的音讯
executors.submit(new Worker(record));
}
}
...
版权声明

本文仅代表作者观点,不代表本站立场。
本文系作者授权发表,未经许可,不得转载。