Java中的并发工具

超、凢脫俗 2022-10-20 01:29 156阅读 0赞

# 1. 等待多线程完成CountDownLatch #

当子线程需要等到子线程执行完成才继续执行可以使用CountDownLatch：

public class CountDownLatchTest { 
    
        /** * 等待其他线程执行完成主线程再继续执行 CountDownLatch * c.countDown(); * c.await(); * * @throws InterruptedException 中断异常 */
        @Test
        public void test() throws InterruptedException { 
            CountDownLatch c = new CountDownLatch(2);
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println(1);
                    c.countDown();
                    System.out.println(2);
                    c.countDown();
                }
            }).start();
            c.await();
            System.out.println("3");
        }
    }

CountDownLatch的构造函数接收一个int类型的参数作为计数器，如果你想等待N个点完 传入N。 当我们调用countDown方法时，N就会减1，await方法 会阻塞当前线程，直到N变成零。由于countDown方法可以用在任何地方，**所以这里说的N个点，可以是N个线程，也可以是1个线程里的N个执行步骤**。用在多个线程时，只需要把这个 CountDownLatch的引用传递到线程里即可。

# 2. 同步屏障CyclicBarrier #

让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，所有被屏障拦截的线程才会继续运行。

/** * 同步屏障 * 如果把new CyclicBarrier(2)修改成new CyclicBarrier(3)，则主线程和子线程会永远等待， * 因为没有第三个线程执行await方法，即没有第三个线程到达屏障，所以之前到达屏障的两个 线程都不会继续执行。 * * @author liulu40 **/
    public class CyclicBarrierShow { 
        static CyclicBarrier c = new CyclicBarrier(2);
    
        public static void main(String[] args) throws Exception { 
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        c.await();
                    } catch (Exception e) { 
                    }
                    System.out.println(1);
                }
            }).start();
            c.await();
            System.out.println(2);
        }
    }

CyclicBarrier还提供一个更高级的构造函数CyclicBarrier（int parties，Runnable barrier- Action），用于在线程到达屏障时，优先执行barrierAction，方便处理更复杂的业务场景

**CyclicBarrier的应用场景：**  
CyclicBarrier可以用于合并计算结果的场景。例如统计银行流水，分别计算4组的流水最后汇总

/** * 银行流水处理服务类 * * @author liulu40 **/
    public class BankWaterService implements Runnable { 
        /** * 创建4个屏障，处理完之后执行当前类的run方法 */
        private CyclicBarrier c = new CyclicBarrier(4, this);
        /** * 假设只有4个sheet，所以只启动4个线程 */
        private Executor executor = Executors.newFixedThreadPool(4);
        /** * 保存每个sheet计算出的银流结果 */
        private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new ConcurrentHashMap<>();
    
        private void count() { 
            for (int i = 0; i < 4; i++) { 
                executor.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        // 计算当前sheet的银流数据，计算代码省略
                        sheetBankWaterCount.put(Thread.currentThread().getName(), 1);
                        try { 
                            // 银流计算完成，插入一个屏障
                            c.await();
                        } catch (InterruptedException |
                                BrokenBarrierException e) { 
                            e.printStackTrace();
                        }
                    }
                });
            }
    
        }
    
        @Override
        public void run() { 
            // 汇总每个sheet计算出的结果
            int result = 0;
            for (Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()) { 
                result += sheet.getValue();
            }
            // 将结果输出
            sheetBankWaterCount.put("result", result);
            System.out.println(result);
        }
    
        public static void main(String[] args) { 
            BankWaterService bankWaterCount = new BankWaterService();
            bankWaterCount.count();
        }
    }

# 3.控制并发线程数的Semaphore #

Semaphore可以用于做流量控制

/** * @author alu4r **/
    public class SemaphoreTest { 
        private static final int THREAD_COUNT = 30;
        private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
        private static Semaphore s = new Semaphore(10);
    
        public static void main(String[] args) { 
            for (int i = 0; i < THREAD_COUNT; i++) { 
                threadPool.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        try { 
                            s.acquire();
                            System.out.println("save data");
                            s.release();
                        } catch (InterruptedException e) { 
                        }
                    }
                });
            }
            threadPool.shutdown();
        }
    }

虽然有30个线程在执行，但是只允许10个并发执行。Semaphore的构造方法 Semaphore（10）表示允许10个线程获取许可证，也就是最大并发数是10。首先线程使用 Semaphore的acquire()方法获取一个许可证，使用完之后调用release()方法归还许可证

*  int availablePermits()：返回此信号量中当前可用的许可证数。
 *  int getQueueLength()：返回正在等待获取许可证的线程数
 *  boolean hasQueuedThreads()：是否有线程正在等待获取许可证
 *  void reducePermits（int reduction）：减少reduction个许可证，是个protected方法
 *  Collection getQueuedThreads()：返回所有等待获取许可证的线程集合，是个protected方 法

# 4.线程间交换数据的Exchanger #

Exchanger用于进行线程间的数据交换。提供一个同步点，在这个同步点，**两个线程**可以交换彼此的数据。这两个线程通过exchange方法交换数据，如果第一个线程先执行exchange()方法，它会一直等待第二个线程也 执行exchange方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。

/** * 数据交换 * * @author alu4r **/
    public class ExchangerTest { 
        private static final Exchanger<String> exgr = new Exchanger<>();
        private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
    
        public static void main(String[] args) { 
            threadPool.execute(new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        // A录入银行流水数据
                        String A = "银行流水A";
                        String B = exgr.exchange(A);
                        System.out.println("A线程收到数据:" + B);
                    } catch (InterruptedException e) { 
                    }
                }
            });
            threadPool.execute(new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        // B录入银行流水数据
                        String B = "银行流水B";
                        String A = exgr.exchange(B);
                        System.out.println("B线程收到数据:" + A);
                    } catch (InterruptedException e) { 
                    }
                }
            });
            threadPool.shutdown();
        }
    }

注意：两个线程之间的交换，当线程C第三次执行exchange方法时候，Exchanger会继续等待第4次执行exchange的方法