经典多线程同步问题详解

Java基础

浏览数:255

2019-1-25

前言

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

原文地址:https://www.jianshu.com/p/af4e692a861b