devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

kafka Consumer 消费者使用多线程并发执行，并保证顺序消费， 第一种使用纯线程方式、第二种使用Executors线程池_kafka消费者多线程消费

作者：我家小花儿 | 2024-04-14 03:07:17

赞

踩

kafka消费者多线程消费

需求内容

单个消费者，每秒需要处理1000条数据，每条数据的处理时间为500ms，相同accNum（客户账号）的数据需要保证消费的顺序。

注意点

1、如果1秒钟生产1000条数据，消费者处理时，每条数据需要500毫秒，则消费者每次拉取数据的条数最好能控制在500条以上，这样1秒内的数据可以拉取两次，每次使用500个线程进行处理，每次耗时500ms,
2*500ms=1秒，基本可以保证1000条数据能够在1秒内处理完成。
如果消费者每100ms拉取一次，每次拉取100条数据，消费者使用100个线程处理这100条数据，耗时500ms，第二次再拉取100条，耗时500ms...这样处理完1秒内的1000条数据将一共需要
10次*500ms=5秒钟，出现较大延迟。
同时，还要注意，一批数据中存在相同的accNum（客户账号）的情况，如果存在2条相同的accNum，因为需要顺序执行，一条执行需要500ms，两条顺序执行完成将花费1秒，这批数据的整体完成时间将变为1秒。
注意这三个参数的调整：
// fetch.max.bytes：一次拉取的最小可返回数据量：1Bety
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 100 * 1024);
// fetch.max.wait.ms：一次拉取的最大等待时间：500ms
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
// max.poll.records: 一次拉取的最大条数
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2000);
注意消费者的拉取延迟时间：
tKafkaConsumer.poll(500);
2、每批次数据处理时，创建的线程数，会根据每次拉取的数据条数自动调整，最大线程数为消费者每次允许拉取的最大数据条数。这样系统可以根据数据量大小自动调整创建的线程数，线程池中的空闲线程可以在一定时间后自动释放。可以保证不同accNum（客户账号）的数据每次都分配一个线程单独处理，从而保证处理的时间（500ms）。

第一种使用纯线程方式（Thread+Callable+FutureTask）

