import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class Race {
    private final CyclicBarrier barrier;

    public Race(int numberOfRunners) {
        barrier = new CyclicBarrier(numberOfRunners, () -> {
            System.out.println("比賽開(kāi)始！");
            // 這里可以放置所有參與者到達(dá)屏障后要執(zhí)行的操作
        });
    }

    public void run() {
        System.out.println("等待其他參賽者...");
        try {
            barrier.await(); // 等待其他線程
            System.out.println("開(kāi)始跑步！");
            // 跑步時(shí)間
            Thread.sleep((long) (Math.random() * 10000));
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        final int numberOfRunners = 5;
        Race race = new Race(numberOfRunners);

        // 創(chuàng)建參賽者線程
        for (int i = 0; i < numberOfRunners; i++) {
            final int runnerNumber = i + 1;
            new Thread(() -> {
                System.out.println("參賽者 " + runnerNumber + " 已準(zhǔn)備就緒");
                race.run();
            }).start();
        }
    }
}

import java.util.concurrent.Callable;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class DataProcessingPipeline {
    private final ExecutorService executor = Executors.newFixedThreadPool(4);
    private final CyclicBarrier barrier;
    private final int numberOfPhases;
    private final int numberOfTasks;

    public DataProcessingPipeline(int numberOfTasks, int numberOfPhases) {
        this.numberOfTasks = numberOfTasks;
        this.numberOfPhases = numberOfPhases;
        this.barrier = new CyclicBarrier(numberOfTasks, () -> {
            System.out.println("一個(gè)階段完成，準(zhǔn)備進(jìn)入下一階段");
        });
    }

    public void processData() throws Exception {
        for (int phase = 1; phase <= numberOfPhases; phase++) {
            System.out.println("階段 " + phase + " 開(kāi)始");
            for (int task = 0; task < numberOfTasks; task++) {
                final int currentTask = task;
                executor.submit(() -> {
                    try {
                        // 數(shù)據(jù)處理任務(wù)
                        System.out.println("任務(wù) " + currentTask + " 在階段 " + phase + " 執(zhí)行");
                        Thread.sleep((long) (Math.random() * 5000));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        try {
                            barrier.await();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
            barrier.await(); // 等待所有任務(wù)完成
        }
        executor.shutdown();
    }

    public static void main(String[] args) {
        DataProcessingPipeline pipeline = new DataProcessingPipeline(4, 3);
        try {
            pipeline.processData();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class Counter {
    // 使用 AtomicInteger 來(lái)確保計(jì)數(shù)器的線程安全
    private final AtomicInteger count = new AtomicInteger(0);

    // 提供一個(gè)方法來(lái)增加計(jì)數(shù)器的值
    public void increment() {
        // 原子地增加計(jì)數(shù)器的值
        count.incrementAndGet();
    }

    // 提供一個(gè)方法來(lái)獲取當(dāng)前計(jì)數(shù)器的值
    public int getCount() {
        // 原子地獲取計(jì)數(shù)器的值
        return count.get();
    }
}

public class DataStore {
    // 使用 AtomicLong 來(lái)統(tǒng)計(jì)數(shù)據(jù)總量
    private final AtomicLong dataCount = new AtomicLong(0);

    public void addData(long size) {
        // 原子地將數(shù)據(jù)大小累加到總量
        dataCount.addAndGet(size);
    }

    public long getDataCount() {
        // 原子地獲取當(dāng)前數(shù)據(jù)總量
        return dataCount.get();
    }
}

// 測(cè)試類
public class AtomicVariablesDemo {
    public static void main(String[] args) {
        Counter counter = new Counter();
        DataStore dataStore = new DataStore();

        // 多線程環(huán)境中對(duì)計(jì)數(shù)器和數(shù)據(jù)總量的更新
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                counter.increment();
                dataStore.addData(100);  // 假設(shè)每次操作增加100單位數(shù)據(jù)
            }).start();
        }

        // 等待所有線程完成
        while (Thread.activeCount() > 1) {
            Thread.yield();
        }

        // 輸出計(jì)數(shù)器的值和數(shù)據(jù)總量
        System.out.println("Counter value: " + counter.getCount());
        System.out.println("Data store size: " + dataStore.getDataCount());
    }
}

import java.util.concurrent.atomic.AtomicInteger;

public class MessageNotificationCounter {
    private final AtomicInteger messageCount = new AtomicInteger(0);

    // 接收新消息時(shí)調(diào)用此方法
    public void receiveMessage() {
        // 原子地增加消息計(jì)數(shù)
        messageCount.incrementAndGet();
        System.out.println("New message received. Total messages: " + messageCount.get());
    }

    // 用戶閱讀消息時(shí)調(diào)用此方法
    public void messagesRead() {
        // 原子地減少消息計(jì)數(shù)
        messageCount.decrementAndGet();
        System.out.println("Messages read. Remaining messages: " + messageCount.get());
    }

    // 獲取當(dāng)前消息計(jì)數(shù)
    public int getMessageCount() {
        return messageCount.get();
    }
}

// 測(cè)試類
public class AtomicVariablesDemo {
    public static void main(String[] args) {
        MessageNotificationCounter counter = new MessageNotificationCounter();

        // 多個(gè)用戶同時(shí)發(fā)送消息
        Thread sender1 = new Thread(() -> {
            counter.receiveMessage();
        });
        Thread sender2 = new Thread(() -> {
            counter.receiveMessage();
        });

        // 用戶閱讀消息
        Thread reader = new Thread(() -> {
            counter.messagesRead();
        });

        sender1.start();
        sender2.start();
        reader.start();

        sender1.join();
        sender2.join();
        reader.join();

        System.out.println("Final message count: " + counter.getMessageCount());
    }
}

import java.util.concurrent.ConcurrentHashMap;
import java.util.Map;

public class ConcurrentHashMapDemo {
    // 創(chuàng)建一個(gè) ConcurrentHashMap 實(shí)例
    private final ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

    // 將一個(gè)鍵值對(duì)插入到 Map 中
    public void put(String key, Integer value) {
        // put 方法是線程安全的
        map.put(key, value);
    }

    // 從 Map 中獲取與指定鍵關(guān)聯(lián)的值
    public Integer get(String key) {
        // get 方法是線程安全的
        return map.get(key);
    }

    // 計(jì)算 Map 中的元素?cái)?shù)量
    public int size() {
        // size 方法是線程安全的
        return map.size();
    }

    // 演示刪除操作
    public void remove(String key) {
        // remove 方法是線程安全的
        map.remove(key);
    }

    // 演示如何批量添加數(shù)據(jù)
    public void addAll(Map<String, Integer> newData) {
        // putAll 方法是線程安全的
        map.putAll(newData);
    }

    public static void main(String[] args) {
        ConcurrentHashMapDemo demo = new ConcurrentHashMapDemo();

        // 批量添加數(shù)據(jù)
        demo.addAll(Map.of("key1", 1, "key2", 2, "key3", 3));

        // 單獨(dú)添加一條數(shù)據(jù)
        demo.put("key4", 4);

        // 獲取并打印一條數(shù)據(jù)
        System.out.println("Value for 'key1': " + demo.get("key1"));

        // 獲取 Map 的大小
        System.out.println("Map size: " + demo.size());

        // 刪除一條數(shù)據(jù)
        demo.remove("key2");

        // 再次獲取 Map 的大小
        System.out.println("Map size after removal: " + demo.size());
    }
}

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

public class DistributedCache<K, V> {
    private final ConcurrentHashMap<K, V> cacheMap = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<K, Long> expirationMap = new ConcurrentHashMap<>();

    public void put(K key, V value, long ttl) {
        cacheMap.put(key, value);
        expirationMap.put(key, System.currentTimeMillis() + ttl);
        scheduleEviction(key, ttl);
    }

    public V get(K key) {
        Long expirationTime = expirationMap.get(key);
        if (expirationTime == null || expirationTime < System.currentTimeMillis()) {
            cacheMap.remove(key);
            expirationMap.remove(key);
            return null;
        }
        return cacheMap.get(key);
    }

    private void scheduleEviction(final K key, final long ttl) {
        new Thread(() -> {
            try {
                TimeUnit.MILLISECONDS.sleep(ttl);
                cacheMap.computeIfPresent(key, (k, v) -> null);
                expirationMap.remove(key);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }

    public static void main(String[] args) {
        DistributedCache<String, String> cache = new DistributedCache<>();
        cache.put("userSession", "sessionData", 5000); // 緩存設(shè)置5秒過(guò)期

        // 多個(gè)線程并發(fā)訪問(wèn)緩存
        for (int i = 0; i < 100; i++) {
            int finalI = i;
            new Thread(() -> {
                String result = cache.get("userSession");
                System.out.println("Thread " + finalI + " retrieved: " + result);
            }).start();
        }
    }
}

import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;

public class ConcurrentSkipListMapDemo {
    // 創(chuàng)建一個(gè) ConcurrentSkipListMap 實(shí)例
    private final ConcurrentSkipListMap<Integer, String> map = new ConcurrentSkipListMap<>();

    // 將一個(gè)鍵值對(duì)插入到 Map 中
    public void put(Integer key, String value) {
        // put 方法是線程安全的
        map.put(key, value);
    }

    // 從 Map 中獲取與指定鍵關(guān)聯(lián)的值
    public String get(Integer key) {
        // get 方法是線程安全的
        return map.get(key);
    }

    // 獲取 Map 的鍵集合
    public java.util.NavigableSet<Integer> keySet() {
        // keySet 方法返回 Map 的鍵集合視圖
        return map.keySet();
    }

    // 獲取 Map 的值集合
    public java.util.Collection<String> values() {
        // values 方法返回 Map 的值集合視圖
        return map.values();
    }

    // 獲取 Map 的大小
    public int size() {
        // size 方法是線程安全的
        return map.size();
    }

    // 演示刪除操作
    public void remove(Integer key) {
        // remove 方法是線程安全的
        map.remove(key);
    }

    public static void main(String[] args) {
        ConcurrentSkipListMapDemo demo = new ConcurrentSkipListMapDemo();

        // 插入一些數(shù)據(jù)
        demo.put(1, "One");
        demo.put(2, "Two");
        demo.put(3, "Three");

        // 獲取并打印一條數(shù)據(jù)
        System.out.println("Value for key 2: " + demo.get(2));

        // 獲取 Map 的大小
        System.out.println("Map size: " + demo.size());

        // 獲取并打印所有鍵
        System.out.println("All keys: " + demo.keySet());

        // 刪除一條數(shù)據(jù)
        demo.remove(2);

        // 再次獲取 Map 的大小
        System.out.println("Map size after removal: " + demo.size());

        // 獲取并打印所有值
        System.out.println("All values: " + demo.values());
    }
}

import java.util.concurrent.ConcurrentSkipListMap;

public class StockPriceIndex {
    private final ConcurrentSkipListMap<String, Double> priceIndex = new ConcurrentSkipListMap<>();

    public void updatePrice(String stockSymbol, Double newPrice) {
        // 更新股票價(jià)格
        priceIndex.put(stockSymbol, newPrice);
    }

    public Double getPrice(String stockSymbol) {
        // 獲取股票價(jià)格
        return priceIndex.get(stockSymbol);
    }

    public void removeStock(String stockSymbol) {
        // 移除股票信息
        priceIndex.remove(stockSymbol);
    }

    public ConcurrentSkipListMap<String, Double> headMap(String toKey) {
        // 獲取指定范圍內(nèi)的股票價(jià)格索引
        return priceIndex.headMap(toKey);
    }

    public static void main(String[] args) {
        StockPriceIndex index = new StockPriceIndex();
        index.updatePrice("AAPL", 150.00);
        index.updatePrice("GOOGL", 2750.50);
        index.updatePrice("MSFT", 250.00);

        System.out.println("Price of AAPL: " + index.getPrice("AAPL"));
        System.out.println("Price of GOOGL: " + index.getPrice("GOOGL"));

        // 獲取所有小于 "MSFT" 的股票價(jià)格索引
        ConcurrentSkipListMap<String, Double> subMap = index.headMap("MSFT");
        subMap.forEach((k, v) -> System.out.println("Stock: " + k + ", Price: " + v));
    }
}

import java.util.concurrent.ConcurrentLinkedQueue;

public class ConcurrentLinkedQueueDemo {
    // 創(chuàng)建一個(gè) ConcurrentLinkedQueue 實(shí)例
    private final ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();

    // 向隊(duì)列中添加元素
    public void add(int number) {
        // add 方法是線程安全的
        queue.add(number);
        System.out.println("Added " + number);
    }

    // 從隊(duì)列中獲取并移除元素
    public Integer poll() {
        // poll 方法是線程安全的，返回并移除隊(duì)列頭部的元素
        Integer result = queue.poll();
        if (result != null) {
            System.out.println("Polled " + result);
        } else {
            System.out.println("Queue is empty");
        }
        return result;
    }

    // 查看隊(duì)列頭部的元素但不移除
    public Integer peek() {
        // peek 方法是線程安全的，返回隊(duì)列頭部的元素但不移除
        Integer result = queue.peek();
        if (result != null) {
            System.out.println("Peeked " + result);
        } else {
            System.out.println("Queue is empty");
        }
        return result;
    }

    // 獲取隊(duì)列的大小
    public int size() {
        // size 方法估算隊(duì)列的大小
        int size = queue.size();
        System.out.println("Queue size: " + size);
        return size;
    }

    public static void main(String[] args) {
        ConcurrentLinkedQueueDemo demo = new ConcurrentLinkedQueueDemo();

        // 啟動(dòng)生產(chǎn)者線程
        Thread producerThread = new Thread(() -> {
            demo.add(1);
            demo.add(2);
            demo.add(3);
        });

        // 啟動(dòng)消費(fèi)者線程
        Thread consumerThread = new Thread(() -> {
            demo.poll();
            demo.poll();
            demo.poll();
            demo.poll(); // 這次調(diào)用應(yīng)該會(huì)返回 null，因?yàn)殛?duì)列已空
        });

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

        producerThread.join();
        consumerThread.join();

        // 在所有線程完成后獲取隊(duì)列大小
        demo.size();
    }
}

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class LogProcessor {
    private final ConcurrentLinkedQueue<String> logQueue = new ConcurrentLinkedQueue<>();
    private final ExecutorService processorPool = Executors.newFixedThreadPool(10);

    public void log(String message) {
        // 生產(chǎn)者線程調(diào)用此方法來(lái)添加日志到隊(duì)列
        logQueue.add(message);
    }

    public void startLogProcessing() {
        // 消費(fèi)者線程池，用于處理隊(duì)列中的日志
        processorPool.submit(() -> {
            while (true) {
                try {
                    // 消費(fèi)者線程調(diào)用此方法來(lái)處理隊(duì)列中的日志
                    String logEntry = logQueue.poll();
                    if (logEntry != null) {
                        processLog(logEntry);
                    } else {
                        TimeUnit.MILLISECONDS.sleep(100); // 避免 CPU 過(guò)載
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });
    }

    private void processLog(String logEntry) {
        // 實(shí)際處理日志的邏輯
        System.out.println("Processing log: " + logEntry);
    }

    public static void main(String[] args) {
        LogProcessor logProcessor = new LogProcessor();
        logProcessor.startLogProcessing();

        // 多個(gè)生產(chǎn)者線程生成日志
        for (int i = 0; i < 100; i++) {
            int finalI = i;
            new Thread(() -> {
                logProcessor.log("Log entry " + finalI);
            }).start();
        }
    }
}

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

public class BlockingQueueDemo {
    // 創(chuàng)建一個(gè) LinkedBlockingQueue 實(shí)例，容量限制為10
    private final BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>(10);

    // 向 BlockingQueue 中添加元素
    public void produce(Integer element) throws InterruptedException {
        // put 方法在隊(duì)列滿時(shí)阻塞，直到隊(duì)列中有空間
        blockingQueue.put(element);
        System.out.println("Produced: " + element);
    }

    // 從 BlockingQueue 中獲取元素
    public Integer consume() throws InterruptedException {
        // take 方法在隊(duì)列空時(shí)阻塞，直到隊(duì)列中有元素
        Integer element = blockingQueue.take();
        System.out.println("Consumed: " + element);
        return element;
    }

    // 獲取 BlockingQueue 的大小
    public int size() {
        // size 方法返回隊(duì)列當(dāng)前的元素?cái)?shù)量
        return blockingQueue.size();
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueueDemo demo = new BlockingQueueDemo();

        // 創(chuàng)建生產(chǎn)者線程
        Thread producerThread = new Thread(() -> {
            try {
                for (int i = 0; i < 15; i++) {
                    demo.produce(i);
                    Thread.sleep(100); // 生產(chǎn)延時(shí)
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        // 創(chuàng)建消費(fèi)者線程
        Thread consumerThread = new Thread(() -> {
            try {
                for (int i = 0; i < 15; i++) {
                    int element = demo.consume();
                    Thread.sleep(150); // 消費(fèi)延時(shí)
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

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

        producerThread.join();
        consumerThread.join();

        // 打印最終隊(duì)列的大小
        System.out.println("Final queue size: " + demo.size());
    }
}

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

public class MessageQueueSystem {
    private final BlockingQueue<Message> messageQueue = new LinkedBlockingQueue<>();

    public void produceMessage(String content) throws InterruptedException {
        // 將消息添加到隊(duì)列中，如果隊(duì)列滿了，生產(chǎn)者線程將被阻塞
        messageQueue.put(new Message(content));
        System.out.println("Message produced: " + content);
    }

    public Message consumeMessage() throws InterruptedException {
        // 從隊(duì)列中取出消息，如果隊(duì)列空了，消費(fèi)者線程將被阻塞
        Message message = messageQueue.take();
        System.out.println("Message consumed: " + message.getContent());
        return message;
    }

    public static void main(String[] args) throws InterruptedException {
        MessageQueueSystem messageQueueSystem = new MessageQueueSystem();

        // 創(chuàng)建生產(chǎn)者線程
        Thread producerThread = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    messageQueueSystem.produceMessage("Message " + i);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 創(chuàng)建消費(fèi)者線程
        Thread consumerThread = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    messageQueueSystem.consumeMessage();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

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

        producerThread.join();
        consumerThread.join();
    }
}

class Message {
    private final String content;

    public Message(String content) {
        this.content = content;
    }

    public String getContent() {
        return content;
    }
}

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class BoundedBuffer {
    private final Object[] buffer;
    private int putPtr, takePtr, count;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    public BoundedBuffer(int size) {
        buffer = new Object[size];
    }

    public void put(Object x) throws InterruptedException {
        lock.lock();
        try {
            while (count == buffer.length) { // 等待直到緩沖區(qū)非滿
                notFull.await();
            }
            buffer[putPtr] = x;
            putPtr = (putPtr + 1) % buffer.length;
            count++;
            notEmpty.signal(); // 通知可能等待的消費(fèi)者
        } finally {
            lock.unlock();
        }
    }

    public Object take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) { // 等待直到緩沖區(qū)非空
                notEmpty.await();
            }
            Object x = buffer[takePtr];
            takePtr = (takePtr + 1) % buffer.length;
            count--;
            notFull.signal(); // 通知可能等待的生產(chǎn)者
            return x;
        } finally {
            lock.unlock();
        }
    }
}

public class ProducerConsumerDemo {
    private final BoundedBuffer buffer;

    public ProducerConsumerDemo(int size) {
        buffer = new BoundedBuffer(size);
    }

    public void produce(String item) {
        buffer.put(item);
    }

    public String consume() {
        return (String) buffer.take();
    }

    public static void main(String[] args) throws InterruptedException {
        final int SIZE = 10;
        final ProducerConsumerDemo demo = new ProducerConsumerDemo(SIZE);

        // 生產(chǎn)者線程
        Thread producerThread = new Thread(() -> {
            for (int i = 0; i < 20; i++) {
                demo.produce("Item " + i);
                try {
                    Thread.sleep(100); // 生產(chǎn)延時(shí)
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        // 消費(fèi)者線程
        Thread consumerThread = new Thread(() -> {
            for (int i = 0; i < 20; i++) {
                String item = demo.consume();
                System.out.println("Consumed: " + item);
                try {
                    Thread.sleep(150); // 消費(fèi)延時(shí)
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

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

        producerThread.join();
        consumerThread.join();
    }
}

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.Date;
import java.util.LinkedList;
import java.util.Queue;

public class TaskScheduler {
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition taskAvailable = lock.newCondition();
    private final Queue<Runnable> tasks = new LinkedList<>();

    public void schedule(Runnable task, long delay) {
        lock.lock();
        try {
            tasks.add(() -> {
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                task.run();
            });
            taskAvailable.signal(); // 通知調(diào)度器有新任務(wù)
        } finally {
            lock.unlock();
        }
    }

    public void startScheduling() {
        new Thread(this::runScheduler).start();
    }

    private void runScheduler() {
        lock.lock();
        try {
            while (true) {
                while (tasks.isEmpty()) { // 如果沒(méi)有任務(wù)，等待
                    taskAvailable.await();
                }
                Runnable task = tasks.poll();
                task.run();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        TaskScheduler scheduler = new TaskScheduler();
        scheduler.schedule(() -> System.out.println("Task 1 executed at " + new Date()), 2000);
        scheduler.schedule(() -> System.out.println("Task 2 executed at " + new Date()), 4000);
        scheduler.startScheduling();
    }
}