BlockingQueue 연산 정책

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;

public final class CheckThenAddRace {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);
        CountDownLatch checked = new CountDownLatch(2);
        CountDownLatch go = new CountDownLatch(1);
        Runnable producer = () -> {
            if (queue.remainingCapacity() > 0) {
                checked.countDown();
                try {
                    go.await();
                    queue.add(Thread.currentThread().getName());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        };
        Thread a = Thread.ofPlatform().name("A").start(producer);
        Thread b = Thread.ofPlatform().name("B").start(producer);
        checked.await();
        go.countDown();
        a.join();
        b.join();
        System.out.println("queue=" + queue);
    }
}

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public final class DeadlineBatchSubmitter<E> {
    public record Result<E>(int accepted, List<E> rejected) {}

    private final BlockingQueue<E> queue;

    public DeadlineBatchSubmitter(BlockingQueue<E> queue) {
        this.queue = queue;
    }

    public Result<E> submitAll(List<E> items, Duration totalBudget) throws InterruptedException {
        long deadline = System.nanoTime() + totalBudget.toNanos();
        int accepted = 0;
        List<E> rejected = new ArrayList<>();
        for (int index = 0; index < items.size(); index++) {
            E item = items.get(index);
            long remaining = deadline - System.nanoTime();
            if (remaining <= 0 || !queue.offer(item, remaining, TimeUnit.NANOSECONDS)) {
                rejected.addAll(items.subList(index, items.size()));
                break;
            }
            accepted++;
        }
        return new Result<>(accepted, List.copyOf(rejected));
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new java.util.concurrent.ArrayBlockingQueue<>(2);
        DeadlineBatchSubmitter<String> submitter = new DeadlineBatchSubmitter<>(queue);
        Result<String> result = submitter.submitAll(
                List.of("one", "two", "three"), Duration.ofMillis(2));
        System.out.println("accepted=" + result.accepted());
        System.out.println("rejected=" + result.rejected());
    }
}

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public final class TimedBatchDrainer<E> {
    private final BlockingQueue<E> queue;
    private final int maxBatch;

    TimedBatchDrainer(BlockingQueue<E> queue, int maxBatch) {
        if (maxBatch < 1) throw new IllegalArgumentException("maxBatch");
        this.queue = queue;
        this.maxBatch = maxBatch;
    }

    List<E> nextBatch(Duration waitForFirst) throws InterruptedException {
        E first = queue.poll(waitForFirst.toNanos(), TimeUnit.NANOSECONDS);
        if (first == null) return List.of();
        List<E> batch = new ArrayList<>(maxBatch);
        batch.add(first);
        queue.drainTo(batch, maxBatch - 1);
        return List.copyOf(batch);
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);
        queue.addAll(List.of(1, 2, 3, 4, 5));
        TimedBatchDrainer<Integer> drainer = new TimedBatchDrainer<>(queue, 3);
        System.out.println("batch=" + drainer.nextBatch(Duration.ofSeconds(1)));
        System.out.println("remaining=" + queue.size());
    }
}