因为每次处理都创建新的线程，造成大量线程同时创建和销毁，线程数波动剧烈，GC频繁，系统各项指标均不平稳。

  1. package org.fiend.kafka.config.kafka;
  2.  
  3. import org.apache.kafka.clients.consumer.*;
  4. import org.apache.kafka.common.TopicPartition;
  5. import org.apache.kafka.common.serialization.StringDeserializer;
  6. import org.slf4j.Logger;
  7. import org.slf4j.LoggerFactory;
  8.  
  9. import java.time.Duration;
  10. import java.util.*;
  11. import java.util.concurrent.ExecutionException;
  12. import java.util.concurrent.FutureTask;
  13.  
  14. /**
  15.  * @author 86133 2023-07-06 15:42:52
  16.  */
  17. public class CustomConsumer {
  18.     private static Logger log = LoggerFactory.getLogger(CustomConsumer.class);
  19.  
  20.     public static void main(String[] args) throws InterruptedException {
  21.         Properties props = new Properties();
  22.         // bootstrap.servers：kafka集群地址
  23.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  24.         // 消费者组id
  25.         props.put("group.id", "test_consumer_group"); // 消费者组
  26.  
  27.         // key.deserializer：key的反序列化器
  28.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  29.  
  30.         // value.deserializer：value的反序列化器
  31.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  32.  
  33.         // fetch.max.bytes：一次拉取的最小可返回数据量：1Byte
  34.         props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 100 * 1024);
  35.  
  36.         // fetch.max.bytes：一次拉取的最大数据量：50M
  37.         props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 50 * 1024 * 1024);
  38.  
  39.         // fetch.max.wait.ms：一次拉取的最大等待时间：500ms
  40.         props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
  41.  
  42.         // max.poll.records: 一次拉取的最大条数
  43.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2000);
  44.  
  45.         // max.partition.fetch.bytes：一次拉取时，每个分区最大拉取数据量，默认1M
  46.         props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1 * 1024 * 1024);
  47.  
  48.         // auto.offset.reset：当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在 （如，数据被删除了）时，自动设置开始消费的偏移量位置，默认latest。
  49.         // earliest：自动重置偏移量到最早的偏移量（从头开始消费）。
  50.         // latest：默认，自动重置偏移量为最新的偏移量（从最新的接收到的数据开始消费）。
  51.         // none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。 anything：向消费者抛异常。
  52.         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
  53.  
  54.         // enable.auto.commit：是否允许自动提交offset，默认是。
  55.         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  56.  
  57.         // auto.commit.interval.ms：自动提交offset的时间间隔，默认5秒。
  58.         props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
  59.  
  60.         // heartbeat.interval.ms：消费者心跳检测时间间隔，默认3秒。
  61.         props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
  62.  
  63.         // session.timeout.ms：session过期时间，默认10秒。
  64.         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
  65.  
  66.         // max.poll.interval.ms：一批次数据最大可以执行时间，默认5分钟。
  67.         props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
  68.  
  69.         // partition.assignment.strategy：分区分配策略，默认5分钟。
  70.         props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName());
  71.  
  72.         KafkaConsumer<String, String> tKafkaConsumer = new KafkaConsumer<>(props);
  73.         tKafkaConsumer.subscribe(Arrays.asList("riskMoniterTopic"));
  74.  
  75.         HashMap<String, List<String>> hashMap;
  76.         while (true) {
  77.             long start = System.currentTimeMillis();
  78.             log.info("[开始]-consumer拉取数据");
  79.             ConsumerRecords<String, String> records = tKafkaConsumer.poll(Duration.ofMillis(500));
  80.             int dataCount = records.count();
  81.             log.info("[完成]-consumer拉取数据-条数=[{}]-耗时=[{}]", dataCount, System.currentTimeMillis() - start);
  82.  
  83.             // 拉取的数据条数大于0时，才进行处理操作
  84.             if (dataCount > 0) {
  85.                 // 初始化hashMap：容量可以设置为拉取数据条数的1倍，或者2倍，2倍更加分散
  86.                 // 消费者参数中设置一次拉取的最大条数为2000，基本不会超过该值。
  87.                 // hashMap的hash碰撞概率较低，2000条数据，分布到4000容量的hashMap中时，基本不会出现碰撞，只有相同的key会在一起，导致整体执行时间为相同多个key顺序执行的时间
  88.                 // [线程执行完成]消费者线程：consumer-thread-VV0039-已处理数据数量=3-已处理的所有客户账号=VV0039,VV0039,VV0039,
  89.                 // [线程执行完成]消费者线程：consumer-thread-AG0097-已处理数据数量=2-已处理的所有客户账号=AG0097,AG0097,
  90.                 // [线程执行完成]消费者线程：consumer-thread-ID0045-已处理数据数量=1-已处理的所有客户账号=ID0045,
  91.                 int arrListCapacity = dataCount * 2;
  92.                 hashMap = new HashMap<>(arrListCapacity);
  93.                 // 将拉取的数据按客户号码分散到HashMap中
  94.                 for (ConsumerRecord<String, String> record : records) {
  95.                     String value = record.value();
  96.                 }
  97.  
  98.                 ArrayList<FutureTask<String>> tFutureTaskArrayList = new ArrayList<>(dataCount);
  99.                 // 循环hashMap，每个value开启一个线程循环处理该List中的全部数据，保证数据处理的顺序
  100.                 hashMap.forEach((k, v) -> {
  101.                     List<String> list = v;
  102.                     String threadName = "";
  103.                     if (list.size() > 0) {
  104.                         threadName = "consumer-thread-" + k;
  105.                         // 使用Callable执行一组数据
  106.                         FutureTask<String> futureTask = new FutureTask<>(() -> {
  107.                             String threadName1 = Thread.currentThread().getName();
  108.                             // log.info("[获取]-消费者线程：{}-获取到待处理数据数量：{}", threadName, busiDataEntities.size());
  109.                             for (String dataStr : list) {
  110.                                 try {
  111.                                     // 模拟业务处理时间，默认500ms
  112.                                     Thread.sleep(500);
  113.                                 } catch (InterruptedException e) {
  114.                                     e.printStackTrace();
  115.                                 }
  116.                             }
  117.                             return "消费者线程：" + threadName1 + "-已处理数据数量=" + list.size() + "-已处理的所有客户账号 = " + list;
  118.                         });
  119.  
  120.                         // 启动一个线程执行一组数据
  121.                         new Thread(futureTask, threadName).start();
  122.                         // 将每个线程的futureTask都放入同一个ArrayList中
  123.                         tFutureTaskArrayList.add(futureTask);
  124.                     }
  125.                 });
  126.  
  127.                 // 循环tFutureTaskArrayList，检查所有futureTask是否都已经返回，没返回的阻塞等待，等都返回后证明所有线程都执行完成，提交offset
  128.                 // 因为每次处理都创建新的线程，大量线程同时创建和销毁，线程数波动剧烈，考虑通过线程池进行优化
  129.                 for (int i = 0; i < tFutureTaskArrayList.size(); i++) {
  130.                     try {
  131.                         String returnStr = tFutureTaskArrayList.get(i).get();
  132.                         log.info("[线程执行完成]" + returnStr);
  133.                     } catch (ExecutionException e) {
  134.                         e.printStackTrace();
  135.                     }
  136.                 }
  137.             }
  138.  
  139.             // 同步提交offset
  140.             // tKafkaConsumer.commitSync();
  141.             // 异步提交
  142.             tKafkaConsumer.commitAsync(new OffsetCommitCallback() {
  143.                 @Override
  144.                 public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
  145.                     if (exception != null) {
  146.                         log.error("[失败]-提交offset失败！" + offsets);
  147.                     } else {
  148.                         log.info("[成功]-提交offset成功！");
  149.                     }
  150.                 }
  151.             });
  152.  
  153.             log.info("【完成处理数据】-条数=[{}]-耗时=[{}]", dataCount, System.currentTimeMillis() - start);
  154.         }
  155.  
  156.     }
  157. }

