AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

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

2024-01-08 38
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 剑指JUC原理-18.同步协作

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());
  });
}

这样就达到了CountDownLatch的效果啦


在这里面需要注意一下，就是 希望线程池的线程数 和 屏障数是一致的


如果不一致，比如说线程池是 三个核心线程，那么当继续走for循环的时候，由于cyclicbarrier的特性，此时就是两个task1 执行完成了，而task2比较慢。这样有可能会影响最终的效果。


文章标签：
Java
网络架构
程序员
API
安全
倪虹升
目录
相关文章
倪虹升
|
5月前
|
消息中间件 存储 Java
剑指JUC原理-20.并发编程实践（下）
剑指JUC原理-20.并发编程实践
倪虹升
58 0
倪虹升
|
5月前
|
NoSQL 前端开发 Java
剑指JUC原理-20.并发编程实践（中）
剑指JUC原理-20.并发编程实践
倪虹升
58 0
竹子爱熊猫
|
3月前
|
存储 Java 索引
（十二）彻悟并发之JUC分治思想产物-ForkJoin分支合并框架原理剖析下篇
在《(十二)彻悟并发之JUC分治思想产物-ForkJoin分支合并框架原理剖析上篇》中，我们曾初步了解了ForkJoin分支合并框架的使用，也分析框架的成员构成以及任务提交和创建工作的原理实现，在本篇则会对框架的任务执行、任务扫描、线程挂起、结果合并以及任务窃取的源码实现进行分析。
竹子爱熊猫
21 2
竹子爱熊猫
|
3月前
|
存储 监控 Java
（十一）彻悟并发之JUC分治思想产物-ForkJoin分支合并框架原理剖析上篇
在上篇文章《深入理解并发之Java线程池、工作原理、复用原理及源码分析》中，曾详细谈到了Java的线程池框架。在其中也说到了JDK提供的四种原生线程池以及自定义线程池，而本文则再来详细谈谈JDK1.7中新推出的线程池：ForkJoinPool。
竹子爱熊猫
21 1
热爱技术的小郑
|
2月前
|
存储 安全 Java
【多线程面试题 七】、 说一说Java多线程之间的通信方式
Java多线程之间的通信方式主要有：使用Object类的wait()、notify()、notifyAll()方法进行线程间协调；使用Lock接口的Condition的await()、signal()、signalAll()方法实现更灵活的线程间协作；以及使用BlockingQueue作为线程安全的队列来实现生产者和消费者模型的线程通信。
热爱技术的小郑
19 0
公众号:码到三十五
|
4月前
|
Java 调度 开发者
揭秘Java并发包(JUC)的基石：AQS原理和应用
揭秘Java并发包(JUC)的基石：AQS原理和应用
公众号:码到三十五
36 0
碳基肥宅
|
5月前
|
存储 安全 Java
Java多线程基础-9：代码案例之阻塞队列（二）
Java多线程基础教程系列中，介绍了如何实现一个简单的阻塞队列（非泛型版本）。
碳基肥宅
42 0
碳基肥宅
|
5月前
|
消息中间件 存储 负载均衡
Java多线程基础-9：代码案例之阻塞队列（一）
阻塞队列是一种遵循先进先出原则的线程安全数据结构，它在队列满时会阻塞入队操作，队列空时会阻塞出队操作，常用于多线程间的协作，简化同步代码编写。Java中提供了`BlockingQueue`接口及其实现类，如`ArrayBlockingQueue`和`LinkedBlockingQueue`，用于实现生产者-消费者模型，以实现负载均衡和资源的有效利用，如削峰填谷，降低系统压力。
碳基肥宅
60 0
倪虹升
|
5月前
|
存储 缓存 安全
剑指JUC原理-11.不可变设计
剑指JUC原理-11.不可变设计
倪虹升
28 0
倪虹升
|
5月前
|
消息中间件 canal Java
剑指JUC原理-20.并发编程实践（上）
剑指JUC原理-20.并发编程实践
倪虹升
57 0