前言
多线程同步问题是操作系统课程重点内容，是所有程序员解决并发问题无法绕开的一个领域，当然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();
}
}
