Kafka 使用ExecutorService 进行消费

---

前言：

     Apache Kafka 作为当下最常用消息中间件之一。给到我的需求是需要我们处理大量的消息（如果单线程处理过多消息会出现性能瓶颈）。

如何使用Java的ExecutorService框架来创建线程池处理大量消息？

  1.创建一个可以从topic中poll()消息后传递到线程池以进行进一步处理。

  2.创建工作线程，以执行每条消息的进一步处理。

1.topic消息传递到ThreadPoolExecutorService

/** kafka 消息处理*/
public class KafkaProcessor {
    private final KafkaConsumer<String, String> myConsumer;
    private ExecutorService executor;
    private static final Properties KAFKA_PROPERTIES = new Properties();

   //基础的kafka配置~
   static {
        KAFKA_PROPERTIES.put("bootstrap.servers", "localhost:9092");
        KAFKA_PROPERTIES.put("group.id", "test-consumer-group");
        KAFKA_PROPERTIES.put("enable.auto.commit", "true");
        KAFKA_PROPERTIES.put("auto.commit.interval.ms", "1000");
        KAFKA_PROPERTIES.put("session.timeout.ms", "30000");
        KAFKA_PROPERTIES.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KAFKA_PROPERTIES.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    }

    public KafkaProcessor() {
        this.myConsumer = new KafkaConsumer<>(KAFKA_PROPERTIES);//初始化配置
        this.myConsumer.subscribe(Arrays.asList("test")); //订阅topic=test
    }
    public void init(int numberOfThreads) {
      //创建一个线程池 
     /**
      * public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler)
       *corePoolSize : 核心线程数，一旦创建将不会再释放。如果创建的线程数还没有达到指定的核心线    程数量，将会继续创建新的核心线程，直到达到最大核心线程数后，核心线程数将不在增加；如果没有空闲的核心线程，同时又未达到最大线程数，则将继续创建非核心线程；如果核心线程数等于最大线程数，则当核心线程都处于激活状态时，任务将被挂起，等待空闲线程来执行。

       *maximumPoolSize : 最大线程数，允许创建的最大线程数量。如果最大线程数等于核心线程数，则无法创建非核心线程；如果非核心线程处于空闲时，超过设置的空闲时间，则将被回收，释放占用的资源。

       *keepAliveTime : 也就是当线程空闲时，所允许保存的最大时间，超过这个时间，线程将被释放销毁，但只针对于非核心线程。

       *unit : 时间单位，TimeUnit.SECONDS等。

       *workQueue : 任务队列，存储暂时无法执行的任务，等待空闲线程来执行任务。

       *threadFactory :  线程工程，用于创建线程。

       *handler : 当线程边界和队列容量已经达到最大时，用于处理阻塞时的程序
      */

      executor = new ThreadPoolExecutor(numberOfThreads, numberOfThreads,0L,TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(1000), new ThreadPoolExecutor.CallerRunsPolicy());

    while (true) {
        ConsumerRecords<String, String> records = myConsumer.poll(100);//每隔时间段进行消息拉取
        for (final ConsumerRecord<String, String> record : records) {
        executor.submit(new KafkaRecordHandler(record)); 
        }
      }
}
    //别忘记线程池的关闭！
    public void shutdown() {
        if (myConsumer != null) {
        myConsumer.close();
        }
        if (executor != null) {
        executor.shutdown();
        }
        try {
          if (executor != null && !executor.awaitTermination(60, TimeUnit.MILLISECONDS)) {
          executor.shutdownNow();
          }
        }catch (InterruptedException e) {
        executor.shutdownNow();
        }
}
}

image.gif

2.创建工作线程

// 创建消息线程进行处理
public class KafkaRecordHandler implements Runnable {

    private ConsumerRecord<String, String> record;

    public KafkaRecordHandler(ConsumerRecord<String, String> record) {
        this.record = record;
    }

    @Override
    public void run() { 
      //业务操作...
        System.out.println("value = "+record.value());
        System.out.println("Thread id = "+ Thread.currentThread().getId());
    }
}

image.gif

3.Using ？

//消费测试
public class ConsumerTest {
    public static void main(String[] args) {
      KafkaProcessor processor = new KafkaProcessor();
      try {
          processor.init(5);//指定相应的线程数！
      }catch (Exception exp) {
          processor.shutdown();
      }
    }
}

image.gif

4.总结

可能并不适合所有方案，按需定制方案。

---