ApacheKafka系列(四)多线程Consumer方案

http://www.cnblogs.com/qizhelongdeyang/p/7355309.html

• Apache Kafka系列(一) 起步
• Apache Kafka系列(二) 命令行工具（CLI）
• Apache Kafka系列(三) Java API使用
• Apache Kafka系列(四) 多线程Consumer方案
• Apache Kafka系列(五) Kafka Connect及FileConnector示例

本文的图片是通过PPT截图出的，读者如果修改意见请联系我

一、Consumer为何需要实现多线程

　　假设我们正在开发一个消息通知模块，该模块允许用户订阅其他用户发送的通知/消息。该消息通知模块采用Apache Kafka，那么整个架构应该是消息的发布者通过Producer调用API写入消息到Kafka Cluster中，然后消息的订阅者通过Consumer读取消息，刚开始的时候系统架构图如下：

          但是，随着用户数量的增多，通知的数据也会对应的增长。总会达到一个阈值，在这个点上，Producer产生的数量大于Consumer能够消费的数量。那么Broker中未消费的消息就会逐渐增多。即使Kafka使用了优秀的消息持久化机制来保存未被消费的消息，但是Kafka的消息保留机制限制（时间，分区大小，消息Key）也会使得始终未被消费的Message被永久性的删除。另一方面从业务上讲，一个消息通知系统的高延迟几乎算作是废物了。所以多线程的Consumer模型是非常有必要的。

二、多线程的Kafka Consumer 模型类别

　　基于Consumer的多线程模型有两种类型：

• 模型一：多个Consumer且每一个Consumer有自己的线程，对应的架构图如下：

• 模型二：一个Consumer且有多个Worker线程

   　　两种实现方式的优点/缺点比较如下：

三、代码实现

3.1 前提

• • Kafka Broker 0.11.0
  • JDK1.8
  • IDEA
  • Maven3
  • Kafka环境搭建及Topic创建修改等请参照本系列的前几篇文章。

 3.2 源码结构

       其中，consumergroup包下面对应的是模型一的代码，consumerthread包下是模型二的代码。ProducerThread是生产者代码。

 3.3 pom.xml

  4.0.0
  com.randy
  kafka_multithread_consumer_model
  war
  1.0-SNAPSHOT
  kafka_multithread_consumer_model Maven Webapp
  http://maven.apache.org


  
    UTF-8
    1.8
    1.8
  

  
    
      org.apache.kafka
      kafka-clients
      0.11.0.0
    
  

  
    kafka_multithread_consumer_model
  

 3.4 方案一：Consumer Group

　　ProducerThread.java是一个生产者线程，发送消息到Broker

　　ConsumerThread.java是一个消费者线程，由于消费消息

　　ConsumerGroup.java用于产生一组消费者线程

　　ConsumerGroupMain.java是入口类     

3.4.1 ProducerThread.java　

 View Code

 3.4.2 ConsumerThread.java

 View Code

3.4.3 ConsumerGroup.java

 View Code

3.4.4 ConsumerGroupMain.java　　

 View Code

3.5 方案二：多线程的Consumer

　　ConsumerThreadHandler.java用于处理发送到消费者的消息

　　ConsumerThread.java是消费者使用线程池的方式初始化消费者线程

　　ConsumerThreadMain.java是入口类

3.5.1 ConsumerThreadHandler.java

 View Code

3.5.2 ConsumerThread.java

 View Code

3.5.3 ConsumerThreadMain.java

 View Code

四. 总结

　　本篇文章列举了两种不同的消费者模式。两者各有利弊。所有代码都上传到了https://github.com/qizhelongdeyang/kafka_multithread_consumer_model.git ，如有疑问或者错误请指正