Java多线程再学习，温故知新（九）线程之间的通信Condition

迈不过友情╰ 2022-04-23 07:58 215阅读 0赞

##  Condition ##

Condition主要是为了在J.U.C框架中提供和Java传统的监视器风格的wait，notify和notifyAll方法类似的功能。

JDK的官方解释如下：   
条件（也称为条件队列 或条件变量）为线程提供了一个含义，以便在某个状态条件现在可能为 true 的另一个线程通知它之前，一直挂起该线程（即让其“等待”）。因为访问此共享状态信息发生在不同的线程中，所以它必须受保护，因此要将某种形式的锁与该条件相关联。等待提供一个条件的主要属性是：以原子方式 释放相关的锁，并挂起当前线程，就像 Object.wait 做的那样。   
Condition实质上是被绑定到一个锁上。

在JUC锁机制(Lock)学习笔记中，我们了解到AQS有一个队列，同样Condition也有一个等待队列，两者是相对独立的队列，因此一个Lock可以有多个Condition，Lock(AQS)的队列主要是阻塞线程的，而Condition的队列也是阻塞线程，但是它是有阻塞和通知解除阻塞的功能   
Condition阻塞时会释放Lock的锁，阻塞流程请看下面的Condition的await()方法。

由于notify的限制，只能随机叫醒等待线程中的一个,conditiocn可以叫醒指定符合条件的线程。

**例子：实现3个线程依次顺序执行，123123...，这里就需要线程间的通信a执行完告诉b，b执行完告诉c。。。**

**首先使用wait()和notifyAll()实现 **

public class Demo {
    	
    	private int signal;
    	
