0

У меня есть такой участок кода:

private List<NtId> getNtIds(List<String> ids) {
    List<NtId> ntIdList = new CopyOnWriteArrayList<>();

    int step = 10000;
    int stepsNumber = ids.size() / step + ids.size() % step == 0 ? 0 : 1;

    CountDownLatch countDownLatch = new CountDownLatch(stepsNumber);
    for (int i = 0; i < stepsNumber - 1; ++i) {
        int current = i * step;
        new Thread(() -> {
            ntIdList.addAll(ntIdRepository.findDistinctByNtIdIn(ids.subList(current, current + step)));
            countDownLatch.countDown();
        }).start();
    }

    int lastStep = stepsNumber * step;
    new Thread(() -> {
        ntIdList.addAll(ntIdRepository.findDistinctByNtIdIn(ids.subList(lastStep, lastStep + ids.size() % step)));
        countDownLatch.countDown();
    }).start();

    try {
        countDownLatch.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return ntIdList;
}

Смысл его заключается в том, что на вход поступает список из id и по ним надо выгрузить объекты. База данных не принимает список более 1000 значений и поэтому, чтобы ускорить процесс делаю обработку во многих потоках.

Проблема в том, что при изменениях переменной step меняется и результат, чего быть не должно. В чем я ошибаюсь и как можно улучшить этот участок ?

5
  • а почему не использовать Future? 14 июл 2017 в 14:24
  • @Mikhail Vaysman в чем преимущества использования Future? просто самому интересно =)
    – Lex Hobbit
    14 июл 2017 в 14:25
  • @MikhailVaysman покажите пример относительной этой задачи ? Не совсем понимаю, что мне это даст в данном случае.
    – faoxis
    14 июл 2017 в 14:26
  • @MikhailVaysman Да, возвращает список объектов типа NtId.
    – faoxis
    14 июл 2017 в 14:31
  • @LexHobbit код будет проще 14 июл 2017 в 16:48

1 ответ 1

1

Возможно вам подойдет вот такой вариант.

private static List<NtId> getNtIds(List<String> ids) {
    int step = 100;
    return split(ids, step).parallel()
            .map(l -> ntIdRepository.findDistinctByNtIdIn(l))
            .flatMap(List::stream)
            .collect(Collectors.toList());
}

Этот метод я взял тут Partition a Java 8 Stream. Но вы можете использовать аналогичный из Guava или любой другой библиотеки.

private static Stream<List<String>> split(List<String> list,  int size) {
    int parts = (list.size() - 1) / size;
    return IntStream.range(0, parts + 1).mapToObj(n -> list.subList(n * size, n == parts ? list.size() : (n + 1) * size));
}

В этом решении есть подводные камни - оно использует ForkJoinPool.commonPool(). А он используется внутри системы повсеместно.

Что бы запустить это внутри своего pool, надо сделать так

ForkJoinPool forkJoinPool = new ForkJoinPool(2);  

forkJoinPool.submit(() ->  
    split(ids, step).parallel()
                .map(l -> ntIdRepository.findDistinctByNtIdIn(l))
                .flatMap(List::stream)
                .collect(Collectors.toList());  
).get();  

Вот вариант с использованием Future

private static List<NtId> getNtIds(List<String> ids) {
    // размер pool можно подобрать
    ExecutorService executor = Executors.newFixedThreadPool(10);

    List<Future<List<NtId>>> futures = new ArrayList<>();
    int step = 100;

    int parts = (ids.size() - 1) / step;
    for (int i = 0; i < parts + 1; i++) {
        final List<String> subList = ids.subList(i * step, i == parts ? ids.size() : (i + 1) * step);
        futures.add(executor.submit(() -> ntIdRepository.findDistinctByNtIdIn(subList)));
    }

    List<NtId> ntIdList = new ArrayList<>();

    for (Future<List<NtId>> future : futures) {
        try {
            ntIdList.addAll(future.get());
        } catch (InterruptedException | ExecutionException e) {
            // игнорирю
        }
    }
    executor.shutdown();
    return ntIdList;
}
5
  • Отличный ответ. Спасибо огромное!
    – faoxis
    17 июл 2017 в 4:54
  • Единственный вопрос - можно ли как-нибудь ограничить количество потоков в первом варианте решения ? Скажем, я хочу, чтобы одновременно выполнялось не более 20 действий. Это можно как-то реализовать ?
    – faoxis
    17 июл 2017 в 5:01
  • 1
    добавил эту информацию в вопрос 17 июл 2017 в 18:14
  • Отлично! Не подскажите, как стрим узнает, что он запускается внутри какого-то внешнего пула ?
    – faoxis
    18 июл 2017 в 6:42
  • 1
    это особенности реализации. стрим будет использовать пул в котором запустили. 18 июл 2017 в 8:27

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.