Java线程通信 #

public class Message {
    private String content;
    private boolean empty = true;
    
    public synchronized String take() throws InterruptedException {
        while (empty) {
            wait();  // 等待消息
        }
        empty = true;
        notifyAll();  // 通知生产者
        return content;
    }
    
    public synchronized void put(String content) throws InterruptedException {
        while (!empty) {
            wait();  // 等待消费
        }
        empty = false;
        this.content = content;
        notifyAll();  // 通知消费者
    }
}

public class Buffer {
    private final Queue<Integer> queue = new LinkedList<>();
    private final int capacity = 10;
    
    public synchronized void produce(int value) throws InterruptedException {
        while (queue.size() == capacity) {
            wait();
        }
        queue.offer(value);
        System.out.println("生产: " + value);
        notifyAll();
    }
    
    public synchronized int consume() throws InterruptedException {
        while (queue.isEmpty()) {
            wait();
        }
        int value = queue.poll();
        System.out.println("消费: " + value);
        notifyAll();
        return value;
    }
}

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

BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);

// 生产者
Thread producer = new Thread(() -> {
    try {
        for (int i = 0; i < 100; i++) {
            queue.put(i);  // 队列满时阻塞
            System.out.println("生产: " + i);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});

// 消费者
Thread consumer = new Thread(() -> {
    try {
        while (true) {
            int value = queue.take();  // 队列空时阻塞
            System.out.println("消费: " + value);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});

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

public class Buffer {
    private final Lock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    
    private final Queue<Integer> queue = new LinkedList<>();
    private final int capacity = 10;
    
    public void produce(int value) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                notFull.await();  // 等待不满
            }
            queue.offer(value);
            notEmpty.signal();  // 通知不空
        } finally {
            lock.unlock();
        }
    }
    
    public int consume() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                notEmpty.await();  // 等待不空
            }
            int value = queue.poll();
            notFull.signal();  // 通知不满
            return value;
        } finally {
            lock.unlock();
        }
    }
}

Lock lock = new ReentrantLock();
Condition conditionA = lock.newCondition();
Condition conditionB = lock.newCondition();

// 线程A等待
lock.lock();
try {
    conditionA.await();
} finally {
    lock.unlock();
}

// 线程B唤醒A
lock.lock();
try {
    conditionA.signal();
} finally {
    lock.unlock();
}

// 无界队列
BlockingQueue<String> unbounded = new LinkedBlockingQueue<>();

// 有界队列
BlockingQueue<String> bounded = new LinkedBlockingQueue<>(100);

// 数组队列
BlockingQueue<String> arrayQueue = new ArrayBlockingQueue<>(100);

// 优先级队列
BlockingQueue<String> priorityQueue = new PriorityBlockingQueue<>();

// 延迟队列
BlockingQueue<Delayed> delayedQueue = new DelayQueue<>();

// 同步队列
BlockingQueue<String> synchronousQueue = new SynchronousQueue<>();

import java.util.concurrent.CountDownLatch;

CountDownLatch latch = new CountDownLatch(3);

// 工作线程
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        try {
            Thread.sleep(1000);
            System.out.println("任务完成");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            latch.countDown();  // 计数减1
        }
    }).start();
}

// 等待所有任务完成
latch.await();
System.out.println("所有任务完成");

import java.util.concurrent.CyclicBarrier;

CyclicBarrier barrier = new CyclicBarrier(3, () -> {
    System.out.println("所有线程到达屏障");
});

for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        try {
            System.out.println("线程到达");
            barrier.await();  // 等待其他线程
            System.out.println("继续执行");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }).start();
}

import java.util.concurrent.Semaphore;

Semaphore semaphore = new Semaphore(3);  // 3个许可

for (int i = 0; i < 10; i++) {
    new Thread(() -> {
        try {
            semaphore.acquire();  // 获取许可
            System.out.println("获取资源");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release();  // 释放许可
        }
    }).start();
}