测试结果：

// [开始]-consumer拉取数据
// [完成]-consumer拉取数据-条数=[2000]-耗时=[5]
// [成功]-提交offset成功！
// 【完成处理数据】-条数=[2000]-耗时=[1731]
// [开始]-consumer拉取数据
// [完成]-consumer拉取数据-条数=[2000]-耗时=[4]
// [成功]-提交offset成功！
// 【完成处理数据】-条数=[2000]-耗时=[1678]
// [开始]-consumer拉取数据
// [完成]-consumer拉取数据-条数=[2000]-耗时=[23]
// [成功]-提交offset成功！
// 【完成处理数据】-条数=[2000]-耗时=[1637]
// 测试结果：2000条可以在2秒处理完成，则可以保证1000条时可以在1秒能处理完成，满足需求内容。
// 因为每次处理都创建新的线程，造成大量线程同时创建和销毁，线程数波动剧烈，GC频繁，系统各项指标均不平稳。

第二种使用Executors线程池（Executors+Callable+FutureTask）

通过线程池进行处理，线程数一直保持在2000个左右。

  1. package com.autoee.demo.kafka.main;
  2.  
  3. import ch.qos.logback.classic.Level;
  4. import ch.qos.logback.classic.LoggerContext;
  5. import cn.hutool.core.date.DateUtil;
  6. import cn.hutool.core.date.TimeInterval;
  7. import cn.hutool.json.JSONUtil;
  8. import com.autoee.demo.riskmonitor.BusiDataEntity;
  9. import org.apache.kafka.clients.consumer.*;
  10. import org.apache.kafka.common.TopicPartition;
  11. import org.apache.kafka.common.serialization.StringDeserializer;
  12. import org.slf4j.Logger;
  13. import org.slf4j.LoggerFactory;
  14.  
  15. import java.util.*;
  16. import java.util.concurrent.*;
  17. import java.util.concurrent.atomic.AtomicInteger;
  18.  
  19. /**
  20.  * Title: <br>
  21.  * Desc: <br>
  22.  * Date: 2022-8-19 <br>
  23.  * @author Double
  24.  * @version 1.0.0
  25.  */
  26. public class KafkaConsumerMutiThreadsTest4_Executors_HashMap {
  27.  
  28.     private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerMutiThreadsTest4_Executors_HashMap.class);
  29.  
  30.     // 设置main方法执行时的日志输出级别
  31.     static {
  32.         LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
  33.         List<ch.qos.logback.classic.Logger> loggerList = loggerContext.getLoggerList();
  34.         loggerList.forEach(logger -> {
  35.             logger.setLevel(Level.INFO);
  36.         });
  37.     }
  38.  
  39.     // 需求内容：单个消费者，每秒需要处理1000条数据，每条数据的处理时间为500ms，相同accNum的数据需要保证消费的顺序
  40.  
  41.     // 测试极限情况：数据已存在大量积压，启动消费者进行消费
  42.     // 每次拉取都达到设置的单次可以拉取的最大条数：2000条
  43.  
  44.     public static void main(String[] args) throws InterruptedException {
  45.  
  46.         Properties props = new Properties();
  47.         // bootstrap.servers：kafka集群地址
  48.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  49.         // 消费者组id
  50.         props.put("group.id", "test_consumer_group"); //消费者组
  51.         // key.deserializer：key的反序列化器
  52.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  53.         // value.deserializer：value的反序列化器
  54.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  55.         // fetch.max.bytes：一次拉取的最小可返回数据量：1Bety
  56.         props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 100 * 1024);
  57.         // fetch.max.bytes：一次拉取的最大数据量：50M
  58.         props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 50 * 1024 * 1024);
  59.         // fetch.max.wait.ms：一次拉取的最大等待时间：500ms
  60.         props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
  61.         // max.poll.records: 一次拉取的最大条数
  62.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2000);
  63.         // max.partition.fetch.bytes：一次拉取时，每个分区最大拉取数据量，默认1M
  64.         props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1 * 1024 * 1024);
  65.         // auto.offset.reset：当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在 （如，数据被删除了）时，自动设置开始消费的偏移量位置，默认latest。
  66.         // earliest：自动重置偏移量到最早的偏移量（从头开始消费）。
  67.         // latest：默认，自动重置偏移量为最新的偏移量（从最新的接收到的数据开始消费）。
  68.         // none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。 anything：向消费者抛异常。
  69.         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
  70.         // enable.auto.commit：是否允许自动提交offset，默认是。
  71.         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  72.         // auto.commit.interval.ms：自动提交offset的时间间隔，默认5秒。
  73.         props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
  74.         // heartbeat.interval.ms：消费者心跳检测时间间隔，默认3秒。
  75.         props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
  76.         // session.timeout.ms：session过期时间，默认10秒。
  77.         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
  78.         // max.poll.interval.ms：一批次数据最大可以执行时间，默认5分钟。
  79.         props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
  80.         // partition.assignment.strategy：分区分配策略，默认5分钟。
  81.         props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName());
  82.  
  83.         KafkaConsumer<String, String> tKafkaConsumer = new KafkaConsumer<String, String>(props);
  84.         tKafkaConsumer.subscribe(Arrays.asList("riskMoniterTopic"));
  85.  
  86.         // 使用Executors中的CachedThreadPool，初始核心线程数为0，最大线程数为无限大，线程最大空闲时间为60秒
  87.         // corePoolSize=0
  88.         // maximumPoolSize=Integer.MAX_VALUE，即2147483647，基本属于无界。
  89.         // keepAliveTime=60秒
  90.         // 工作队列使用没有容量的 SynchronousQueue，来一个任务处理一个任务，不进行缓存。如果提交任务速度高于线程池中线程处理任务的速度，则会不断创建新线程。极端情况下会创建过多的线程，耗尽 CPU 和内存资源。
  91.         // 可以自定义线程池进行优化
  92.         ExecutorService executorService = Executors.newCachedThreadPool();
  93.  
  94.         HashMap<String, List<BusiDataEntity>> busiDataHashMap;
  95.         while (true) {
  96.             TimeInterval timer = DateUtil.timer();
  97.             logger.info("[开始]-consumer拉取数据");
  98.             ConsumerRecords<String, String> records = tKafkaConsumer.poll(500);
  99.             int dataCount = records.count();
  100.             AtomicInteger tAtomicInteger = new AtomicInteger();
  101.             logger.info("[完成]-consumer拉取数据-条数=[{}]-耗时=[{}]", dataCount, timer.intervalMs());
  102.             // 拉取的数据条数大于0时，才进行处理操作
  103.             timer = DateUtil.timer();
  104.             if (dataCount > 0) {
  105.                 // 初始化hashMap：容量可以设置为拉取数据条数的1倍，或者2倍，2倍更加分散
  106.                 // 消费者参数中设置一次拉取的最大条数为2000，基本不会超过该值。
  107.                 // hashMap的hash碰撞概率较低，2000条数据，分布到4000容量的hashMap中时，基本不会出现碰撞，只有相同的key会在一起，导致整体执行时间为相同多个key顺序执行的时间
  108.                 // [线程执行完成]消费者线程：pool-1-thread-1898-已处理数据数量=3-已处理的所有客户账号=GW0032,GW0032,GW0032,
  109.                 // [线程执行完成]消费者线程：pool-1-thread-1193-已处理数据数量=2-已处理的所有客户账号=KE0055,KE0055,
  110.                 // [线程执行完成]消费者线程：pool-1-thread-1187-已处理数据数量=2-已处理的所有客户账号=0E0005,0E0005,
  111.                 int capacity = dataCount * 2;
  112.                 busiDataHashMap = new HashMap<>(capacity);
  113.                 // 将拉取的数据按客户号码分散到HashMap中
  114.                 for (ConsumerRecord<String, String> record : records) {
  115.                     Object value = record.value();
  116.                     String jsonStr = JSONUtil.toJsonStr(value);
  117.                     // logger.info("[获取]-传入报文=[{}]", jsonStr);
  118.                     BusiDataEntity busiDataEntity = JSONUtil.toBean(jsonStr, BusiDataEntity.class);
  119.                     String accNum = busiDataEntity.getAccNum();
  120.  
  121.                     if (busiDataHashMap.containsKey(accNum)) {
  122.                         busiDataHashMap.get(accNum).add(busiDataEntity);
  123.                     } else {
  124.                         List<BusiDataEntity> newList = new ArrayList<>();
  125.                         newList.add(busiDataEntity);
  126.                         busiDataHashMap.put(accNum, newList);
  127.                     }
  128.                 }
  129.  
  130.                 ArrayList<FutureTask<String>> tFutureTaskArrayList = new ArrayList<>(dataCount);
  131.                 // 循环hashMap，每个value开启一个线程循环处理该List中的全部数据，保证数据处理的顺序
  132.                 int num = 0;
  133.                 busiDataHashMap.forEach((k, v) -> {
  134.                     List<BusiDataEntity> busiDataEntities = v;
  135.                     String threadName = "";
  136.                     if (busiDataEntities.size() > 0) {
  137.                         threadName = k;
  138.                         // 使用Callable执行同一个Key下的一组数据
  139.                         FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
  140.                             @Override
  141.                             public String call() {
  142.                                 String threadName = Thread.currentThread().getName();
  143.                                 // logger.info("[获取]-消费者线程：{}-获取到待处理数据数量：{}", threadName, busiDataEntities.size());
  144.                                 String allAccNum = "";
  145.                                 for (BusiDataEntity busiDataEntity : busiDataEntities) {
  146.                                     allAccNum = allAccNum + busiDataEntity.getAccNum() + ",";
  147.                                     try {
  148.                                         // 模拟业务处理时间，默认500ms
  149.                                         Thread.sleep(500);
  150.                                     } catch (InterruptedException e) {
  151.                                         e.printStackTrace();
  152.                                     }
  153.                                 }
  154.                                 return "消费者线程：" + threadName + "-已处理数据数量=" + busiDataEntities.size() + "-已处理的所有客户账号=" + allAccNum;
  155.                             }
  156.                         });
  157.  
  158.                         // 通过线程池进行任务处理
  159.                         executorService.submit(futureTask);
  160.                         // 将每个线程的futureTask都放入同一个ArrayList中
  161.                         tFutureTaskArrayList.add(futureTask);
  162.                     }
  163.                 });
  164.                 // 循环tFutureTaskArrayList，检查所有futureTask是否都已经返回，没返回的阻塞等待，等都返回后证明所有线程都执行完成，提交offset
  165.                 // 使用线程池后，线程数一直保持在2000个左右。
  166.                 for (int i = 0; i < tFutureTaskArrayList.size(); i++) {
  167.                     try {
  168.                         String returnStr = tFutureTaskArrayList.get(i).get();
  169.                         logger.info("[线程执行完成]" + returnStr);
  170.                     } catch (ExecutionException e) {
  171.                         e.printStackTrace();
  172.                     }
  173.                 }
  174.             }
  175.  
  176.             //同步提交offset
  177.             // tKafkaConsumer.commitSync();
  178.             //异步提交
  179.             tKafkaConsumer.commitAsync(new OffsetCommitCallback() {
  180.                 @Override
  181.                 public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
  182.                     if (exception != null) {
  183.                         logger.error("[失败]-提交offset失败！" + offsets);
  184.                     } else {
  185.                         logger.info("[成功]-提交offset成功！");
  186.                     }
  187.                 }
  188.             });
  189.  
  190.             logger.info("【完成处理数据】-条数=[{}]-耗时=[{}]", dataCount, timer.intervalMs());
  191.         }
  192.     }
  193. }

