Exchanger در جاوا — به زبان ساده

1@Test
2public void givenThreads_whenMessageExchanged_thenCorrect() {
3    Exchanger<String> exchanger = new Exchanger<>();
4 
5    Runnable taskA = () -> {
6        try {
7            String message = exchanger.exchange("from A");
8            assertEquals("from B", message);
9        } catch (InterruptedException e) {
10            Thread.currentThread.interrupt();
11            throw new RuntimeException(e);
12        }
13    };
14 
15    Runnable taskB = () -> {
16        try {
17            String message = exchanger.exchange("from B");
18            assertEquals("from A", message);
19        } catch (InterruptedException e) {
20            Thread.currentThread.interrupt();
21            throw new RuntimeException(e);
22        }
23    };
24    CompletableFuture.allOf(
25      runAsync(taskA), runAsync(taskB)).join();
26}

1@Test
2public void givenThread_WhenExchangedMessage_thenCorrect() throws InterruptedException {
3    Exchanger<String> exchanger = new Exchanger<>();
4 
5    Runnable runner = () -> {
6        try {
7            String message = exchanger.exchange("from runner");
8            assertEquals("to runner", message);
9        } catch (InterruptedException e) {
10            Thread.currentThread.interrupt();
11            throw new RuntimeException(e);
12        }
13    };
14    CompletableFuture<Void> result 
15      = CompletableFuture.runAsync(runner);
16    String msg = exchanger.exchange("to runner");
17    assertEquals("from runner", msg);
18    result.join();
19}

1@Test
2public void givenData_whenPassedThrough_thenCorrect() throws InterruptedException {
3 
4    Exchanger<Queue<String>> readerExchanger = new Exchanger<>();
5    Exchanger<Queue<String>> writerExchanger = new Exchanger<>();
6 
7    Runnable reader = () -> {
8        Queue<String> readerBuffer = new ConcurrentLinkedQueue<>();
9        while (true) {
10            readerBuffer.add(UUID.randomUUID().toString());
11            if (readerBuffer.size() >= BUFFER_SIZE) {
12                readerBuffer = readerExchanger.exchange(readerBuffer);
13            }
14        }
15    };
16 
17    Runnable processor = () -> {
18        Queue<String> processorBuffer = new ConcurrentLinkedQueue<>();
19        Queue<String> writerBuffer = new ConcurrentLinkedQueue<>();
20        processorBuffer = readerExchanger.exchange(processorBuffer);
21        while (true) {
22            writerBuffer.add(processorBuffer.poll());
23            if (processorBuffer.isEmpty()) {
24                processorBuffer = readerExchanger.exchange(processorBuffer);
25                writerBuffer = writerExchanger.exchange(writerBuffer);
26            }
27        }
28    };
29 
30    Runnable writer = () -> {
31        Queue<String> writerBuffer = new ConcurrentLinkedQueue<>();
32        writerBuffer = writerExchanger.exchange(writerBuffer);
33        while (true) {
34            System.out.println(writerBuffer.poll());
35            if (writerBuffer.isEmpty()) {
36                writerBuffer = writerExchanger.exchange(writerBuffer);
37            }
38        }
39    };
40    CompletableFuture.allOf(
41      runAsync(reader), 
42      runAsync(processor),
43      runAsync(writer)).join();
44}