并发编程 - 线程的并发工具类

线程的并发工具类

线程的并发工具类

• Fork-Join:分治思想
• CountDownLatch:是一组线程等待其他的线程完成工作以后在执行,加强版join await用来等待,countDown负责计数器的减1
• CyclicBarrier: 让一组线程达到某个屏障,被阻塞,一直到组内最后一个线程达到屏障时,屏障开放,所有被阻塞的线程会继续运行
  • countdownlatch放行由第三者控制,CyclicBarrier放行由一组线程本身控制
  • countdownlatch放行条件》=线程数,CyclicBarrier放行条件=线程数
• Semaphore:信号量
• Exchange:交换
• Callable、Future和FutureTask

Fork-Join

继承 RecursiveAction/RecursiveTask/ForkJoinTask,重写compute方法

/**
 *类说明：遍历指定目录（含子目录）找寻指定类型文件
 */
public class FindDirsFiles extends RecursiveAction{
    private File path;//当前任务需要搜寻的目录
    public FindDirsFiles(File path) {
        this.path = path;
    }

    public static void main(String [] args){
        try {
            // 用一个 ForkJoinPool 实例调度总任务
            ForkJoinPool pool = new ForkJoinPool();
            FindDirsFiles task = new FindDirsFiles(new File("F:/"));

            pool.execute(task);//异步调用
            // pool.invokeAll(task);
            System.out.println("Task is Running......");
            Thread.sleep(1);
            int otherWork = 0;
            System.out.println("Main Thread done sth......,otherWork="+otherWork);
            task.join();//阻塞的方法
            System.out.println("Task end");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
	@Override
	protected void compute() {
		List<FindDirsFiles> subTasks = new ArrayList<>();
		File[] files = path.listFiles();
		if(files!=null) {
			for(File file:files) {
				if(file.isDirectory()) {
					subTasks.add(new FindDirsFiles(file));
				}else {
					//遇到文件，检查
					if(file.getAbsolutePath().endsWith("txt")) {
						System.out.println("文件："+file.getAbsolutePath());
					}
				}
			}
			if(!subTasks.isEmpty()) {
				for(FindDirsFiles subTask:invokeAll(subTasks)) {
					subTask.join();//等待子任务执行完成
				}
			}
		}
	}
}

CountDownLatch

// 设置等待次数
CountDownLatch latch = new CountDownLatch(1);

// 其他线程做完后次数减1
latch.countDown();

// 阻塞 latch 次数为0时执行
latch.await();

CyclicBarrier

// 设置放开次数
CyclicBarrier barrier = new CyclicBarrier(4,new CollectThread());

// 阻塞住 达到设置放开次数开放次数是会继续执行
barrier.await();

Semaphore

//定义信号量
Semaphore semaphore=new Semaphore(3);

// 闸门 判断是否有信号量是否还有 没有就阻塞
semaphore.acquire();

// 可用信号量
semaphore.availablePermits();

// 释放信号量
semaphore.release();

Callable、Future和FutureTask

赏

thank you for your support

支付宝

微信