测试结果：

// [开始]-consumer拉取数据
// [完成]-consumer拉取数据-条数=[2000]-耗时=[5]
// [成功]-提交offset成功！
// 【完成处理数据】-条数=[2000]-耗时=[1731]
// [开始]-consumer拉取数据
// [完成]-consumer拉取数据-条数=[2000]-耗时=[4]
// [成功]-提交offset成功！
// 【完成处理数据】-条数=[2000]-耗时=[1678]
// [开始]-consumer拉取数据
// [完成]-consumer拉取数据-条数=[2000]-耗时=[23]
// [成功]-提交offset成功！
// 【完成处理数据】-条数=[2000]-耗时=[1637]
// 测试结果：2000条可以在2秒处理完成，则可以保证1000条时可以在1秒能处理完成，满足需求内容。
// 使用线程池后，线程数一直保持在2000个左右

第三种使用Executors线程池（Executors+Runnable+CountDownLatch）

  1. package com.autoee.demo.kafka.main;
  2.  
  3. import ch.qos.logback.classic.Level;
  4. import ch.qos.logback.classic.LoggerContext;
  5. import cn.hutool.core.date.DateUtil;
  6. import cn.hutool.core.date.TimeInterval;
  7. import cn.hutool.json.JSONUtil;
  8. import com.autoee.demo.riskmonitor.BusiDataEntity;
  9. import org.apache.kafka.clients.consumer.*;
  10. import org.apache.kafka.common.TopicPartition;
  11. import org.apache.kafka.common.serialization.StringDeserializer;
  12. import org.slf4j.Logger;
  13. import org.slf4j.LoggerFactory;
  14.  
  15. import java.util.*;
  16. import java.util.concurrent.*;
  17. import java.util.concurrent.atomic.AtomicInteger;
  18.  
  19. /**
  20.  * Title: <br>
  21.  * Desc: <br>
  22.  * Date: 2022-8-19 <br>
  23.  * @author Double
  24.  * @version 1.0.0
  25.  */
  26. public class KafkaConsumerMutiThreadsTest5 {
  27.     private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerMutiThreadsTest5.class);
  28.  
  29.     // 设置main方法执行时的日志输出级别
  30.     static {
  31.         LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
  32.         List<ch.qos.logback.classic.Logger> loggerList = loggerContext.getLoggerList();
  33.         loggerList.forEach(logger -> {
  34.             logger.setLevel(Level.INFO);
  35.         });
  36.     }
  37.  
  38.     // 需求内容：单个消费者，每秒需要处理1000条数据，每条数据的处理时间为500ms，相同accNum的数据需要保证消费的顺序
  39.  
  40.     // 测试极限情况：数据已存在大量积压，启动消费者进行消费
  41.     // 每次拉取都达到设置的单次可以拉取的最大条数：2000条
  42.  
  43.     // [开始]-consumer拉取数据
  44.     // [完成]-consumer拉取数据-条数=[2000]-耗时=[5]
  45.     // [成功]-提交offset成功！
  46.     // 【完成处理数据】-条数=[2000]-耗时=[1731]
  47.     // [开始]-consumer拉取数据
  48.     // [完成]-consumer拉取数据-条数=[2000]-耗时=[4]
  49.     // [成功]-提交offset成功！
  50.     // 【完成处理数据】-条数=[2000]-耗时=[1678]
  51.     // [开始]-consumer拉取数据
  52.     // [完成]-consumer拉取数据-条数=[2000]-耗时=[23]
  53.     // [成功]-提交offset成功！
  54.     // 【完成处理数据】-条数=[2000]-耗时=[1637]
  55.  
  56.     // 测试结果：2000条可以在2秒处理完成，则可以保证1000条时可以在1秒能处理完成，满足需求内容。
  57.     // 通过线程池进行处理，线程数非常平稳，而且只需要十个左右线程就能处理每次2000条的数据。
  58.  
  59.     public static void main(String[] args) throws InterruptedException {
  60.  
  61.         Properties props = new Properties();
  62.         // bootstrap.servers：kafka集群地址
  63.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  64.         // 消费者组id
  65.         props.put("group.id", "test_consumer_group"); //消费者组
  66.         // key.deserializer：key的反序列化器
  67.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  68.         // value.deserializer：value的反序列化器
  69.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  70.         // fetch.max.bytes：一次拉取的最小可返回数据量：1Bety
  71.         props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 100 * 1024);
  72.         // fetch.max.bytes：一次拉取的最大数据量：50M
  73.         props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 50 * 1024 * 1024);
  74.         // fetch.max.wait.ms：一次拉取的最大等待时间：500ms
  75.         props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
  76.         // max.poll.records: 一次拉取的最大条数
  77.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2000);
  78.         // max.partition.fetch.bytes：一次拉取时，每个分区最大拉取数据量，默认1M
  79.         props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1 * 1024 * 1024);
  80.         // auto.offset.reset：当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在 （如，数据被删除了）时，自动设置开始消费的偏移量位置，默认latest。
  81.         // earliest：自动重置偏移量到最早的偏移量（从头开始消费）。
  82.         // latest：默认，自动重置偏移量为最新的偏移量（从最新的接收到的数据开始消费）。
  83.         // none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。 anything：向消费者抛异常。
  84.         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
  85.         // enable.auto.commit：是否允许自动提交offset，默认是。
  86.         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  87.         // auto.commit.interval.ms：自动提交offset的时间间隔，默认5秒。
  88.         props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
  89.         // heartbeat.interval.ms：消费者心跳检测时间间隔，默认3秒。
  90.         props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
  91.         // session.timeout.ms：session过期时间，默认10秒。
  92.         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
  93.         // max.poll.interval.ms：一批次数据最大可以执行时间，默认5分钟。
  94.         props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
  95.         // partition.assignment.strategy：分区分配策略，默认5分钟。
  96.         props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName());
  97.  
  98.         KafkaConsumer<String, String> tKafkaConsumer = new KafkaConsumer<String, String>(props);
  99.         tKafkaConsumer.subscribe(Arrays.asList("riskMoniterTopic"));
  100.  
  101.         // 使用Executors中的CachedThreadPool，初始核心线程数为0，最大线程数为无限大，线程最大空闲时间为60秒
  102.         // corePoolSize=0
  103.         // maximumPoolSize=Integer.MAX_VALUE，即2147483647，基本属于无界。
  104.         // keepAliveTime=60秒
  105.         // 工作队列使用没有容量的 SynchronousQueue，来一个任务处理一个任务，不进行缓存。如果提交任务速度高于线程池中线程处理任务的速度，则会不断创建新线程。极端情况下会创建过多的线程，耗尽 CPU 和内存资源。
  106.         // 可以自定义线程池进行优化
  107.         ExecutorService executorService = Executors.newCachedThreadPool();
  108.  
  109.         HashMap<String, List<BusiDataEntity>> busiDataHashMap;
  110.         while (true) {
  111.             TimeInterval timer = DateUtil.timer();
  112.             logger.info("[开始]-consumer拉取数据");
  113.             ConsumerRecords<String, String> records = tKafkaConsumer.poll(500);
  114.             int dataCount = records.count();
  115.             AtomicInteger tAtomicInteger = new AtomicInteger();
  116.             logger.info("[完成]-consumer拉取数据-条数=[{}]-耗时=[{}]", dataCount, timer.intervalMs());
  117.             // 拉取的数据条数大于0时，才进行处理操作
  118.             timer = DateUtil.timer();
  119.             if (dataCount > 0) {
  120.                 // 初始化hashMap：容量可以设置为拉取数据条数的1倍，或者2倍，2倍更加分散
  121.                 // 消费者参数中设置一次拉取的最大条数为2000，基本不会超过该值。
  122.                 // hashMap的hash碰撞概率较低，2000条数据，分布到4000容量的hashMap中时，基本不会出现碰撞，只有相同的key会在一起，导致整体执行时间为相同多个key顺序执行的时间
  123.                 // [线程执行完成]消费者线程：pool-1-thread-1898-已处理数据数量=3-已处理的所有客户账号=GW0032,GW0032,GW0032,
  124.                 // [线程执行完成]消费者线程：pool-1-thread-1193-已处理数据数量=2-已处理的所有客户账号=KE0055,KE0055,
  125.                 // [线程执行完成]消费者线程：pool-1-thread-1187-已处理数据数量=2-已处理的所有客户账号=0E0005,0E0005,
  126.                 int capacity = dataCount * 2;
  127.                 busiDataHashMap = new HashMap<>(capacity);
  128.                 // 将拉取的数据按客户号码分散到ArrayList中
  129.                 for (ConsumerRecord<String, String> record : records) {
  130.                     Object value = record.value();
  131.                     String jsonStr = JSONUtil.toJsonStr(value);
  132.                     // logger.info("[获取]-传入报文=[{}]", jsonStr);
  133.                     BusiDataEntity busiDataEntity = JSONUtil.toBean(jsonStr, BusiDataEntity.class);
  134.                     String accNum = busiDataEntity.getAccNum();
  135.  
  136.                     if (busiDataHashMap.containsKey(accNum)) {
  137.                         busiDataHashMap.get(accNum).add(busiDataEntity);
  138.                     } else {
  139.                         List<BusiDataEntity> newList = new ArrayList<>();
  140.                         newList.add(busiDataEntity);
  141.                         busiDataHashMap.put(accNum, newList);
  142.                     }
  143.                 }
  144.  
  145.                 // 循环ArrayList，每个下标中的List数据条数大于0时，开启一个线程循环处理该List中的全部数据，保证数据处理的顺序
  146.                 int num = 0;
  147.                 int busiDataHashMapSize = busiDataHashMap.keySet().size();
  148.                 // 使用CountDownLatch判断是否所有子线程都已执行完成，子线程个数等于busiDataHashMap中key的个数
  149.                 CountDownLatch tCountDownLatch = new CountDownLatch(busiDataHashMapSize);
  150.                 busiDataHashMap.forEach((k, v) -> {
  151.                     List<BusiDataEntity> busiDataEntities = v;
  152.                     String threadName = "";
  153.                     if (busiDataEntities.size() > 0) {
  154.                         threadName = k;
  155.                         // 使用Runnable执行同一个Key下的一组数据
  156.                         Runnable runnableTask = new Runnable() {
  157.                             @Override
  158.                             public void run() {
  159.                                 String threadName = Thread.currentThread().getName();
  160.                                 // logger.info("[获取]-消费者线程：{}-获取到待处理数据数量：{}", threadName, busiDataEntities.size());
  161.                                 String allAccNum = "";
  162.                                 String allBatchNo = "";
  163.                                 for (BusiDataEntity busiDataEntity : busiDataEntities) {
  164.                                     allAccNum = allAccNum + busiDataEntity.getAccNum() + ",";
  165.                                     allBatchNo = allBatchNo + busiDataEntity.getBatchNo() + ",";
  166.                                     try {
  167.                                         // 模拟业务处理时间，默认500ms
  168.                                         Thread.sleep(500);
  169.                                     } catch (InterruptedException e) {
  170.                                         e.printStackTrace();
  171.                                     }
  172.                                 }
  173.                                 logger.info("[线程执行完成]-消费者线程：" + threadName + "-已处理数据数量=" + busiDataEntities.size() + "-已处理的所有客户账号=" + allAccNum + "-已处理的所有批次号=" + allBatchNo);
  174.                                 // 每个线程处理完成后，将tCountDownLatch减1
  175.                                 tCountDownLatch.countDown();
  176.                             }
  177.                         };
  178.  
  179.                         // 通过线程池进行任务处理
  180.                         executorService.submit(runnableTask);
  181.                     }
  182.                 });
  183.  
  184.                 // 通过CountDownLatch阻塞等待，等待所有线程都执行完成，提交offset
  185.                 tCountDownLatch.await();
  186.  
  187.                 //同步提交offset
  188.                 // tKafkaConsumer.commitSync();
  189.                 //异步提交
  190.                 tKafkaConsumer.commitAsync(new OffsetCommitCallback() {
  191.                     @Override
  192.                     public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
  193.                         if (exception != null) {
  194.                             logger.error("[失败]-提交offset失败！" + offsets);
  195.                         } else {
  196.                             logger.info("[成功]-提交offset成功！");
  197.                         }
  198.                     }
  199.                 });
  200.  
  201.                 logger.info("【完成处理数据】-条数=[{}]-耗时=[{}]", dataCount, timer.intervalMs());
  202.                 logger.info("----------------------------------------------------------------------------------------------------------------------------------------");
  203.  
  204.             }
  205.         }
  206.     }
  207. }

测试结果：

和第二种的执行时间差不多，但是各项性能指标好像更加平稳了，但是很出现线程阻塞的情况。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/419876
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号