批量处理类：使用Java8 函数操作

    1. //
    2. // Source code recreated from a .class file by IntelliJ IDEA
    3. // (powered by FernFlower decompiler)
    4. //
    5. import java.util.Collection;
    6. import java.util.function.Consumer;
    7. import java.util.stream.Collectors;
    8. import java.util.stream.Stream;
    10. public final class BatchJobUtils {
    11.     public static int DEFAULT_MAX_BATCH_SIZE = 1000;
    13.     public BatchJobUtils() {
    14.     }
    16.     public static int countStep(int size) {
    17.         if (size < DEFAULT_MAX_BATCH_SIZE) {
    18.             return 1;
    19.         } else {
    20.             int step = size / DEFAULT_MAX_BATCH_SIZE;
    21.             return size % DEFAULT_MAX_BATCH_SIZE == 0 ? step : step + 1;
    22.         }
    23.     }
    25.     public static int countStep(int size, int limit) {
    26.         if (size < limit) {
    27.             return 1;
    28.         } else {
    29.             int step = size / limit;
    30.             return size % limit == 0 ? step : step + 1;
    31.         }
    32.     }
    34.     public static void doBatch(int size, Consumer<Integer> consumer) {
    35.         int step = countStep(size, DEFAULT_MAX_BATCH_SIZE);
    36.         ((Stream)Stream.iterate(0, (n) -> {
    37.             return n + 1;
    38.         }).limit((long)step).parallel()).forEach(consumer);
    39.     }
    41.     public static <T> void doBatch(Collection<T> list, Consumer<Collection<T>> consumer) {
    42.         int step = countStep(list.size(), DEFAULT_MAX_BATCH_SIZE);
    43.         ((Stream)Stream.iterate(0, (n) -> {
    44.             return n + 1;
    45.         }).limit((long)step).parallel()).forEach((n) -> {
    46.             consumer.accept(((Stream)list.stream().skip((long)n * (long)DEFAULT_MAX_BATCH_SIZE).limit((long)DEFAULT_MAX_BATCH_SIZE).parallel()).collect(Collectors.toList()));
    47.         });
    48.     }
    50.     public static <T> void doBatch(Collection<T> list, int batchSize, Consumer<Collection<T>> consumer) {
    51.         int step = countStep(list.size(), batchSize);
    52.         ((Stream)Stream.iterate(0, (n) -> {
    53.             return n + 1;
    54.         }).limit((long)step).parallel()).forEach((n) -> {
    55.             consumer.accept(((Stream)list.stream().skip((long)n * (long)batchSize).limit((long)batchSize).parallel()).collect(Collectors.toList()));
    56.         });
    57.     }
    58. }

    该工具类的使用

    1. // 查询名称重复的供应商
    2. CountDownLatch querySupplierCountDownLatch = new CountDownLatch(BatchJobUtils.countStep(supplierNameSet.size()));
    4. // doBatch 前一个是要消费的数据，后面是消费的具体逻辑，这里使用到了多线程去处理逻辑，组装了对象，十分的方便
    5. BatchJobUtils.doBatch(Arrays.asList(supplierNameSet.toArray()), subList -> {
    6.     threadPoolTaskExecutor.execute(() -> {
    7.         // 供应商查重
    8.         List<JdiSupplierDto> supplierDtoList = this.jdiSupplierService.listBySupplierNames(subList, jdiImportExportOperateMsg.getShopId());
    9.         if (CollectionUtils.isNotEmpty(supplierDtoList)) {
    10.             for (JdiSupplierDto supplierDto : supplierDtoList) {
    11.                 supplierMap.put(supplierDto.getSupplierName(), supplierDto);
    12.             }
    13.         }
    14.         querySupplierCountDownLatch.countDown();
    15.     });
    16. });
    17. try {
    18.     querySupplierCountDownLatch.await(30, TimeUnit.SECONDS);
    19. } catch (InterruptedException e) {
    20.     throw new BasicException("importAddOrModifyJdiSupplierArray.findJdiSupplierBySupplierNames", ResultEnum.OUTER_SERVICE_EXCEPTION.getMessage(), e);
    21. }