Java多线程再学习，温故知新（十二）CountDownLatch

àì夳堔傛蜴生んèń 2022-04-23 09:16 151阅读 0赞

## CountDownLatch ##

一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。

用给定的计数 初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier。

CountDownLatch 是一个通用同步工具，它有很多用途。将计数 1 初始化的 CountDownLatch 用作一个简单的开/关锁存器，或入口：在通过调用 countDown() 的线程打开入口前，所有调用 await 的线程都一直在入口处等待。用 N 初始化的 CountDownLatch 可以使一个线程在 N 个线程完成某项操作之前一直等待，或者使其在某项操作完成 N 次之前一直等待。

## 实现一个多线程按行累加计算，当所有行计算完成后，再将所有行累加 ##

数据样例

12,43,543,55,15   
343,44  
4,6,6,6

1.先看用自旋方式的实现

import java.io.BufferedReader;
    import java.io.FileReader;
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    
    public class Demo {
    
    	private int[] nums;
    
    	public Demo(int line) {
    		nums = new int[line];
    	}
    
    	public void calc(String line, int index) {
    		String[] nus = line.split(","); // 切分出每个值
    		int total = 0;
    		for (String num : nus) {
    			total += Integer.parseInt(num);
    		}
    		nums[index] = total; // 把计算的结果放到数组中指定的位置
    		System.out.println(Thread.currentThread().getName() + " 执行计算任务... " + line + " 结果为：" + total);
    	}
    
    	public void sum() {
    		System.out.println("汇总线程开始执行... ");
    		int total = 0;
    		for (int i = 0; i < nums.length; i++) {
    			total += nums[i];
    		}
    		System.out.println("最终的结果为：" + total);
    	}
    
    	public static void main(String[] args) {
    
    		List<String> contents = readFile();
    		int lineCount = contents.size();
    
    		Demo d = new Demo(lineCount);
    		for (int i = 0; i < lineCount; i++) {
    			final int j = i;
    			new Thread(new Runnable() {
    				@Override
    				public void run() {
    					d.calc(contents.get(j), j);
    				}
    			}).start();
    		}
    
    		while (Thread.activeCount() > 1) {//等待所有按行计算操作完成
    
    		}
    
    		d.sum();
    	}
    
    	private static List<String> readFile() {
    		List<String> contents = new ArrayList<>();
    		String line = null;
    		BufferedReader br = null;
    		try {
    			br = new BufferedReader(new FileReader("e:\\nums.txt"));
    			while ((line = br.readLine()) != null) {
    				contents.add(line);
    			}
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			if (br != null) {
    				try {
    					br.close();
    				} catch (IOException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    			}
    		}
    		return contents;
    	}
    
    }

2.使用CountDownLatch

import java.io.BufferedReader;
    import java.io.FileReader;
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    
    public class Demo2 {
    
    	private int[] nums;
    
    	public Demo2(int line) {
    		nums = new int[line];
    	}
    
    	public void calc(String line, int index, CountDownLatch latch) {
    		String[] nus = line.split(","); // 切分出每个值
    		int total = 0;
    		for (String num : nus) {
    			total += Integer.parseInt(num);
    		}
    		nums[index] = total; // 把计算的结果放到数组中指定的位置
    		System.out.println(Thread.currentThread().getName() + " 执行计算任务... " + line + " 结果为：" + total);
    		latch.countDown();//递减锁存器的计数，如果计数到达零，则释放所有等待的线程。
    	}
    
    	public void sum() {
    		System.out.println("汇总线程开始执行... ");
    		int total = 0;
    		for (int i = 0; i < nums.length; i++) {
    			total += nums[i];
    		}
    		System.out.println("最终的结果为：" + total);
    	}
    
    	public static void main(String[] args) {
    
    		List<String> contents = readFile();
    		int lineCount = contents.size();
    		
    		CountDownLatch latch = new CountDownLatch(lineCount);
    
    		Demo2 d = new Demo2(lineCount);
    		for (int i = 0; i < lineCount; i++) {
    			final int j = i;
    			new Thread(new Runnable() {
    				@Override
    				public void run() {
    					d.calc(contents.get(j), j, latch);
    				}
    			}).start();
    		}
    
    		try {
    			latch.await();// 使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断。
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    
    		d.sum();
    	}
    
    	private static List<String> readFile() {
    		List<String> contents = new ArrayList<>();
    		String line = null;
    		BufferedReader br = null;
    		try {
    			br = new BufferedReader(new FileReader("e:\\nums.txt"));
    			while ((line = br.readLine()) != null) {
    				contents.add(line);
    			}
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			if (br != null) {
    				try {
    					br.close();
    				} catch (IOException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    			}
    		}
    		return contents;
    	}
    
    }

## 源码介绍 ##

CountDownLatch是一个共享锁，也就是说线程计数如果不为零所有线程都能访问。

### countDown方法 ###

每次调用countDown方法线程计数都会减少一个，但计数变成0所有等待线程都会被叫醒

// 计数减1
     public void countDown() {
           sync.releaseShared(1);
     }
    
    public final boolean releaseShared(int arg) {
            if (tryReleaseShared(arg)) {//尝试释放共享锁
                doReleaseShared();//释放共享锁，上面有泛型不在重复
                //由于state等于0,所以从队列中，从头结点一个接着一个退出(break),for循环等待，
                return true;
            }
            return false;
        }
    //重写AQS的tryReleaseShared方法
    protected boolean tryReleaseShared(int releases) {
                // Decrement count; signal when transition to zero
                for (;;) {
                    int c = getState();
                    if (c == 0) 
                        return false;
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))//比较并设置state
                        return nextc == 0; //等于0的时候，肯定还有一些Node在队列中，所以要调用doReleaseShared(),来清理
                }
            }
    //AQS里的共享锁释放
     private void doReleaseShared() {
            for (;;) {
                Node h = head;//获得头结点
                if (h != null && h != tail) {
                    int ws = h.waitStatus;//获取头结点的状态默认值为0
                    if (ws == Node.SIGNAL) {如果等于SIGNAL唤醒状态
                    	//将头结点的状态置成0，并使用Node.SIGNAL(-1)与0比较,continue，h的状态设置为0,不会再进入if (ws == Node.SIGNAL)
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);
                    }//判断ws是否为0，并且h的状态不等于0，这里是个坑啊，ws等于0，h就是0啊,所以if进不来的，并设置节点为PROPAGATE
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                //如果头结点等于h,其实也没有一直loop，由于上面写的Node h = head,就算前面的条件都不满足，这里一定会break
                if (h == head)                   // loop if head changed
                    break;
            }
        }

### await方法  ###

执行await方法后线程会等待，计数器会加一

// 等待
        public void await() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
    
        public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
            if (Thread.interrupted())//判断是否发生中断
                throw new InterruptedException();
            if (tryAcquireShared(arg) < 0)//注意：-1表示获取到了共享锁，1表示没有获取共享锁
                doAcquireSharedInterruptibly(arg);
        }
    //重写AQS的tryAcquireShared方法
    protected int tryAcquireShared(int acquires) {
                return (getState() == 0) ? 1 : -1;//这里的state就是最开始new CountDownLatch(int count),count等于state
     }
    //如果获取共享锁继续调用doAcquireSharedInterruptibly(arg)
    private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
            //由于采用的公平锁，所以要将节点放到队列里
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
                for (;;) {//本质是等待共享锁的释放
                    final Node p = node.predecessor();//获得节点的前继
                    if (p == head) { //如果前一个节点等于前继
                        int r = tryAcquireShared(arg);//就判断尝试获取锁
                        /*
                        这里要注意一下r的值就2种情况-1和1：
                        	情况1.r为-1,latch没有调用countDown(),state是没有变化的导致state一直大于0或者调用了countDown(),但是state不等于0，直接在for循环中等待
                        	情况2.r为1,证明countDown(),已经减到0,当前线程还在队列中，state已经等于0了.接下来就是唤醒队列中的节点 
                        */
                        if (r >= 0) {
                            setHeadAndPropagate(node, r);//将当前节点设置头结点。
                            p.next = null; // help GC  删除旧的头结点
                            failed = false;
                            return;
                        }
                    }
                    //当前节点不是头结点，当前线程一直等待，直到获取到共享锁。
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }