Java编程思想-并发

Posted 2022-12-14 vanpersie_9987

在对象上同步

使用synchronized块必须给定一个在其上进行同步的对象。最常见的是synchronized(this),这表示，如果一个线程获得了当前对象的锁，那么该对象其他的synchronized方法和临界区就不能被调用了。

下面同步用了两个锁，所以并发访问时，它们相互独立：

class DualSynch 
    private Object syncObject = new Object;
    //用this即当前对象作为锁
    public synchronized void f() 
        for(int i = 0; i < 5; ++i) 
            System.out.println("f()");
            Thread.yield();
        
    
    public void g() 
        // 使用syncObject作为锁
        synchronized(syncObject) 
            for(int i = 0; i < 5; ++i) 
                System.out.println("g()");
                Thread.yield();
            


        
    




public class SyncTest 
    public static void main(String[] args) 
        final DualSynch ds = new DualSynch();
        new Thread() 
            public void run() 
                ds.f();
             
        .start();
        ds.g();
    

//输出：
g()
f()
g()
f()
g()
f()
g()
f()
g()
f()

线程本地存储

使用ThreadLocal可以避免资源的共享冲突问题，因为ThreadLocal是一种线程本地存储，它可以为使用相同变量的每个不同线程都创建不同的存储：

class Accessor implements Runnable 
    private final int id;
    public Accessor(int idn) 
        this.id = idn;
    
    public void run() 
        while(Thread.currentThread().isInterrupted()) 
            ThreadLocalVariableHolder.increment();
            System.out.println(this);
            Thread.yield();
        
    
    public String toString() 
        return "#" + id + ": " + ThreadLocalVariableHolder.get();
    


public class ThreadLocalVariableHolder 
    private static ThreadLocal<Integer> value = new ThreadLocal<>() 
        private Random rand = new Random(47);
        protected synchronized Integer initalValue() 
            return rand.nextInt();
        
    ;
    public static void increment() 
        value.set(value.get() + 1);
    
    public static int get() 
        return value.get();
    
    public static void main(String[] args) 
        ExecutorService exec = Executors.newCachedThreadPool();
        for(int i = 0 ; i < 5; ++i) 
            exec.execute(new Accessor(i));
        
        TimeUnit.SECONS.sleep(3);
        exec.shutdownNow();
    

//输出
#0: 9295
#1: 556
#2: 432
#3: 949
#4: 43
#0: 9296
#1: 557
#2: 433
#3: 950
#4: 44
... ...

从输出可以看出，每个线程都维护着自己的自增变量，而不会发生冲突。ThreadLocal通常被当做静态存储域。

线程之间的协作

任务的协作，最关键的是这些任务的握手。为了实现这种握手，可以使用互斥的基础特性，在这种情况下，互斥能够确保只有一个任务可以相应某个信号，这样就可以根除任何可能的竞争条件。在互斥之上，可以将自身挂起，直至某些外部条件发生变化。这就是任务间的握手问题，这种握手问题可以通过Object的wait和notify（notifyAll）实现。在JDK1.5的版本中还提供了具有await和signal的Condition对象。

wait可以将某个线程挂起（阻塞），直至其他线程完成了某项工作，或是达成某种条件，可以使挂起的线程继续执行下去（notify/notifyAll）。

注意：调用sleep、yield也是将线程挂起，但是没有释放锁。而wait释放了锁。

由于wait将释放锁。这意味着另一个任务可以获得这个锁。因此在该对象中的其他synchronized方法可以在wait期间被调用。

乍一看起来，wait、notify、notifyAll是Object的方法有些奇怪，毕竟它们只是用来挂起和叫醒线程用的，似乎如果属于Thread方法更加合适。

实际上，wait可以放到任何同步控制方法或是同步块中，这意味着它可以释放任何对象的锁，不管这个对象是否实现了Runnable接口还是继承Thread类。所以，显然wait属于Object类更加合适（notify和notifyAll同理）。

而调用sleep是不释放锁的，即它不操作锁，所以sleep可以出现在非同步控制方法中。这也是sleep属于Thread的一个原因。

看一个关于wait和notify的示例：

class Car 
    private boolean waxOn = false;
    public synchronized void waxed() 
        waxOn = true;
        notifyAll();
    
    public synchronized void buffed() 
        waxOn = false;
        notifyAll();
    
    public synchronized void waitForBuffing() throws InterruptedException 
        while(waxOn == true) 
            wait();
        

    
    public synchronized void waitForWaxing() throws InterruptedException 
        while(waxOn == false) 
            wait();
        

    

    class WaxOn implemments Runnable 
        private Car car;
        public Car(Car c) 
            this.car = c;
        
        public void run() 
            try 
                while(!Thread.interrupted()) 
                    System.out.print("Wax On!");
                    TimeUnit.MILLSECONDS.sleep(200);
                    car.waxed();
                    car.waitForBuffing();
                
             catch(InterruptedException e) 
                System.out.print("Exiting via interrput!");
            
            System.out.print("Ending Wax On task");
        
    
    class WaxOff implemments Runnable 
        private Car car;
        public Car(Car c) 
            this.car = c;
        
        public void run() 
            try 
                while(!Thread.interrupted()) 
                    car.waitForWaxing();
                    System.out.print("Wax Off!");
                    TimeUnit.MILLSECONDS.sleep(200);
                    car.buffed();

                
             catch(InterruptedException e) 
                System.out.print("Exiting via interrput!");
            
            System.out.print("Ending Wax Off task");
        
    



public class WaxMatic 
    public static void main(String[] args) 
        Car c = new Car();
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new WaxOff(c));
        exec.execute(new WaxOn(c));
        TimeUnit.SECONDS.sleep(5);
        exec.shutdown();
    

//输出：
Wax On! Wax Off! Wax On! Wax Off! Wax On! Wax Off! Wax On! Wax Off! Wax On! Wax Off! Wax On! Wax Off! Wax On! Wax Off! Wax On! Wax Off! Wax On! Wax Off! Wax On! Wax Off! Wax On! Wax Off! Wax On! Wax Off!Wax On! Exiting via interrput! Ending Wax On task Exiting via interrput! Ending Wax Off task

Car类中有一个单一的属性waxOn，该属性就是所谓的共享资源。线程池中的两个线程中各执行了一个任务WaxOn和WaxOff，它们就是通过不断根据waxOn这个资源的值而协调工作的。

错失的信号

当两个线程进行notify/notifyAll 、wait进行协作时，有可能会错失某个信号，造成死锁，假设T1线程是通知T2的线程：

T1:
synchronized(shareMonitor) 
    <setup condition for T2>
    shareMonitor.notify();



T2:
while(someCondition) 
    synchronized(shareMonitor) 
        shareMonitor.wait();
    

两个线程T1和T2并发执行，T2线程阻塞，它要等T1线程做好准备条件后，由T1线程叫醒T2线程。

但是在执行过程中，如果首先是T2线程执行了while（someCondition），发现条件满足，返回true，那么将执行synchronized(shareMonitor)，T2线程或得了shareMonitor锁，shareMonitor.wait();必将执行，于是T2阻塞，锁被释放；接着一定是T1线程执行，然后获得shareMonitor这把锁，最后调用shareMonitor.notify();解除T2阻塞。这样可以正常执行。但是如果当T2执行了while（someCondition），发现条件满足，返回true，即将获得shareMonitor锁之前，T2被挂起了，此时谁也没获得锁，T1执行，获得锁，执行shareMonitor.notify();，由于之前T2没有被wait过，也没有阻塞，所以shareMonitor.notify();不会叫醒任何线程，之后T1被挂起，T2执行，它获得了锁，执行wait，把T2挂起，释放锁，它必须等待T1调用notify将它唤醒，可是此时T1已经执行完了，再也没有任何线程叫醒T2，T2会无限忙等。也就是说，如果T2线程先执行完someCondition，发现条件满足，此时线程挂起执行T1线程时，就有可能造成T2无限忙等，造成死锁。

解决方式是防止在someCondition变量上产生竞争：

//T2正确写法，解决死锁
synchronized(shareMonitor) 
    while(someCondition) 
        shareMonitor.wait();
    

如果是T2先执行，将率先获得锁，并且将自己挂起，同时释放锁。T1接着执行，获得锁，调用notify解除T1的阻塞；如果是T1先执行，先获得锁，当控制返回时，T1获得锁，但发现条件已经不满足了，于是wait将得不到执行，这就避免了死锁。

