剑指JUC原理-18.同步协作

👏作者简介：大家好，我是爱吃芝士的土豆倪，24届校招生Java选手，很高兴认识大家
📕系列专栏：Spring源码、JUC源码
🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
🍂博主正在努力完成2023计划中：源码溯源，一探究竟
📝联系方式：nhs19990716，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀

文章目录

- Semaphore
- - 基本使用
  - 限制对共享资源的使用
  - - semaphore 实现
  - Semaphore 原理
  - - 加锁解锁流程
- CountdownLatch
- - 源码
  - 应用之同步等待多线程准备完毕
  - 应用之同步等待多个远程调用结束
- CyclicBarrier

Semaphore

基本使用

[ˈsɛməˌfɔr] 信号量，用来限制能同时访问共享资源的线程上限。

public static void main(String[] args) {
        // 1. 创建 semaphore 对象
        Semaphore semaphore = new Semaphore(3);
        // 2. 10个线程同时运行
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                // 3. 获取许可
                try {
                    semaphore.acquire();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    log.debug("running...");
                    sleep(1);
                    log.debug("end...");
                } finally {
                    // 4. 释放许可
                    semaphore.release();
                }
            }).start();
        }
    }

输出

07:35:15.485 c.TestSemaphore [Thread-2] - running... 
07:35:15.485 c.TestSemaphore [Thread-1] - running... 
07:35:15.485 c.TestSemaphore [Thread-0] - running... 
07:35:16.490 c.TestSemaphore [Thread-2] - end... 
07:35:16.490 c.TestSemaphore [Thread-0] - end... 
07:35:16.490 c.TestSemaphore [Thread-1] - end... 
07:35:16.490 c.TestSemaphore [Thread-3] - running... 
07:35:16.490 c.TestSemaphore [Thread-5] - running... 
07:35:16.490 c.TestSemaphore [Thread-4] - running... 
07:35:17.490 c.TestSemaphore [Thread-5] - end... 
07:35:17.490 c.TestSemaphore [Thread-4] - end... 
07:35:17.490 c.TestSemaphore [Thread-3] - end... 
07:35:17.490 c.TestSemaphore [Thread-6] - running... 
07:35:17.490 c.TestSemaphore [Thread-7] - running... 
07:35:17.490 c.TestSemaphore [Thread-9] - running... 
07:35:18.491 c.TestSemaphore [Thread-6] - end... 
07:35:18.491 c.TestSemaphore [Thread-7] - end... 
07:35:18.491 c.TestSemaphore [Thread-9] - end... 
07:35:18.491 c.TestSemaphore [Thread-8] - running... 
07:35:19.492 c.TestSemaphore [Thread-8] - end...

限制对共享资源的使用

semaphore 实现

使用 Semaphore 限流，在访问高峰期时，让请求线程阻塞，高峰期过去再释放许可，当然它只适合限制单机
线程数量，并且仅是限制线程数

用 Semaphore 实现简单连接池，对比『享元模式』下的实现（用wait notify），性能和可读性显然更好

Semaphore 实现简单连接池相对于使用"享元模式"（使用 wait 和 notify）的实现在性能和可读性方面更好的原因主要有以下几点：

简洁的接口：Semaphore 提供了 acquire 和 release 方法来获取和释放资源，这样可以更直观地控制资源的访问。而"享元模式"下的实现需要手动管理线程的等待和唤醒，使用 wait 和 notify 的机制更为复杂，可读性较差。

并发控制：Semaphore 可以灵活地控制并发线程的数量，通过控制许可证的数量来限制同时访问资源的线程数量。而"享元模式"中的 wait 和 notify 机制需要手动管理线程的等待和唤醒，容易出现死锁和同步问题。

性能优化：Semaphore 可以通过设置初始许可证数量、公平性等参数来进行性能优化，而"享元模式"中的 wait 和 notify 更多地依赖于程序员手动编写的同步逻辑，容易出现性能瓶颈和难以调试的问题。

可维护性：Semaphore 提供了一个一致的、标准的接口，易于理解和维护。而"享元模式"下的实现需要程序员手动管理线程的等待和唤醒，代码复杂度高，可维护性差。

因此，Semaphore 实现简单连接池在性能和可读性上更优，它提供了更直观、简洁和安全的方式来管理并发访问资源。同时，Semaphore 对并发的控制更为灵活，使得整个连接池的管理更加高效和可靠。

