Java多线程工具类

/*
CountDownLatch 闭锁：可以延迟线程的进度，直到锁到达终止状态。闭锁的作用相当于一扇门，在锁到达终止状态之前这扇门一直是关闭的。
当锁到达终止状态时，允许所有线程通过。CountDownLatch 有一个初始值，通过调用 countDown 可以减少该值，一直到 0 时到达终止状态。
　　FutureTask 用于执行一个可返回结果的长任务，任务在单独的线程中执行，其他线程可以用 get 方法取任务结果，如果任务尚未完成，
线程在 get 上阻塞。
　　Semaphore 用于控制同时访问某资源，或同时执行某操作的线程数目。信号量有一个初始值即可以分配的信号量总数目。线程任务开始前
先调用 acquire 取得信号量，任务结束后调用 release 释放信号量。在 acquire 是如果没有可用信号量，线程将阻塞在 acquire 上，直到
其他线程释放一个信号量。
　　CyclicBarrier 栅栏用于多个线程多次迭代时进行同步，在一轮任务中，任何线程完成任务后都在 barrier 上等待，直到所有其他线程也
完成任务，然后一起释放，同时进入下一轮迭代。
*/
　　CountDownLatch 的例子：
　　import java.util.concurrent.CountDownLatch;
　　import java.util.concurrent.atomic.AtomicInteger;
　　public class DemoOfLatch {
　　// 利用闭锁 CountDownLatch 控制主线程和子线程的同步
　　public static void main（String[] args） {
　　int numberOfThread = 5;
//控制子线程开始
　　final CountDownLatch startLatch = new CountDownLatch（1）；
// 子线程计数
　　final CountDownLatch stopLatch = new CountDownLatch（numberOfThread）；
// 用于分配子线程唯一标识
　　final AtomicInteger count = new AtomicInteger（0）；
　　System.out.println（"Main thread start…"）；
　　for （ int i=0; i<numberOfThread; i++ ） {
　　Thread thread = new Thread（ new Runnable（） {
　　@Override
　　public void run（） {
　　int tid = count.getAndIncrement（）；
　　try {
　　// 等代主线程打开启动信号
　　startLatch.await（）；
　　System.out.printf（"Thread %d started…%n", tid）；
　　int duration = （int）（Math.random（） * 5000）；
　　Thread.sleep（duration）；
　　} catch （InterruptedException e） {
　　e.printStackTrace（）；
　　Thread.currentThread（）。interrupt（）；
　　} finally {
　　System.out.printf（"Thread %d stoped…%n", tid）；
　　// 线程终止前减少线程计数
　　stopLatch.countDown（）；
　　}
　　}
　　}）；
　　thread.start（）；
　　}
　　// 在放行子线程之前做点什么别的事情
　　System.out.println（"Main thread do preparation work for child threads…"）；
　　try {
　　Thread.sleep（2000）；
　　} catch （InterruptedException e） {
　　e.printStackTrace（）；
　　}
　　// 打开闭锁放行所有子线程
　　System.out.println（"Main thread let child threads go…"）；
　　startLatch.countDown（）；
　　try {
　　// 等待子线程计数降为 0 即所有子线程执行完毕
　　System.out.println（"Main thread wait for all child threads…"）；
　　stopLatch.await（）；
　　} catch （InterruptedException e） {
　　e.printStackTrace（）；
　　}
　　System.out.println（"Main thread exit…"）；
　　}
　　}
FutureTask 的例子：
　　import java.util.concurrent.Callable;
　　import java.util.concurrent.ExecutionException;
　　import java.util.concurrent.FutureTask;
　　public class DemoOfFutureTask {
　　public static void main（String[] args） {
　　// 创建一个 Future Task 用于并发执行长任务
　　final FutureTask<Movie> future = new FutureTask<Movie>（ new Callable<Movie>（） {
　　@Override
　　public Movie call（） throws Exception {
　　System.out.println（"Future task started…"）；
　　Thread.sleep（5000）；
　　System.out.println（"Future task finished…"）；
　　return new Movie（"2012","Unknown"）；
　　}
　　}
）；
　　// 在子线程中启动任务
　　Thread thread = new Thread（future）；
　　thread.start（）；
　　// 主线程干点别的事情
　　System.out.println（"Now let's do sth eles…"）；
　　try {
　　Thread.sleep（1000）；
　　} catch （InterruptedException e1） {
　　e1.printStackTrace（）；
　　}
　　// 主线程开始取结果
　　System.out.println（"Now wait for result of future task…"）；
　　try {
　　Movie res = future.get（）；
　　System.out.printf（"Result from task is name=%s, actor=%s", res.name, res.actor）；
　　} catch （InterruptedException e） {
　　e.printStackTrace（）；
　　} catch （ExecutionException e） {
　　e.printStackTrace（）；
　　}
　　}
　　public static class Movie {
　　final public String name;
　　final public String actor;
　　public Movie（String name, String actor） {
　　this.name = name;
　　this.actor = actor;
　　}
　　}
　　}
　　Semaphore 的例子：
　　import java.util.concurrent.Semaphore;
　　import java.util.concurrent.atomic.AtomicInteger;
　　public class DemoOfSemaphore {
　　public static void main（String[] args） {
　　final int numOfThread = 5;
// 分配唯一线程标识
　　final AtomicInteger count = new AtomicInteger（0）；
// 控制并发线程数目
　　final Semaphore semaphore = new Semaphore（numOfThread）；
　　for （int i = 0; i < 10; i++） {
　　Thread thread = new Thread（new Runnable（） {
　　@Override
　　public void run（） {
　　int tid = count.getAndIncrement（）；
　　try {
// 等待直到取得信号量
　　System.out.printf（"Thread %d wait on semaphore…%n", tid）；
　　semaphore.acquire（）；
　　// 取得信号量之后做点事情
　　System.out.printf（"Thread %d get semaphore…%n", tid）；
　　int duration = （int）（Math.random（） * 5000）；
　　Thread.sleep（duration）；
　　} catch （InterruptedException e） {
　　e.printStackTrace（）；
　　} finally {
　　// 做完后释放信号量
　　System.out.printf（"Thread %d release semaphore…%n", tid）；
　　semaphore.release（）；
　　}
　　}
　　}）；
　　thread.start（）；
　　}
　　}
　　}
　　CyclicBarrier 的例子：
　　import java.util.concurrent.BrokenBarrierException;
　　import java.util.concurrent.CyclicBarrier;
　　public class DemoOfBarrier {
　　public static void main（String[] args） {
　　final int numOfThread = 2;
　　final int numOfIteration = 2;
　　// 创建一个用于线程同步的 Barrier 对象
　　final CyclicBarrier barrier = new CyclicBarrier（numOfThread,
　　new Runnable（） {
　　// 当所有线程到达 Barrier 后会执行这个任务
　　// 任务在第一个到达 Barrier 的线程中执行
　　@Override
　　public void run（） {
　　long tid = Thread.currentThread（）。getId（）；
　　// 当所有线程完成一轮迭代之后做点清除/准备/提交工作
　　System.out.printf（"[%d] - All threads arrived barrier…%n", tid）；
　　try {
　　Thread.sleep（2000）；
　　} catch （InterruptedException e） {
　　e.printStackTrace（）；
　　}
　　System.out.printf（"[%d] - Clear work done…%n", tid）；
　　}
　　}）；
　　// 创建并启动多个线程，他们在 Barrier 上同步
　　for （int i = 0; i < numOfThread; i++） {
　　Thread thread = new Thread（new Runnable（） {
　　@Override
　　public void run（） {
　　long tid = Thread.currentThread（）。getId（）；
　　for （ int k=0; k<numOfIteration; k++ ） {
　　try {
　　// 线程进行一轮迭代
　　System.out.printf（"Thread %d start its work…%n", tid）；
　　long duration = （int）（Math.random（）*5000）；
　　Thread.sleep（duration）；
　　// 做完迭代后等待其他线程完成迭代
　　System.out.printf（"Thread %d wait on barrier…%n", tid）；
　　int num = barrier.await（）；
　　// 显示完成的顺序
　　System.out.printf（"Thread %d pass barrier with order=%d…%n", tid, num）；
　　} catch （InterruptedException e） {
　　e.printStackTrace（）；
　　Thread.currentThread（）。interrupt（）；
　　} catch （BrokenBarrierException e） {
　　e.printStackTrace（）；
　　}
　　}
　　}
　　}）；
　　thread.start（）；
　　}
　　}
　　}

复制代码

帐号		自动登录	找回密码
密码			加入黑马

Java多线程工具类

0 个回复