notify()和notifyAll()

notifyAll()将唤醒“所有正在等待的任务”，事实上，这并不意味着任何处于wait()状态中的任务都将被notifyAll()唤醒，当notifyAll()被某个特定锁而被调用时，只有等待这个锁的任务才能被唤醒：

class Blocker 
    synchronized void waitingCall() 
        try 
            while(!Thread.interrupted()) 
                wait();
                System.out.print(Thread.currentThread() + " ");
            
         catch(InterruptedException e) 

        
    
    synchronized void prod() 
        notify();
    
    synchronized void prodAll() 
        notifyAll();
    



class Task implements Runnable 
    static Blocker blocker = new Blocker();
    public void run() 
        blocker.waitingCall();
     

class Task2 implements Runnable 
    static Blocker blocker = new Blocker();
    public void run() 
        blocker.waitingCall();
     


public class NotifyVsNotifyAll 
    public static void main(String[] args) 
        ExecutorService exec = Executors.newCachesThreadPool();
        for(int i = 0;i < 5; ++i) 
            exec.execute(new Task());

        
        exec.execute(new Task2());
        Timer timer = new Timer();
        timer.scheduleAtFixedRate(new TimerTask() 
            boolean prod = true;
            public void run() 
                if(prod) 
                    System.out.print("\\nnotify() ");
                    Task.blocker.prod();
                    prod = false;
                 else 
                    System.out.print("\\notifyAll() ");
                    Task.blocker.prodAll();
                    prod = true;
                

            
        , 400, 400);
        TimerUnit.SECONDS.sleep(5);
        timer.cancel();
        System.out.println("\\nTimer canceled");
        TimerUnit.MILLISECONDS.sleep(500);
        System.out.print("Task2.blocker.prodAll() ");
        Task2.blocker.prodAll();
        TimerUnit.MILLISECONDS.sleep(500);
        System.out.println("\\nShutting down");
        exec.shutdown();
    

//输出略

主线程创建的线程池执行了6个任务，其中有5个是Task，1个是Task2。当这些任务并发执行的时候，5个Task会依次获得Task中的锁Blocker，接着会依次阻塞住，而Task2会获得Task2中的锁Blocker，j接着同样阻塞。接着主线程初始化一个TimerTask用于每隔0.4秒执行交替调用Task的blocker的notify和notifyAll。事实上这两个方法都是对由Task的锁blockers阻塞线程进行解除阻塞的操作，而Task2中的任务依旧阻塞。即便是在主线程中cancel掉了定时任务，Task的5个任务仍然在运行中，而Task2的任务依旧阻塞，直到主线程调用了Task2的blocker的notifyAll()。

生产者与消费者

生产者与消费者是经典的wait与notify协同工作的案例。考虑一个饭店中有一名厨师和一名服务员。服务员必须等待厨师准备好菜之后才能上菜，否则只能等待；同样地，厨师必须等待服务员上完菜才能继续做菜，否则也只能等待：

class Meal 
    private final int orderNum;
    public Meal(int orderNum) 
        this.orderNum = orderNum;
    
    public String toString() 
        return "Meal " + orderNum;
    

//服务员类
class WaitPerson implements Runnable 
    private Restaurant restaurant;
    public waitPerson(Restaurant r) 
        this.restaurant = r;
    
    public void run() 
        try 
            while(!Thread.interrupted()) 
                //在当前对象上同步
                synchronized(this) 
                    //如果没饭了，服务员等待
                    while(restaurant.meal == null) 
                        wait();
                    
                
                //有菜，上菜
                System.out.print("Waitperson got " + restaurant.meal);
                //在Restaurant的chef对象上同步
                synchronized(restaurant.chef) 
                    //菜上完了
                    restaurant.meal = null;
                    //唤醒厨师，让他继续做菜
                    restaurant.chef.notifyAll();
                
            
         catch(InterruptedException e) 
            System.out.print("WaitPerson interrupted!");
        
    