class Pool {
    // 1. 连接池大小
    private final int poolSize;
    // 2. 连接对象数组
    private Connection[] connections;
    // 3. 连接状态数组 0 表示空闲， 1 表示繁忙
    private AtomicIntegerArray states;
    private Semaphore semaphore;
    // 4. 构造方法初始化
    public Pool(int poolSize) {
        this.poolSize = poolSize;
        // 让许可数与资源数一致
        this.semaphore = new Semaphore(poolSize);
        this.connections = new Connection[poolSize];
        this.states = new AtomicIntegerArray(new int[poolSize]);
        for (int i = 0; i < poolSize; i++) {
            connections[i] = new MockConnection("连接" + (i+1));
        }
    }
    // 5. 借连接
    public Connection borrow() {// t1, t2, t3
        // 获取许可
        try {
            semaphore.acquire(); // 没有许可的线程，在此等待
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        for (int i = 0; i < poolSize; i++) {
            // 获取空闲连接
            if(states.get(i) == 0) {
                if (states.compareAndSet(i, 0, 1)) {
                    log.debug("borrow {}", connections[i]);
                    return connections[i];
                }
            }
        }
        // 不会执行到这里
        return null;
    }
    // 6. 归还连接
    public void free(Connection conn) {
        for (int i = 0; i < poolSize; i++) {
            if (connections[i] == conn) {
                states.set(i, 0);
                log.debug("free {}", conn);
                semaphore.release();
                break;
            }
        }
    }
}

Semaphore 原理

加锁解锁流程

Semaphore 有点像一个停车场，permits 就好像停车位数量，当线程获得了 permits 就像是获得了停车位，然后
停车场显示空余车位减一

刚开始，permits（state）为 3，这时 5 个线程来获取资源

在这里插入图片描述

public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }

static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -2694183684443567898L;

        NonfairSync(int permits) {
            super(permits);
        }

        protected int tryAcquireShared(int acquires) {
            return nonfairTryAcquireShared(acquires);
        }
    }

abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1192457210091910933L;

    // 核心在这里将state设置进来
        Sync(int permits) {
            setState(permits);
        }

        final int getPermits() {
            return getState();
        }

        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }

假设其中 Thread-1，Thread-2，Thread-4 cas 竞争成功，而 Thread-0 和 Thread-3 竞争失败，进入 AQS 队列
park 阻塞

以Thread1为例：

public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
    // 然后又调用 tryAcquireShared
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

protected int tryAcquireShared(int acquires) {
            for (;;) {
                if (hasQueuedPredecessors())
                    return -1;
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    // 以我们当前的例子来说，其实是符合的，3-1 = 2，返回2 不小于0 所以加锁成功
                    // 同理 thread 1 2 3都是一样的
                    return remaining;
            }
        }

// 如果竞争失败了呢？
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);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                // 然后走到这里来 park
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

在这里插入图片描述

这时 Thread-4 释放了 permits，状态如下

public void release() {
        sync.releaseShared(1);
    }

public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            // 释放了之后进入 doReleaseShared方法
            doReleaseShared();
            return true;
        }
        return false;
    }

protected final boolean tryReleaseShared(int releases) {
            for (;;) {
                int current = getState();
                int next = current + releases;
                if (next < current) // overflow
                    throw new Error("Maximum permit count exceeded");
                if (compareAndSetState(current, next))
                    return true;
            }
        }

private void doReleaseShared() {
        
   
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    // unpark唤醒
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

// 唤醒了以后
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) {
                    // 尝试再获取一次
                    // 唤醒了之后，再次for循环就能到这里
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                // 然后走到这里来 park
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }


private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
        
    // 唤醒了以后重新设置头结点即可
    
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

在这里插入图片描述

接下来 Thread-0 竞争成功，permits 再次设置为 0，设置自己为 head 节点，断开原来的 head 节点，unpark 接
下来的 Thread-3 节点，但由于 permits 是 0，因此 Thread-3 在尝试不成功后再次进入 park 状态

在这里插入图片描述

CountdownLatch

用来进行线程同步协作，等待所有线程完成倒计时。

其中构造参数用来初始化等待计数值，await() 用来等待计数归零(等待归零，只有归零了才能继续运行)，countDown() 用来让计数减一

