Java多线程安全共享变量与周期性输出实践指南

class UnsafeCounter {
    int counter = 0;

    public void increment() {
        counter++;
    }

    public int getCounter() {
        return counter;
    }
}

public class UnsafeSharedVariableDemo {
    public static void main(String[] args) throws InterruptedException {
        UnsafeCounter unsafeCounter = new UnsafeCounter();

        // 线程1：递增计数器
        Thread incrementer = new Thread(() -> {
            for (int i = 0; i < 10000; i++) {
                unsafeCounter.increment();
            }
            System.out.println("Incrementer finished. Final count (from incrementer's perspective): " + unsafeCounter.getCounter());
        });

        // 线程2：周期性打印计数器
        Thread printer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    System.out.println("Printer read counter: " + unsafeCounter.getCounter());
                    Thread.sleep(100); // 每100毫秒打印一次
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        incrementer.start();
        printer.start();

        incrementer.join(); // 等待递增线程结束
        printer.join();     // 等待打印线程结束
        System.out.println("Main thread: Final counter value: " + unsafeCounter.getCounter());
    }
}

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicCounterDemo {
    private final AtomicInteger counter = new AtomicInteger(0); // 使用AtomicInteger作为共享计数器

    public void startThreads() {
        // 线程1：递增计数器
        Thread incrementerThread = new Thread(() -> {
            try {
                System.out.println("Incrementer thread started.");
                for (int i = 0; i < 50; i++) {
                    counter.incrementAndGet(); // 原子性递增
                    Thread.sleep(50); // 模拟工作负载
                }
                System.out.println("Incrementer thread finished. Final count: " + counter.get());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println("Incrementer thread interrupted.");
            }
        }, "Incrementer");

        // 线程2：周期性打印计数器
        Thread printerThread = new Thread(() -> {
            try {
                System.out.println("Printer thread started.");
                while (!Thread.currentThread().isInterrupted()) {
                    System.out.println("Current counter value: " + counter.get()); // 原子性获取
                    Thread.sleep(200); // 每200毫秒打印一次
                }
            } catch (InterruptedException e) {
                System.out.println("Printer thread interrupted. Exiting loop.");
            } finally {
                System.out.println("Printer thread finished.");
            }
        }, "Printer");

        incrementerThread.start();
        printerThread.start();

        // 等待递增线程完成
        try {
            incrementerThread.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        // 递增线程完成后，中断打印线程
        printerThread.interrupt();
        try {
            printerThread.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        System.out.println("All threads finished. Final counter value from main: " + counter.get());
    }

    public static void main(String[] args) {
        new AtomicCounterDemo().startThreads();
    }
}

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class MessageQueueCounterDemo {
    private final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10); // 容量为10的阻塞队列

    public void startThreads() {
        // 线程1：生产者 - 递增并放入队列
        Thread producerThread = new Thread(() -> {
            try {
                System.out.println("Producer thread started.");
                for (int i = 1; i <= 50; i++) {
                    queue.put(i); // 将当前计数值放入队列，如果队列满则阻塞
                    System.out.println("Producer added: " + i + ", Queue size: " + queue.size());
                    Thread.sleep(50); // 模拟生产数据的耗时
                }
                System.out.println("Producer finished adding 50 items. Signaling consumer to stop.");
                queue.put(-1); // 发送一个特殊值作为结束信号
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println("Producer thread interrupted.");
            } finally {
                System.out.println("Producer thread finished.");
            }
        }, "Producer");

        // 线程2：消费者 - 周期性从队列取出并打印
        Thread consumerThread = new Thread(() -> {
            try {
                System.out.println("Consumer thread started.");
                while (true) {
                    Integer value = queue.take(); // 从队列取出数据，如果队列空则阻塞
                    if (value == -1) { // 收到结束信号
                        System.out.println("Consumer received termination signal. Exiting.");
                        break;
                    }
                    System.out.println("Consumer retrieved: " + value + ", Queue size: " + queue.size());
                    Thread.sleep(200); // 模拟消费数据的耗时
                }
            } catch (InterruptedException e) {
                System.out.println("Consumer thread interrupted. Exiting loop.");
            } finally {
                System.out.println("Consumer thread finished.");
            }
        }, "Consumer");

        producerThread.start();
        consumerThread.start();

        // 等待生产者和消费者线程完成
        try {
            producerThread.join();
            consumerThread.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        System.out.println("All threads finished. Final queue size: " + queue.size());
    }

    public static void main(String[] args) {
        new MessageQueueCounterDemo().startThreads();
    }
}