前言
多线程同步问题是操作系统课程重点内容，是所有程序员解决并发问题无法绕开的一个领域，当然PHP、NodeJS例外。同步问题看起来很复杂，但是只要把那几道经典例题搞懂，也就那么回事。
生产者消费者问题
生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。主要通过对缓冲区加锁，然后适时执行wait、notify即可。
package top.sourcecode.thread;import java.util.Stack;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;import java.util.concurrent.atomic.AtomicInteger;public class ProducerConsumer {
private final int itemsNo; private Stack<Integer> stack; private final int STACK_SIZE; public ProducerConsumer(int itemsNo, int stackSize) { this.itemsNo = itemsNo; this.STACK_SIZE = stackSize; this.stack = new Stack<Integer>();
}
private class Producer implements Runnable {
private int count;
public Producer() { this.count = 0;
} public void run() { while(true) {
synchronized (stack) { while(stack.size() >= STACK_SIZE) { try { stack.wait();
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
}
} if(count < itemsNo) {
produce(); try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
} stack.notifyAll();
} else { break;
}
}
}
}
private void produce() { stack.push(++count);
System.out.println(Thread.currentThread().getName() + " is producing item " + count);
}
}
private class Consumer implements Runnable {
private int count;
public Consumer() { this.count = 0;
}
public void run() { while(true) {
synchronized (stack) { while(stack.empty() && !isFinished()) { try { stack.wait();
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
}
}
consume(); try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
} stack.notifyAll(); //执行notify后，依然要把下面的代码执行完才会真正解锁。
if(isFinished()) { break;
}
}
}
}
public boolean isFinished() { return count == itemsNo;
}
private void consume() { if(isFinished()) { return;
}
System.out.println(Thread.currentThread().getName() + " is consuming item " + stack.pop());
++count;
}
}
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
ProducerConsumer pc = new ProducerConsumer(10, 3);
Producer producer = pc.new Producer();
Consumer consumer = pc.new Consumer(); int producerNo = 3; int consumerNo = 2; for(int i = 0; i < producerNo; ++i) {
exec.execute(producer);
} for(int i = 0; i < consumerNo; ++i) {
exec.execute(consumer);
}
exec.shutdown();
}
}哲学家就餐问题
一圆桌前坐着5位哲学家，两个人中间有一只筷子，桌子中央有面条。哲学家思考问题，当饿了的时候拿起左右两只筷子吃饭，必须拿到两只筷子才能吃饭。上述问题会产生死锁的情况，当5个哲学家都拿起自己右手边的筷子，准备拿左手边的筷子时产生死锁现象。解决办法是资源分级，把筷子从0到4编号，每个哲学家左手筷子的编号必须要比右手筷子小，拿筷子的时候先用左手。当四位哲学家同时拿起他们手边编号较低的餐叉时，只有编号最高的餐叉留在桌上，从而第五位哲学家就不能使用任何一只餐叉了。
package top.sourcecode.thread;import java.util.Random;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class PhilosopherDining { public static void main(String[] args) { int factor = 3; int num = 5;
ExecutorService exec = Executors.newCachedThreadPool();
Chopstick[] chopsticks = new Chopstick[num]; for(int i = 0; i < num; ++i) {
chopsticks[i] = new Chopstick();
} for(int i = 0; i < num; ++i) {
Philosopher philosopher = null; if(i < num - 1) {
philosopher = new Philosopher(chopsticks[i], chopsticks[i + 1], i, factor);
} else {
philosopher = new Philosopher(chopsticks[0], chopsticks[i], i, factor);
}
exec.execute(philosopher);
}
exec.shutdown();
}
}class Chopstick {
private boolean taken;
public Chopstick() {
taken = false;
}
public void take() { synchronized (this) { while(taken) { try { this.wait();
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
}
}
taken = true;
}
}
public void drop() { synchronized (this) {
taken = false; this.notifyAll();
}
}
}class Philosopher implements Runnable {
private Chopstick left; private Chopstick right; private int id; private int factor; private Random random; private int count;
public Philosopher(Chopstick left, Chopstick right, int id, int factor) { this.left = left; this.right = right; this.id = id; this.factor = factor;
count = factor;
random = new Random();
}
public void eat() {
left.take();
right.take();
System.out.println("Philosopher " + id + " is eating."); try {
TimeUnit.MICROSECONDS.sleep(random.nextInt(factor) * 100);
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
}
}
public void think() {
left.drop();
right.drop();
System.out.println("Philosopher " + id + " is thinking."); try {
TimeUnit.MICROSECONDS.sleep(random.nextInt(factor) * 100);
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
}
}
public void run() { while(count-- > 0) {
eat();
think();
}
}
}读者写者问题
有读者和写者两组并发线程，共享一个文件。要求：要求：①允许多个读者可以同时对文件执行读操作；②只允许一个写者往文件中写信息；③任一写者在完成写操作之前不允许其他读者或写者工作；④写者执行写操作前，应让已有的读者和写者全部退出。这个问题主要通过信号量来解决，写者优先。
package top.sourcecode.thread;import java.util.Random;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;import java.util.concurrent.TimeUnit;public class ReaderWriter { private int readerCount; private Semaphore wmutex;//优先写
private Semaphore rwmutex;//读者与写者之间互斥
private Semaphore readerCountMutex;//读者更新readerCount时互斥
public ReaderWriter() {
readerCount = 0;
wmutex = new Semaphore(1);
rwmutex = new Semaphore(1);
readerCountMutex = new Semaphore(1);
}
private class Reader implements Runnable { public void read() { try {
wmutex.acquire();
readerCountMutex.acquire(); if(readerCount == 0) {
rwmutex.acquire();
}
++readerCount;
readerCountMutex.release();
wmutex.release();
System.out.println(Thread.currentThread().getName() + " is reading.");
TimeUnit.MICROSECONDS.sleep(new Random().nextInt(10) * 10);
readerCountMutex.acquire();
--readerCount; if(readerCount == 0) {
rwmutex.release();
}
readerCountMutex.release();
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
}
}
public void run() {
read();
}
}
private class Writer implements Runnable { public void write() { try {
wmutex.acquire();
rwmutex.acquire();
System.out.println(Thread.currentThread().getName() + " is writing.");
TimeUnit.MICROSECONDS.sleep(new Random().nextInt(100) * 1000);
rwmutex.release();
wmutex.release();
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
}
}
public void run() {
write();
}
}
public static void main(String[] args) { int readerNum = 5; int writerNum = 2;
ReaderWriter rw = new ReaderWriter();
Reader reader = rw.new Reader();
Writer writer = rw.new Writer();
ExecutorService exec = Executors.newCachedThreadPool(); for(int i = 0; i < writerNum; ++i) {
exec.execute(writer);
} for(int i = 0; i < readerNum; ++i) {
exec.execute(reader);
}
exec.shutdown();
}
}readerCountMutex和rwmutex可以合并为一个锁。
public class ReaderWriter { private int readerNum; private Semaphore wmutex;//优先写
private Semaphore rwmutex;//读者与写者之间互斥
public ReaderWriter(int readerNum) { this.readerNum = readerNum;
wmutex = new Semaphore(1);
rwmutex = new Semaphore(readerNum);
} private class Reader implements Runnable { public void read() { try {
wmutex.acquire();
rwmutex.acquire();
wmutex.release();
System.out.println(Thread.currentThread().getName() + " is reading.");
TimeUnit.MICROSECONDS.sleep(new Random().nextInt(10) * 10);
rwmutex.release();
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
}
} public void run() {
read();
}
} private class Writer implements Runnable { public void write() { try {
wmutex.acquire();
rwmutex.acquire(readerNum);
System.out.println(Thread.currentThread().getName() + " is writing.");
TimeUnit.MICROSECONDS.sleep(new Random().nextInt(100) * 1000);
rwmutex.release(readerNum);
wmutex.release();
} catch (InterruptedException e) { // TODO Auto-generated catch block
e.printStackTrace();
}
} public void run() {
write();
}
} public static void main(String[] args) { int readerNum = 20; int writerNum = 2;
ReaderWriter rw = new ReaderWriter(readerNum);
Reader reader = rw.new Reader();
Writer writer = rw.new Writer();
ExecutorService exec = Executors.newCachedThreadPool(); for(int i = 0; i < readerNum; ++i) {
exec.execute(reader);
} for(int i = 0; i < writerNum; ++i) {
exec.execute(writer);
}
exec.shutdown();
}
}
作者：MountainKing
链接：https://www.jianshu.com/p/af4e692a861b