public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);
        new Thread(() -> {
            log.debug("begin...");
            sleep(1);
            latch.countDown();
            log.debug("end...{}", latch.getCount());
        }).start();
        new Thread(() -> {
            log.debug("begin...");
            sleep(2);
            latch.countDown();
            log.debug("end...{}", latch.getCount());
        }).start();
        new Thread(() -> {
            log.debug("begin...");
            sleep(1.5);
            latch.countDown();
            log.debug("end...{}", latch.getCount());
        }).start();
        log.debug("waiting...");
        latch.await();
        log.debug("wait end...");
    }

输出

18:44:00.778 c.TestCountDownLatch [main] - waiting... 
18:44:00.778 c.TestCountDownLatch [Thread-2] - begin... 
18:44:00.778 c.TestCountDownLatch [Thread-0] - begin... 
18:44:00.778 c.TestCountDownLatch [Thread-1] - begin... 
18:44:01.782 c.TestCountDownLatch [Thread-0] - end...2 
18:44:02.283 c.TestCountDownLatch [Thread-2] - end...1 
18:44:02.782 c.TestCountDownLatch [Thread-1] - end...0 
18:44:02.782 c.TestCountDownLatch [main] - wait end...

源码

public class CountDownLatch {
    
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);
        }

        int getCount() {
            return getState();
        }

        // 这里面去判断获得锁的条件 等不等于0，等于0就相当于获得了这个锁
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

        // 如果其他线程调用 realease
        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))
                    // 什么时候 成为0了，就唤醒被阻塞的线程
                    return nextc == 0;
            }
        }
    }

    ...
}

可以配合线程池使用，改进如下

在这里说明一下，其实join也可以达到这样的效果，但是join是属于比较底层的api，而countDownlatch属于比较高级的api。

