ラムダ

2021年11月15日 / 最終更新日時 : 2021年12月1日 akiyama 技術講習

複数スレッドで平行処理し、順序そのままで結果を格納する

こんにちわ、猫好きリーマンのほげPGです。今回は複数スレッドで平行処理し、要求時の順序で結果を配列に格納するコードを紹介します。ソースコード

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

public class Hoge {

    public static void main(String[] args) throws Exception {
        log("start ...");
        ExecutorService streamPool = Executors.newFixedThreadPool(10);  //①
        List<String> requests = Arrays.asList(new String[] {"11", "22", "33", "44", "55"});  //②
        String[] array = new String[requests.size()];  //③
        try {
            CountDownLatch latch = new CountDownLatch(requests.size());  //④
            int i = 0;
            for (String val : requests) {
                Integer x = i++;           //⑤
                streamPool.submit(() -> {  //⑥
                    log("start..." + x);
                    try {
                        array[x] = doHoge(val);  //⑦
                    } catch (Exception e) {
                        log(e.toString());
                    } finally {
                        latch.countDown();
                    }
                    log("end..." + x);
                });
            }
            latch.await();
        } finally {
            streamPool.shutdown();
        }
        List<String> results = Arrays.stream(array)
        .filter(pi -> pi != null)       //⑧
        .collect(Collectors.toList());  //⑨
        log(results.toString());
        log("end ...");
    }

    static Random rand = new Random();
    static String doHoge(String v) throws InterruptedException {
        if ("44".equals(v)) throw new RuntimeException("test");
        int n = rand.nextInt(10);
        Thread.sleep(n * 1000);
        return v + "-" + n;
    }
    public static void log(String msg) {
        System.out.println(msg);
    }
}

import java.util.Arrays;

import java.util.List;

import java.util.Random;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.stream.Collectors;

public class Hoge {

public static void main(String[] args) throws Exception {

log("start ...");

ExecutorService streamPool = Executors.newFixedThreadPool(10); //①

List<String> requests = Arrays.asList(new String[] {"11", "22", "33", "44", "55"}); //②

String[] array = new String[requests.size()]; //③

CountDownLatch latch = new CountDownLatch(requests.size()); //④

for (String val : requests) {

Integer x = i++; //⑤

streamPool.submit(() -> { //⑥

log("start..." + x);

array[x] = doHoge(val); //⑦

} catch (Exception e) {

log(e.toString());

latch.countDown();

log("end..." + x);

streamPool.shutdown();

List<String> results = Arrays.stream(array)

.filter(pi -> pi != null) //⑧

.collect(Collectors.toList()); //⑨

log(results.toString());

log("end ...");

static Random rand = new Random();

static String doHoge(String v) throws InterruptedException {

if ("44".equals(v)) throw new RuntimeException("test");

int n = rand.nextInt(10);

Thread.sleep(n * 1000);

return v + "-" + n;

public static void log(String msg) {

System.out.println(msg);