    	public synchronized void a() {
    		while(signal != 0 ) {
    			try {
    				wait();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		System.out.println("a");
    		signal ++;
    		notifyAll();
    	}
    	
    	public synchronized void b() {
    		while(signal != 1) {
    			try {
    				wait();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    		System.out.println("b");
    		signal ++;
    		notifyAll();
    	}
    	
    	public synchronized void c () {
    		while(signal != 2) {
    			try {
    				wait();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    		System.out.println("c");
    		signal = 0;
    		notifyAll();
    	}
    	
    	public static void main(String[] args) {
    		
    		Demo d = new Demo();
    		A a = new A(d);
    		B b = new B(d);
    		C c = new C(d);
    		
    		new Thread(a).start();
    		new Thread(b).start();
    		new Thread(c).start();
    		
    	}
    }
    
    class A implements Runnable {
    	
    	private Demo demo;
    	
    	public A(Demo demo) {
    		this.demo = demo;
    	}
    
    	@Override
    	public void run() {
    		while(true) {
    			demo.a();
    			try {
    				Thread.sleep(1000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    }
    class B implements Runnable {
    	
    	private Demo demo;
    	
    	public B(Demo demo) {
    		this.demo = demo;
    	}
    	
    	@Override
    	public void run() {
    		while(true) {
    			demo.b();
    			try {
    				Thread.sleep(1000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    }
    class C implements Runnable {
    	
    	private Demo demo;
    	
    	public C(Demo demo) {
    		this.demo = demo;
    	}
    	
    	@Override
    	public void run() {
    		while(true) {
    			demo.c();
    			try {
    				Thread.sleep(1000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    }

**使用 Condition实现**

import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class Demo {
    	
    	private int signal;
    	
    	Lock lock = new ReentrantLock();
    	Condition a = lock.newCondition();
    	Condition b = lock.newCondition();
    	Condition c = lock.newCondition();
    	
    	
    	public void a() {
    		lock.lock();
    		while(signal != 0 ) {
    			try {
    				a.await();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		System.out.println("a");
    		signal ++;
    		b.signal();
    		lock.unlock();
    	}
    	
    	public  void b() {
    		lock.lock();
    		while(signal != 1) {
    			try {
    				b.await();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    		System.out.println("b");
    		signal ++;
    		c.signal();
    		lock.unlock();
    	}
    	
    	public  void c () {
    		lock.lock();
    		while(signal != 2) {
    			try {
    				c.await();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		System.out.println("c");
    		signal = 0;
    		a.signal();
    		lock.unlock();
    	}
    	
    	public static void main(String[] args) {
    		
    		Demo d = new Demo();
    		A a = new A(d);
    		B b = new B(d);
    		C c = new C(d);
    		
    		new Thread(a).start();
    		new Thread(b).start();
    		new Thread(c).start();
    		
    	}
    }
    
    class A implements Runnable {
    	
    	private Demo demo;
    	
    	public A(Demo demo) {
    		this.demo = demo;
    	}
    
    	@Override
    	public void run() {
    		while(true) {
    			demo.a();
    			try {
    				Thread.sleep(1000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    }
    class B implements Runnable {
    	
    	private Demo demo;
    	
    	public B(Demo demo) {
    		this.demo = demo;
    	}
    	
    	@Override
    	public void run() {
    		while(true) {
    			demo.b();
    			try {
    				Thread.sleep(1000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    }
    class C implements Runnable {
    	
    	private Demo demo;
    	
    	public C(Demo demo) {
    		this.demo = demo;
    	}
    	
    	@Override
    	public void run() {
    		while(true) {
    			demo.c();
    			try {
    				Thread.sleep(1000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	
    }

实现一个固定大小的队列，队列用数组来实现先进先出

import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class MyQueue<E> {
    
    	private Object[] obj;
    
    	private int addIndex;
    	private int removeIndex;
    	private int queueSize;
    
    	private Lock lock = new ReentrantLock();
    	Condition addCondition = lock.newCondition();
    	Condition removeCondition = lock.newCondition();
    
    	public MyQueue(int count) {
    		obj = new Object[count];
    	}
    
    	public void add(E e) {
    		lock.lock();
    		while (queueSize == obj.length) {
    			try {
    				addCondition.await();
    			} catch (InterruptedException e1) {
    				e1.printStackTrace();
    			}
    		}
    		obj[addIndex] = e;
    
    		if (++addIndex == obj.length) {
    			addIndex = 0;
    		}
    
    		queueSize++;
    		removeCondition.signal();
    		lock.unlock();
    	}
    
    	public void remove() {
    		lock.lock();
    
    		while (queueSize == 0) {
    			try {
    				removeCondition.await();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		obj[removeIndex] = null;
    
    		if (++removeIndex == obj.length) {
    			removeIndex = 0;
    		}
    
    		queueSize--;
    
    		addCondition.signal();
    
    		lock.unlock();
    	}
    	
    }

Condition的源码解析，ReentrantLock下的newCondition方法是用Sync同步器来实现的

private final Sync sync;
    
    //同步器sync的实现
    abstract static class Sync extends AbstractQueuedSynchronizer {
    //...
    }
    //使用同步器Sync下的newCondition方法
    public Condition newCondition() {
            return sync.newCondition();
    }
    //同步器的newCondition方法，创建一个ConditionObject对象
    final ConditionObject newCondition() {
          return new ConditionObject();
    }
    //ConditionObject对象是AQS的一个内部类实现了Condition接口
    public class ConditionObject implements Condition, java.io.Serializable {
    
    }

ConditionObject是Condition在java并发中的具体实现，它是AQS的内部类。因为Condition相关操作都需要获取锁，所以作为AQS的内部类很合理。我们主要关心ConditionObject 下面的await()和signal()方法，另外Condition使用单项列表数据结构。

先了解两个概念

同步队列：调用lock方法或者，遇到同步代码块多线程排队时进入AQS同步队列，竞争获取CPU资源；

等待队列：调用await进入等待状态的线程，每个Condition会有一个等待队列，里面是单项链表结构；

### await方法 ###

ReentrantLock是独占锁，一个线程拿到锁后如果不释放，那么另外一个线程肯定是拿不到锁，所以在lock.lock()和lock.unlock()之间可能有一次释放锁的操作（同样也必然还有一次获取锁的操作）。在进入lock.lock()后唯一可能释放锁的操作就是await()了。也就是说await()操作实际上就是释放锁，然后挂起线程，一旦条件满足就被唤醒，再次获取锁！

public final void await() throws InterruptedException {
        // 1.如果当前线程被中断，则抛出中断异常
        if (Thread.interrupted())
            throw new InterruptedException();
        // 2.将节点加入到Condition队列中去，这里如果lastWaiter是cancel状态，那么会把它踢出Condition队列。
        Node node = addConditionWaiter();
        // 3.调用tryRelease，释放当前线程的锁
        long savedState = fullyRelease(node);
        int interruptMode = 0;
        // 4.为什么会有在AQS的等待队列的判断？
        // 解答：signal操作会将Node从Condition队列中拿出并且放入到等待队列中去，在不在AQS等待队列就看signal是否执行了
        // 如果不在AQS等待队列中，就park当前线程，如果在，就退出循环，这个时候如果被中断，那么就退出循环
        while (!isOnSyncQueue(node)) {
            LockSupport.park(this);
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        // 5.这个时候线程已经被signal()或者signalAll()操作给唤醒了，退出了4中的while循环
        // 自旋等待尝试再次获取锁，调用acquireQueued方法
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        if (node.nextWaiter != null)
            unlinkCancelledWaiters();
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }

## addConditionWaiter方法 ##

将当前线程包装成节点，添加到条件队列。

1、如果条件队列中没有节点，设置firstWaiter的引用；如果条件队列中已有节点，设置lastWaiter的nextWaiter引用；

2、修改lastWaiter的引用；

/**
             * Adds a new waiter to wait queue.
             * @return its new wait node
             */
            private Node addConditionWaiter() {
                Node t = lastWaiter;
                // If lastWaiter is cancelled, clean out.
                if (t != null && t.waitStatus != Node.CONDITION) {
                    unlinkCancelledWaiters();
                    t = lastWaiter;
                }
                Node node = new Node(Thread.currentThread(), Node.CONDITION);//将当前线程封装成节点
                if (t == null)   //如果尾节点为空，说明条件队列中没有节点
                    firstWaiter = node;  //设置条件队列的头节点的引用
                else   //如果条件队列中已有节点
                    t.nextWaiter = node;  //设置条件队列的尾节点的nextWaiter引用
                lastWaiter = node;   //最后修改条件队列的尾节点的引用
                return node;
            }

##  unlinkCancelledWaiters方法 ##

从条件队列中解除cencelled节点的链接。

private void unlinkCancelledWaiters() {
                Node t = firstWaiter;
                Node trail = null;
                while (t != null) {
                    Node next = t.nextWaiter;
                    if (t.waitStatus != Node.CONDITION) {
                        t.nextWaiter = null;  //删除它的next引用
                        if (trail == null)
                            firstWaiter = next; //让firstWaiter指向它原先的next引用
                        else
                            trail.nextWaiter = next;//让它的前驱的next引用指向它原先的next引用
                        if (next == null)
                            lastWaiter = trail;
                    }
                    else
                        trail = t; //用trail记录循环过程中节点的有效（waitStatus为condition）前驱节点
                    t = next;
                }
            }

##  fullyRelease方法 ##

根据当前状态值调用release方法，返回当前状态值。

/**
         * Invokes release with current state value; returns saved state.
         * Cancels node and throws exception on failure.
         * @param node the condition node for this wait
         * @return previous sync state
         */
        final int fullyRelease(Node node) {
            boolean failed = true;
            try {
                int savedState = getState();//获取同步状态的当前值
                if (release(savedState)) {  
                    failed = false;
                    return savedState;
                } else {
                    throw new IllegalMonitorStateException();
                }
            } finally {
                if (failed)
                    node.waitStatus = Node.CANCELLED;
            }
        }

## release方法 ##

唤醒同步队列的头节点的后继节点。

/**
         * Releases in exclusive mode.  Implemented by unblocking one or
         * more threads if {@link #tryRelease} returns true.
         * This method can be used to implement method {@link Lock#unlock}.
         *
         * @param arg the release argument.  This value is conveyed to
         *        {@link #tryRelease} but is otherwise uninterpreted and
         *        can represent anything you like.
         * @return the value returned from {@link #tryRelease}
         */
        public final boolean release(int arg) {
            if (tryRelease(arg)) {
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h); //唤醒头节点的后继节点
                return true;
            }
            return false;
        }

## awaitUninterruptibly方法 ##

conditionObject与object的wait/notify方法的一个很大区别就是，实现了在阻塞等待条件变量时，可以不响应中断。

在conditionObject中，判断park方法是从因中断返回，还是因为signal返回有2种方式：

1、调用Thread.interrupted()判断线程的中断状态是否为true。

2、调用isOnSyncQueue(node)判断节点是否在同步队列中，如果在同步队列中，则是因signal返回。

下面会说到，调用conditionObject的signal方法，会将线程所在的节点从条件队列转移到同步队列中。

在该方法实现中，如果是因为中断唤醒，则节点不在同步队列中，仍满足while循环条件，会继续调用park方法阻塞等待。

/**
             * Implements uninterruptible condition wait.
             * <ol>
             * <li> Save lock state returned by {@link #getState}.
             * <li> Invoke {@link #release} with saved state as argument,
             *      throwing IllegalMonitorStateException if it fails.
             * <li> Block until signalled.
             * <li> Reacquire by invoking specialized version of
             *      {@link #acquire} with saved state as argument.
             * </ol>
             */
            public final void awaitUninterruptibly() {
                Node node = addConditionWaiter();
                int savedState = fullyRelease(node);
                boolean interrupted = false;
                while (!isOnSyncQueue(node)) {  //如果节点不在同步队列中
                    LockSupport.park(this);  //调用park方法阻塞等待
                    if (Thread.interrupted())
                        interrupted = true;
                }
                if (acquireQueued(node, savedState) || interrupted)
                    selfInterrupt();
            }

## 唤醒signal方法 ##

1、前置检查，判断当前线程是否是获取了锁的线程，如果不是抛出IllegalMonitorStateException。

2、获取条件队列的头结点，头结点不为空执行doSignal方法。

/**
             * Moves the longest-waiting thread, if one exists, from the
             * wait queue for this condition to the wait queue for the
             * owning lock.
             *
             * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
             *         returns {@code false}
             */
            public final void signal() {
                if (!isHeldExclusively())
                    throw new IllegalMonitorStateException();
                Node first = firstWaiter;
                if (first != null)
                    doSignal(first);
            }

## doSignal方法 ##

调用transferForSignal方法将条件队列的头节点从条件队列转移到同步队列，并且，将该节点从条件队列删除。

/**
             * Removes and transfers nodes until hit non-cancelled one or
             * null. Split out from signal in part to encourage compilers
             * to inline the case of no waiters.
             * @param first (non-null) the first node on condition queue
             */
            private void doSignal(Node first) {
                do {
                    if ( (firstWaiter = first.nextWaiter) == null)
                        lastWaiter = null;
                    first.nextWaiter = null;
                } while (!transferForSignal(first) &&
                         (first = firstWaiter) != null);
            }

## transferForSignal方法 ##

将节点从条件队列转移到同步队列，如果转移成功，返回true。如果在signal该节点之前，该节点已经取消，返回false。

1、使用cas将节点的waitStatus从CONDITION设置为0，如果不成功，说明节点的waitStatus已经是cancelled，返回false。

2、将节点放入同步队列，并获取它在同步队列的前驱节点的waitStatus。

3、如果前驱节点的waitStatus不为cancelled，使用cas将该前驱节点的waitStatus设置为SIGNAL。如果前驱节点的waitStatus为cancelled，或者cas操作执行不成功，调用unpark方法唤醒该节点的线程进行resync。

为什么在当前驱节点的waitStatus为cancelled，或者将waitStatus设置为SIGNAL的cas操作执行不成功时，才调用unpark方法唤醒该节点的线程？调用signal方法不总应该是调用unpark方法唤醒线程吗？前驱节点的waitStatus为cancelled，还唤醒后继节点的线程，是否意味着该前驱节点的waitStatus是错误的？

首先，调用ConditionObject的signal方法，不像Object.notify方法那样，最终总会调用unpark方法唤醒线程。它将节点从条件队列转移到同步队列，将调用unpark方法唤醒线程的操作交给了节点的前驱节点去做。当前驱节点成为头节点后，释放锁时会唤醒该节点。当然，有个前提条件是，节点的前驱节点的waitStatus必须为SIGNAL，才会在释放锁时唤醒后继节点。所以，只有当前驱节点的waitStatus为cancelled，或者将其waitStatus设置为SIGNAL的cas操作执行不成功时，才会在这里就调用unpark方法唤醒线程。

/**
         * Transfers a node from a condition queue onto sync queue.
         * Returns true if successful.
         * @param node the node
         * @return true if successfully transferred (else the node was
         * cancelled before signal)
         */
        final boolean transferForSignal(Node node) {
            /*
             * If cannot change waitStatus, the node has been cancelled.
             */
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                return false;
     
            /*
             * Splice onto queue and try to set waitStatus of predecessor to
             * indicate that thread is (probably) waiting. If cancelled or
             * attempt to set waitStatus fails, wake up to resync (in which
             * case the waitStatus can be transiently and harmlessly wrong).
             */
            Node p = enq(node); //将节点放入同步队列，并返回它的前驱节点
            int ws = p.waitStatus;
            if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
                LockSupport.unpark(node.thread); 
            return true;
        }