public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);
        ExecutorService service = Executors.newFixedThreadPool(4);
        service.submit(() -> {
            log.debug("begin...");
            sleep(1);
            latch.countDown();
            log.debug("end...{}", latch.getCount());
        });
        service.submit(() -> {
            log.debug("begin...");
            sleep(1.5);
            latch.countDown();
            log.debug("end...{}", latch.getCount());
        });
        service.submit(() -> {
            log.debug("begin...");
            sleep(2);
            latch.countDown();
            log.debug("end...{}", latch.getCount());
        });
        service.submit(()->{
            try {
                log.debug("waiting...");
                latch.await();
                log.debug("wait end...");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

输出

18:52:25.831 c.TestCountDownLatch [pool-1-thread-3] - begin... 
18:52:25.831 c.TestCountDownLatch [pool-1-thread-1] - begin... 
18:52:25.831 c.TestCountDownLatch [pool-1-thread-2] - begin... 
18:52:25.831 c.TestCountDownLatch [pool-1-thread-4] - waiting... 
18:52:26.835 c.TestCountDownLatch [pool-1-thread-1] - end...2 
18:52:27.335 c.TestCountDownLatch [pool-1-thread-2] - end...1 
18:52:27.835 c.TestCountDownLatch [pool-1-thread-3] - end...0 
18:52:27.835 c.TestCountDownLatch [pool-1-thread-4] - wait end...

应用之同步等待多线程准备完毕

以经典的王者荣耀等待为例子，准备阶段，必须要十个人都准备好了，游戏才能开始，那么就这样来模拟

AtomicInteger num = new AtomicInteger(0);
    ExecutorService service = Executors.newFixedThreadPool(10, (r) -> {
        return new Thread(r, "t" + num.getAndIncrement());
    });
    CountDownLatch latch = new CountDownLatch(10);
    String[] all = new String[10];
    Random r = new Random();
	for (int j = 0; j < 10; j++) {
        int x = j;
        service.submit(() -> {
        	for (int i = 0; i <= 100; i++) {
        	try {
                // 因为加的是随机输出，所以会出现差异值
        		Thread.sleep(r.nextInt(100));
        	} catch (InterruptedException e) {
        	}
        	all[x] = Thread.currentThread().getName() + "(" + (i + "%") + ")";
        	System.out.print("\r" + Arrays.toString(all));
        }
        latch.countDown();
    });
}
latch.await();
System.out.println("\n游戏开始...");
service.shutdown();

中间输出

[t0(52%), t1(47%), t2(51%), t3(40%), t4(49%), t5(44%), t6(49%), t7(52%), t8(46%), t9(46%)]

最后输出

[t0(100%), t1(100%), t2(100%), t3(100%), t4(100%), t5(100%), t6(100%), t7(100%), t8(100%), t9(100%)] 
游戏开始...

应用之同步等待多个远程调用结束

@RestController
public class TestCountDownlatchController {
    @GetMapping("/order/{id}")
    public Map<String, Object> order(@PathVariable int id) {
        HashMap<String, Object> map = new HashMap<>();
        map.put("id", id);
        map.put("total", "2300.00");
        sleep(2000);
        return map;
    }
    @GetMapping("/product/{id}")
    public Map<String, Object> product(@PathVariable int id) {

        HashMap<String, Object> map = new HashMap<>();
        if (id == 1) {
            map.put("name", "小爱音箱");
            map.put("price", 300);
        } else if (id == 2) {
            map.put("name", "小米手机");
            map.put("price", 2000);
        }
        map.put("id", id);
        sleep(1000);
        return map;
    }
    @GetMapping("/logistics/{id}")
    public Map<String, Object> logistics(@PathVariable int id) {
        HashMap<String, Object> map = new HashMap<>();
        map.put("id", id);
        map.put("name", "中通快递");
        sleep(2500);
        return map;
    }
    private void sleep(int millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

rest 远程调用 CountDownLatch

 RestTemplate restTemplate = new RestTemplate();
        log.debug("begin");
        ExecutorService service = Executors.newCachedThreadPool();
        CountDownLatch latch = new CountDownLatch(4);
        service.submit(() -> {
            Map<String, Object> r =
                restTemplate.getForObject("http://localhost:8080/order/{1}", Map.class, 1);
            latch.CountDown();
            log.debug(r);
        });
        service.submit(() -> {
            Map<String, Object> r =
                restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 1);
            latch.CountDown();
            log.debug(r);
        });
        = service.submit(() -> {
            Map<String, Object> r =
                restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 2);
            latch.CountDown();
            log.debug(r);
        });
        = service.submit(() -> {
            Map<String, Object> r =
                restTemplate.getForObject("http://localhost:8080/logistics/{1}", Map.class, 1);
            latch.CountDown();
            log.debug(r);
        });
        latch.await();
        log.debug("执行完毕");
        service.shutdown();

rest 远程调用 future 带返回值

RestTemplate restTemplate = new RestTemplate();
        log.debug("begin");
        ExecutorService service = Executors.newCachedThreadPool();
        CountDownLatch latch = new CountDownLatch(4);
        Future<Map<String,Object>> f1 = service.submit(() -> {
            Map<String, Object> r =
                restTemplate.getForObject("http://localhost:8080/order/{1}", Map.class, 1);
            return r;
        });
        Future<Map<String, Object>> f2 = service.submit(() -> {
            Map<String, Object> r =
                restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 1);
            return r;
        });
        Future<Map<String, Object>> f3 = service.submit(() -> {
            Map<String, Object> r =
                restTemplate.getForObject("http://localhost:8080/product/{1}", Map.class, 2);
            return r;
        });
        Future<Map<String, Object>> f4 = service.submit(() -> {
            Map<String, Object> r =
                restTemplate.getForObject("http://localhost:8080/logistics/{1}", Map.class, 1);
        return r;
        });
        System.out.println(f1.get());
        System.out.println(f2.get());
        System.out.println(f3.get());
        System.out.println(f4.get());
        log.debug("执行完毕");
        service.shutdown();

执行结果会比同步快非常多

但是需要注意的是，没有返回结果的时候，用countdownlatch比较合适，如果有返回值结果的话，还有用future

CyclicBarrier

public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(3);
    	ExecutorService service = Executors.newFixedThreadPool(5);
    	service.submit(() -> {
            log.debug("task1 start...");
            sleep(1);
            latch.countDown();
        });
        service.submit(() -> {
            log.debug("task2 start...");
            sleep(2);
            latch.countDown();
        });
    	try{
            latch.await();
        }catch(InterruptedException e){
            e.printStackTrace();
        }
        log.debug("task1 task2 finish ...");
    	service.shutdown();
    }

但是目前的需求是 task1 task2 被反复的运行三遍

最简单的办法就是将上述的代码放置到for循环中去

public static void main(String[] args) throws InterruptedException {
        
    	ExecutorService service = Executors.newFixedThreadPool(5);
    	for(int i = 0 ; i < 3 ; i++){
            CountDownLatch latch = new CountDownLatch(3);
            service.submit(() -> {
            	log.debug("task1 start...");
            	sleep(1);
            	latch.countDown();
        	});
        	service.submit(() -> {
            	log.debug("task2 start...");
            	sleep(2);
            	latch.countDown();
        	});
    		try{
            	latch.await();
        	}catch(InterruptedException e){
            	e.printStackTrace();
        	}
        }
        log.debug("task1 task2 finish ...");
    	service.shutdown();
    }

这样依然能够实现功能，但是，其实CountDownLatch也被创建了三次。

那能不能重用呢？这个CountDownLatch只能在构造方法的时候，给一个初始值，以后就不能改了。

[ˈsaɪklɪk ˈbæriɚ] 循环栅栏，用来进行线程协作，等待线程满足某个计数。构造时设置『计数个数』，每个线程执
行到某个需要“同步”的时刻调用 await() 方法进行等待，当等待的线程数满足『计数个数』时，继续执行

CyclicBarrier cb = new CyclicBarrier(2); // 个数为2时才会继续执行
ExecutorService service = Executors.newFixedThreadPool(2);
service.submit(()->{
 	System.out.println("线程1开始.."+new Date());
 	try {
 		cb.await(); // 当个数不足时，等待
 	} catch (InterruptedException | BrokenBarrierException e) {
 		e.printStackTrace();
 	}
 	System.out.println("线程1继续向下运行..."+new Date());
});

service.submit(()->{
 	System.out.println("线程2开始.."+new Date());
 	try { Thread.sleep(2000); } catch (InterruptedException e) { }
 		try {
 			cb.await(); // 2 秒后，线程个数够2，继续运行
 		} catch (InterruptedException | BrokenBarrierException e) {
 			e.printStackTrace();
 	}
 	System.out.println("线程2继续向下运行..."+new Date());
});

当await等于0的时候，此时就能够继续往下运行了。

CyclicBarrier cb = new CyclicBarrier(2，()->{
    // 等到两个都执行完成后，就会执行这里面的方法
	log.debug("finish");
}); // 个数为2时才会继续执行

ExecutorService service = Executors.newFixedThreadPool(2);

service.submit(()->{
 	System.out.println("线程1开始.."+new Date());
 	try {
 		cb.await(); // 当个数不足时，等待
 	} catch (InterruptedException | BrokenBarrierException e) {
 		e.printStackTrace();
 	}
 	System.out.println("线程1继续向下运行..."+new Date());
});

service.submit(()->{
 	System.out.println("线程2开始.."+new Date());
 	try { Thread.sleep(2000); } catch (InterruptedException e) { }
 		try {
 			cb.await(); // 2 秒后，线程个数够2，继续运行
 		} catch (InterruptedException | BrokenBarrierException e) {
 			e.printStackTrace();
 	}
 	System.out.println("线程2继续向下运行..."+new Date());
});

再次调用await就会恢复成2，在一个for循环中调用

CyclicBarrier cb = new CyclicBarrier(2，()->{
    // 等到两个都执行完成后，就会执行这里面的方法
	log.debug("finish");
}); // 个数为2时才会继续执行

ExecutorService service = Executors.newFixedThreadPool(2);
for(int i = 0; i < 3 ; i++){
	service.submit(()->{
 		System.out.println("线程1开始.."+new Date());
 		try {
 			cb.await(); // 当个数不足时，等待
 		} catch (InterruptedException | BrokenBarrierException e) {
 			e.printStackTrace();
 		}
 		System.out.println("线程1继续向下运行..."+new Date());
	});

	service.submit(()->{
 		System.out.println("线程2开始.."+new Date());
 		try { Thread.sleep(2000); } catch (InterruptedException e) { }
 			try {
 				cb.await(); // 2 秒后，线程个数够2，继续运行
 			} catch (InterruptedException | BrokenBarrierException e) {
 				e.printStackTrace();
 		}
 		System.out.println("线程2继续向下运行..."+new Date());
	});
}