三个消费者只放一个结束标记，为什么程序还不退出？

"""One stop item is consumed once; queue.join is not Thread.join.

All waiting has a bounded escape path for this finite demonstration.
"""
from queue import Empty, Queue
from threading import Barrier, Event, Lock, Thread


def main():
    jobs = Queue()
    stop = object()
    abort = Event()
    ready = Barrier(4, timeout=3)
    lock = Lock()
    successes, failures, stopped = [], [], []
    monitors = []
    checks = 0

    def check(condition, message):
        nonlocal checks
        if not condition:
            raise AssertionError(message)
        checks += 1

    def worker(index):
        ready.wait()
        while not abort.is_set():
            try:
                item = jobs.get(timeout=0.1)
            except Empty:
                continue
            try:
                if item is stop:
                    with lock:
                        stopped.append(index)
                    return
                try:
                    result = 10 // item
                except ZeroDivisionError:
                    with lock:
                        failures.append(item)
                else:
                    with lock:
                        successes.append((item, result))
            finally:
                # The stop item and a failed work item also need one task_done.
                jobs.task_done()

    def wait_for_queue():
        monitor = Thread(target=jobs.join, name="queue-join-monitor")
        monitors.append(monitor)
        monitor.start()
        monitor.join(timeout=3)
        if monitor.is_alive():
            raise AssertionError("queue.join did not finish within the guard")

    workers = [Thread(target=worker, args=(i,), name=f"worker-{i}") for i in range(3)]
    try:
        for thread in workers:
            thread.start()
        ready.wait()
        for item in (2, 0, 5):
            jobs.put(item)
        jobs.put(stop)
        wait_for_queue()
        with lock:
            stopped_copy = list(stopped)
        check(len(stopped_copy) == 1, "one stop item must be consumed by one worker")
        workers[stopped_copy[0]].join(timeout=3)
        alive_after_one = sum(thread.is_alive() for thread in workers)
        check(alive_after_one == 2, "queue.join does not stop the other workers")
        check(sorted(successes) == [(2, 5), (5, 2)], "successful work is recorded separately")
        check(failures == [0], "task_done is not a business-success certificate")
        print(f"after_one_stop: queue_join_returned=True; workers_alive={alive_after_one}")
        print(f"work_successes={sorted(successes)}; work_failures={failures}")

        for _ in range(alive_after_one):
            jobs.put(stop)
        wait_for_queue()
        for thread in workers:
            thread.join(timeout=3)
        check(not any(thread.is_alive() for thread in workers), "all workers must exit")
        check(sorted(stopped) == [0, 1, 2], "each worker consumes its own stop item")
        print("after_three_stops: queue_join_returned=True; workers_alive=0")
    finally:
        abort.set()
        ready.abort()
        for thread in workers:
            if thread.ident is not None:
                thread.join(timeout=3)
        # Release a join monitor even if an earlier assertion interrupted the run.
        while True:
            try:
                jobs.get_nowait()
            except Empty:
                break
            else:
                jobs.task_done()
        for monitor in monitors:
            monitor.join(timeout=3)
        if any(t.is_alive() for t in workers + monitors):
            raise AssertionError("demonstration thread did not stop")
    print(f"PASS checks={checks}; all_threads_stopped=True")


if __name__ == "__main__":
    main()

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/** Three fixed consumers; completion accounting is separate from BlockingQueue. */
public final class BlockingQueueStopDemo {
    private record Job(int value, boolean stop) {}
    private static final Job STOP = new Job(0, true);
    private static int checks;

    private static void check(boolean condition, String message) {
        if (!condition) throw new AssertionError(message);
        checks++;
    }

    private static void await(CountDownLatch latch, String description) throws InterruptedException {
        if (!latch.await(3, TimeUnit.SECONDS)) {
            throw new AssertionError(description + " did not finish within the guard");
        }
    }

    public static void main(String[] args) throws Exception {
        BlockingQueue<Job> jobs = new LinkedBlockingQueue<>();
        CountDownLatch ready = new CountDownLatch(3);
        CountDownLatch businessDone = new CountDownLatch(3);
        CountDownLatch firstStop = new CountDownLatch(1);
        AtomicBoolean abort = new AtomicBoolean();
        Map<Integer, Integer> successes = new ConcurrentHashMap<>();
        ConcurrentLinkedQueue<Integer> failures = new ConcurrentLinkedQueue<>();
        ConcurrentLinkedQueue<Integer> stopped = new ConcurrentLinkedQueue<>();
        ConcurrentLinkedQueue<String> workerErrors = new ConcurrentLinkedQueue<>();
        List<Thread> workers = new ArrayList<>();
        for (int index = 0; index < 3; index++) {
            final int id = index;
            workers.add(new Thread(() -> {
                ready.countDown();
                try {
                    while (!abort.get()) {
                        Job job = jobs.take();
                        if (job == STOP) {
                            stopped.add(id);
                            firstStop.countDown();
                            return;
                        }
                        try {
                            successes.put(job.value(), 10 / job.value());
                        } catch (ArithmeticException expected) {
                            failures.add(job.value());
                        } finally {
                            // This explicit latch belongs to our finite batch.
                            businessDone.countDown();
                        }
                    }
                } catch (InterruptedException interrupted) {
                    if (!abort.get()) workerErrors.add("unexpected worker interruption");
                    Thread.currentThread().interrupt();
                } catch (RuntimeException error) {
                    workerErrors.add(error.toString());
                }
            }, "queue-worker-" + id));
        }
        try {
            for (Thread worker : workers) worker.start();
            await(ready, "worker startup");
            for (int value : new int[] {2, 0, 5}) jobs.add(new Job(value, false));
            jobs.add(STOP);
            await(businessDone, "business batch");
            await(firstStop, "first stop item");
            check(stopped.size() == 1, "one stop item must stop exactly one consumer");
            workers.get(stopped.element()).join(3000);
            long aliveAfterOne = workers.stream().filter(Thread::isAlive).count();
            check(aliveAfterOne == 2, "business completion does not stop the other workers");
            check(businessDone.getCount() == 0 && jobs.isEmpty(),
                    "the finite batch is accounted for and no queue item remains");
            check(successes.equals(Map.of(2, 5, 5, 2)), "successful work must be recorded");
            check(List.copyOf(failures).equals(List.of(0)), "failed work must be recorded separately");
            System.out.println("after_one_stop: business_latch=0; workers_alive=" + aliveAfterOne);
            System.out.println("work_successes=" + new TreeMap<>(successes) + "; work_failures=" + failures);

            for (int i = 0; i < aliveAfterOne; i++) jobs.add(STOP);
            for (Thread worker : workers) worker.join(3000);
            check(workers.stream().noneMatch(Thread::isAlive), "all consumers must exit");
            check(stopped.stream().sorted().toList().equals(List.of(0, 1, 2)),
                    "each consumer must receive a stop item");
            check(workerErrors.isEmpty(), "no unexpected worker error may be hidden");
            System.out.println("after_three_stops: workers_alive=0; stopped_workers=[0, 1, 2]");
        } finally {
            abort.set(true);
            for (Thread worker : workers) worker.interrupt();
            for (Thread worker : workers) worker.join(3000);
            if (workers.stream().anyMatch(Thread::isAlive)) {
                throw new AssertionError("demonstration worker did not stop");
            }
            jobs.clear();
        }
        System.out.println("PASS checks=" + checks + "; all_threads_stopped=true");
    }
}