import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class HighConcurrencyDemo {
    private static final int TOTAL_REQUESTS = 10_000;
    private static final int MAX_CONCURRENCY = 500;
    private static final int QUEUE_CAPACITY = 1_000;

    record Result(boolean success, long latencyMillis) {
    }

    public static void main(String[] args) throws Exception {
        long started = System.nanoTime();
        AtomicInteger successCount = new AtomicInteger();
        AtomicInteger failedCount = new AtomicInteger();
        List<Long> latencies = Collections.synchronizedList(new ArrayList<>());

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                MAX_CONCURRENCY,
                MAX_CONCURRENCY,
                30,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        try {
            List<Future<Result>> futures = new ArrayList<>(TOTAL_REQUESTS);

            for (int i = 1; i <= TOTAL_REQUESTS; i++) {
                final int requestId = i;
                futures.add(executor.submit(() -> callDownstream(requestId)));
            }

            for (Future<Result> future : futures) {
                Result result = future.get();
                latencies.add(result.latencyMillis());
                if (result.success()) {
                    successCount.incrementAndGet();
                } else {
                    failedCount.incrementAndGet();
                }
            }
        } finally {
            executor.shutdown();
            executor.awaitTermination(1, TimeUnit.MINUTES);
        }

        double elapsedSeconds = Duration.ofNanos(System.nanoTime() - started).toMillis() / 1000.0;
        double avgLatency = latencies.stream().mapToLong(Long::longValue).average().orElse(0);
        long maxLatency = latencies.stream().mapToLong(Long::longValue).max().orElse(0);

        System.out.println("total requests: " + TOTAL_REQUESTS);
        System.out.println("max concurrency: " + MAX_CONCURRENCY);
        System.out.println("success: " + successCount.get() + ", failed: " + failedCount.get());
        System.out.printf("elapsed: %.2fs%n", elapsedSeconds);
        System.out.printf("throughput: %.2f req/s%n", TOTAL_REQUESTS / elapsedSeconds);
        System.out.printf("avg latency: %.2fms%n", avgLatency);
        System.out.println("max latency: " + maxLatency + "ms");
    }

    private static Result callDownstream(int requestId) throws InterruptedException {
        long started = System.nanoTime();

        // Simulate an I/O request such as HTTP, Redis, database, or message queue access.
        Thread.sleep(ThreadLocalRandom.current().nextInt(20, 121));

        boolean success = requestId % 211 != 0;
        long latencyMillis = Duration.ofNanos(System.nanoTime() - started).toMillis();
        return new Result(success, latencyMillis);
    }
}