Concurrency Araçları — Executor, Lock, Atomic ve Deadlock

1
// Incoming EFT messages arrive as a file; each line credits one beneficiary account.
2
public class IncomingEftPipeline {
3

4
    private static final String POISON = "__END__";
5
    private final BlockingQueue<String> queue = new ArrayBlockingQueue<>(1_000);   // bounded: backpressure
6

7
    public int run(Path csv, Consumer<String> sink) throws Exception {
8
        var imported = new AtomicInteger();
9
        try (var executor = Executors.newFixedThreadPool(2)) {
10
            executor.submit(() -> {                              // producer
11
                try (var lines = Files.lines(csv)) {
12
                    for (String line : (Iterable<String>) lines::iterator) {
13
                        queue.put(line);                         // blocks when the consumer falls behind
14
                    }
15
                } finally {
16
                    queue.put(POISON);                           // always tell the consumer we are done
17
                }
18
                return null;
19
            });
20
            executor.submit(() -> {                              // consumer
21
                for (String line = queue.take(); !line.equals(POISON); line = queue.take()) {
22
                    sink.accept(line);
23
                    imported.incrementAndGet();
24
                }
25
                return null;
26
            });
27
        }   // close(): no new tasks, then waits for both to finish
28
        return imported.get();
29
    }
30
}

1
public class StartupGate {
2

3
    private final CountDownLatch ready;
4

5
    public StartupGate(int dependencies) {
6
        this.ready = new CountDownLatch(dependencies);
7
    }
8

9
    public void dependencyReady(String name) {
10
        System.out.println("ready: " + name);
11
        ready.countDown();
12
    }
13

14
    // Returns false instead of hanging forever if a dependency never shows up.
15
    public boolean awaitAll(Duration timeout) throws InterruptedException {
16
        return ready.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
17
    }
18
}

1
// "Your card statement is ready" to every customer whose statement was cut today.
2
public class BulkSmsSender {
3

4
    private final Semaphore inFlight = new Semaphore(10);   // the SMS provider allows 10 parallel calls
5
    private final SmsClient client;
6

7
    public BulkSmsSender(SmsClient client) {
8
        this.client = client;
9
    }
10

11
    public void sendAll(List<Sms> messages) {
12
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
13
            for (Sms sms : messages) {
14
                executor.submit(() -> {
15
                    inFlight.acquire();              // waits cheaply on a virtual thread
16
                    try {
17
                        client.send(sms);
18
                    } finally {
19
                        inFlight.release();          // always, even when send throws
20
                    }
21
                    return null;
22
                });
23
            }
24
        }
25
    }
26
}

1
public class Main {
2

3
    public static void main(String[] args) throws Exception {
4
        var gate = new StartupGate(2);
5
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
6
            executor.submit(() -> { Thread.sleep(200); gate.dependencyReady("core-banking"); return null; });
7
            executor.submit(() -> { Thread.sleep(300); gate.dependencyReady("hsm"); return null; });   // PIN/crypto keys
8

9
            if (!gate.awaitAll(Duration.ofSeconds(5))) {
10
                System.out.println("dependencies not ready, refusing traffic");
11
                return;
12
            }
13
        }
14
        System.out.println("accepting traffic");
15

16
        int count = new IncomingEftPipeline().run(Path.of("incoming-eft.csv"), line -> { /* credit the account */ });
17
        System.out.println("credited " + count + " incoming transfers");
18
    }
19
}