//厨师类
class Chef implements Runnable 
    private Restaurant restaurant;
    //为饭的数量计数
    private int count = 0;
    public Chef(Restaurant r) 
        this.resutaurant = r; 
    
    public void run() 
        try 
            while(!Thread.interrupted()) 
                //在当前厨师对象上同步
                synchronized(this) 
                    //如果有菜，等待
                    while(restautant.meal != null) 
                        wait();
                    

                
                //如果没有菜，做一份菜
                if(++count == 10) 
                    System.out.print("Out of food, closing!");
                    restaurant.shutdown();
                

                System.out,print("Order up!");
                //通知服务员上菜
                synchronized(restaurant.waitPerson) 
                    restaurant.meal = new Meal(count);
                    restaurant.waitPerson.notifyAll();
                
                TimeUnit.MILLSECONDS.sleep(100);

            
         catch(InterruptedException e) 
            System.out,print("Chef interrupted");
        
    


public class Restautant 
    Meal meal;
    ExecutorService exec = Executors.newCachedThreadPool();
    WaitPerson waitPerson = new WaitPerson(this);
    Chef chef = new Chef(this);
    public Restaurant() 
        exec.execute(chef);
        exec.exectte(waitPerson);
    
    public static void main(String[] args) 
        new Restaurant();
    

//输出：
Order up! WaitPerson got Meal 1
Order up! WaitPerson got Meal 2
Order up! WaitPerson got Meal 3
Order up! WaitPerson got Meal 4
Order up! WaitPerson got Meal 5
Order up! WaitPerson got Meal 6
Order up! WaitPerson got Meal 7
Order up! WaitPerson got Meal 8
Order up! WaitPerson got Meal 9
Out of food, closing!
WaitPerson interrupted!
Chef interrupted

使用显式的Lock和Condition

在JDK1.5中，java.util.condition包中还有一个Condition类，该类中的两个主要方法await()和signal()/signalAll()分别表示阻塞和唤醒线程，它们对应于Object中的wait()和notify/notifyAll()。使用Condition类同步比Object更加安全。下面是使用Condition重写的waitForWaxing()和waitForBuffing()：

class Car 
    private Lock lock = new ReetrantLock();
    private Condition condition = lock.newCondition();
    private boolean waxOn = false;
    public void waxed() 
        //使用condition加锁，相当于synchronized(condition)
        lock.lock();
        try 
            waxOn = true;
            //唤醒被condition调用await而阻塞了的线程
            condition.signalAll();


        
        //必须在finally块中解锁，这不同于synchronized，使用lock必须显式释放锁 
        finally 
            lock.unlock();
        
    
    public void buffed() 
        lock.lock();
        try 
            waxOn = false;
            //解除由condition调用await的所在线程的阻塞
            condition.signalAll();
        
    
    public void waitForWaxing() throws InterruptedException 
        lock.lock();
        try 
            while(waxOn == false) 
                //相当于调用了wait：阻塞线程，释放锁
                condition.await();


             finally 
                lock.unlock();
            
        
     
    public void waitForBuffing() throws InterruptedException 
        lock.lock();
        try 
            while(waxOn == true) 
                condition.await();
            
         finally 
            lock.unlock();
        
    


class WaxOn implements Runnable 
    private Car car;
    public WaxOn(Car car) 
        this.car = car;
    
    public void run() 
        while(!Thread.interrupted()) 
            System.out,print("Wax On! ");
            TimeUnit.MILLSECONDS.sleep(200);
            car.waxed();
            car.waitForBuffing();

         catch(InterruptedException e) 
            System.out,print("Exiting via interrupt");
        
        System.out.print("End Wax On Task");
    



class WaxOff implements Runnable 
    private Car car;
    public WaxOff(Car car) 
        this.car = car;
    
    public void run() 
        while(!Thread.interrupted()) 
            car.waitForWaxing();
            System.out,print("Wax Off! ");
            TimeUnit.MILLSECONDS.sleep(200);
            car.buffed();


         catch(InterruptedException e) 
            System.out,print("Exiting via interrupt");
        
        System.out.print("End Wax Off Task");
    


public class WaxMatic2 
    public static void main(String[] args) 
        Car car = new Car();
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new WaxOff(car));
        exec.execute(new WaxOn(car));
        TimeUnit.SECONDS.sleep(5);
        exec.shutdown();
    

//输出余之前相同，略