Java中较为高级的线程同步方法总结!
2014-02-12 13:06
381 查看
1.Semaphore
1.1二进制Semaphore
Semaphore算是比较高级点的线程同步工具了,在许多其他语言里也有类似的实现。Semaphore有一个最大的好处就是在初始化时,可以显式的控制并发数。其内部维护这一个c计数器,当计数器小于等于0时,是不允许其他线程访问并发区域的,反之则可以,因此,若将并发数设置为1,则可以确保单一线程同步。下面的例子模拟多线程打印,每个线程提交打印申请,然后执行打印,最后宣布打印结束,代码如下:import java.util.concurrent.Semaphore; public class Program{ public static void main(String[] agrs){ PrintQueue p=new PrintQueue(); Thread[] ths=new Thread[10]; for(int i=0;i<10;i++){ ths[i]=new Thread(new Job(p),"Thread"+i); } for(int i=0;i<10;i++){ ths[i].start(); } } } class PrintQueue{ private Semaphore s; public PrintQueue(){ s=new Semaphore(1);//二进制信号量 } public void printJob(Object document){ try{ s.acquire(); long duration=(long)(Math.random()*100); System.out.printf("线程名:%s 睡眠:%d",Thread.currentThread().getName(),duration); Thread.sleep(duration); } catch(InterruptedException e){ e.printStackTrace(); } finally{ s.release(); } } } class Job implements Runnable{ private PrintQueue p; public Job(PrintQueue p){ this.p=p; } @Override public void run(){ System.out.printf("%s:正在打印一个任务\n ",Thread.currentThread().getName()); this.p.printJob(new Object()); System.out.printf("%s:文件已打印完毕\n ",Thread.currentThread().getName()); } }执行结果如下:
Thread0:正在打印一个任务
Thread9:正在打印一个任务
Thread8:正在打印一个任务
Thread7:正在打印一个任务
Thread6:正在打印一个任务
Thread5:正在打印一个任务
Thread4:正在打印一个任务
Thread3:正在打印一个任务
Thread2:正在打印一个任务
Thread1:正在打印一个任务
线程名:Thread0 睡眠:32 Thread0:文件已打印完毕
线程名:Thread9 睡眠:44 Thread9:文件已打印完毕
线程名:Thread8 睡眠:45 Thread8:文件已打印完毕
线程名:Thread7 睡眠:65 Thread7:文件已打印完毕
线程名:Thread6 睡眠:12 Thread6:文件已打印完毕
线程名:Thread5 睡眠:72 Thread5:文件已打印完毕
线程名:Thread4 睡眠:98 Thread4:文件已打印完毕
线程名:Thread3 睡眠:58 Thread3:文件已打印完毕
线程名:Thread2 睡眠:24 Thread2:文件已打印完毕
线程名:Thread1 睡眠:93 Thread1:文件已打印完毕
可以看到,所有线程提交打印申请后,按照并发顺序一次执行,没有任何并发冲突,谁先获得信号量,谁就先执行,其他剩余线程均等待。这里面还有一个公平信号与非公平信号之说:基本上java所有的多线程工具都支持初始化的时候指定一个布尔变量,true时表明公平,即所有处于等待的线程被筛选的条件为“谁等的时间长就选谁进行执行”,有点first in first out的感觉,而false时则表明不公平(默认是不non-fairness),即所有处于等待的线程被筛选执行是随机的。这也就是为什么多线程往往执行顺序比较混乱的原因。
1.2多重并发控制
若将上面的代码改为s=new Semaphore(3);//即让其每次可以并发3条线程,则输出如下:Thread0:正在打印一个任务
Thread9:正在打印一个任务
Thread8:正在打印一个任务
Thread7:正在打印一个任务
Thread6:正在打印一个任务
Thread5:正在打印一个任务
Thread3:正在打印一个任务
Thread4:正在打印一个任务
Thread2:正在打印一个任务
Thread1:正在打印一个任务
线程名:Thread9 睡眠:26线程名:Thread8 睡眠:46线程名:Thread0 睡眠:79 Thread9:文件已打印完毕
线程名:Thread7 睡眠:35 Thread8:文件已打印完毕
线程名:Thread6 睡眠:90 Thread7:文件已打印完毕
线程名:Thread5 睡眠:40 Thread0:文件已打印完毕
线程名:Thread3 睡眠:84 Thread5:文件已打印完毕
线程名:Thread4 睡眠:13 Thread4:文件已打印完毕
线程名:Thread2 睡眠:77 Thread6:文件已打印完毕
线程名:Thread1 睡眠:12 Thread1:文件已打印完毕
Thread3:文件已打印完毕
Thread2:文件已打印完毕
很明显已经并发冲突了。若要实现分组(每组3个)并发吗,则每一组也要进行同步,代码修改如下:
import java.util.concurrent.Semaphore; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class Program{ public static void main(String[] agrs){ PrintQueue p=new PrintQueue(); Thread[] ths=new Thread[10]; for(int i=0;i<10;i++){ ths[i]=new Thread(new Job(p),"Thread"+i); } for(int i=0;i<10;i++){ ths[i].start(); } } } class PrintQueue{ private Semaphore s; private boolean[] freePrinters; private Lock lock; public PrintQueue(){ s=new Semaphore(3);//二进制信号量 freePrinters=new boolean[3]; for(int i=0;i<3;i++){ freePrinters[i]=true; } lock=new ReentrantLock(); } public void printJob(Object document){ try{ s.acquire(); int printerIndex=getIndex(); long duration=(long)(Math.random()*100); System.out.printf("线程名:%s 睡眠:%d\n",Thread.currentThread().getName(),duration); Thread.sleep(duration); freePrinters[printerIndex]=true;//恢复信号,供下次使用 } catch(InterruptedException e){ e.printStackTrace(); } finally{ s.release(); } } //返回一个内部分组后的同步索引 public int getIndex(){ int index=-1; try{ lock.lock(); for(int i=0;i<freePrinters.length;i++){ if(freePrinters[i]){ freePrinters[i]=false; index=i; break; } } } catch(Exception e){ e.printStackTrace(); } finally{ lock.unlock(); } return index; } } class Job implements Runnable{ private PrintQueue p; public Job(PrintQueue p){ this.p=p; } @Override public void run(){ System.out.printf("%s:正在打印一个任务\n ",Thread.currentThread().getName()); this.p.printJob(new Object()); System.out.printf(" %s:文件已打印完毕\n ",Thread.currentThread().getName()); } }
其中getIndex()方法主要为了维护内部分组后(支持并发3个)组内数据的同步(用lock来同步)。
输出如下:
Thread0:正在打印一个任务
Thread9:正在打印一个任务
Thread8:正在打印一个任务
Thread7:正在打印一个任务
Thread6:正在打印一个任务
Thread5:正在打印一个任务
Thread4:正在打印一个任务
Thread3:正在打印一个任务
Thread2:正在打印一个任务
Thread1:正在打印一个任务
线程名:Thread0 睡眠:82 打印机:0号
线程名:Thread8 睡眠:61 打印机:2号
线程名:Thread9 睡眠:19 打印机:1号
Thread9:文件已打印完毕
线程名:Thread7 睡眠:82 打印机:1号
Thread8:文件已打印完毕
线程名:Thread6 睡眠:26 打印机:2号
Thread0:文件已打印完毕
线程名:Thread5 睡眠:31 打印机:0号
Thread6:文件已打印完毕
线程名:Thread4 睡眠:44 打印机:2号
Thread7:文件已打印完毕
线程名:Thread3 睡眠:54 打印机:1号
Thread5:文件已打印完毕
线程名:Thread2 睡眠:48 打印机:0号
Thread4:文件已打印完毕
线程名:Thread1 睡眠:34 打印机:2号
Thread3:文件已打印完毕
Thread2:文件已打印完毕
Thread1:文件已打印完毕
2.CountDownLatch
CountDownLatch同样也是支持多任务并发的一个工具。它主要用于“等待多个并发事件”,它内部也有一个计数器,当调用await()方法时,线程处于等待状态,只有当内部计数器为0时才继续(countDown()方法来减少计数),也就说,假若有一个需求是这样的:主线程等待所有子线程都到达某一条件时才执行,那么只需要主线程await,然后在启动每个子线程的时候进行countDown操作。下面模拟了一个开会的例子,只有当所有人员都到齐了,会议才能开始。import java.util.concurrent.CountDownLatch; public class Program{ public static void main(String[] agrs){ //开启可容纳10人的会议室 VideoConference v=new VideoConference(10); new Thread(v).start(); //参与人员陆续进场 for(int i=0;i<10;i++){ Participant p=new Participant(i+"号人员",v); new Thread(p).start(); } } } class VideoConference implements Runnable{ private CountDownLatch controller; public VideoConference(int num){ controller=new CountDownLatch(num); } public void arrive(String name){ System.out.printf("%s 已经到达!\n",name); controller.countDown(); System.out.printf("还需要等 %d 个成员!\n",controller.getCount()); } @Override public void run(){ try{ System.out.printf("会议正在初始化...!\n"); controller.await(); System.out.printf("所有人都到齐了,开会吧!\n"); } catch(InterruptedException e){ e.printStackTrace(); } } } class Participant implements Runnable{ private VideoConference conference; private String name; public Participant(String name,VideoConference conference){ this.name=name; this.conference=conference; } @Override public void run(){ long duration=(long)(Math.random()*100); try{ Thread.sleep(duration); conference.arrive(this.name); } catch(InterruptedException e){ } } }
输出:
会议正在初始化...!
0号人员 已经到达!
还需要等 9 个成员!
1号人员 已经到达!
还需要等 8 个成员!
9号人员 已经到达!
还需要等 7 个成员!
4号人员 已经到达!
还需要等 6 个成员!
8号人员 已经到达!
还需要等 5 个成员!
5号人员 已经到达!
还需要等 4 个成员!
6号人员 已经到达!
还需要等 3 个成员!
3号人员 已经到达!
还需要等 2 个成员!
7号人员 已经到达!
还需要等 1 个成员!
2号人员 已经到达!
还需要等 0 个成员!
所有人都到齐了,开会吧!
3.Phaser
import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;
import java.util.List;
import java.util.ArrayList;
import java.io.File;
import java.util.Date;
public class Program{
public static void main(String[] agrs){
Phaser phaser=new Phaser(3);
FileSearch system=new FileSearch("C:\\Windows", "log",phaser);
FileSearch apps=new FileSearch("C:\\Program Files","log",phaser);
FileSearch documents=new FileSearch("C:\\Documents And Settings","log",phaser);
Thread systemThread=new Thread(system,"System");
systemThread.start();
Thread appsThread=new Thread(apps,"Apps");
appsThread.start();
Thread documentsThread=new Thread(documents, "Documents");
documentsThread.start();
try {
systemThread.join();
appsThread.join();
documentsThread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Terminated: "+ phaser.isTerminated());
}
}
class FileSearch implements Runnable{
private String initPath;
private String end;
private List<String> results;
private Phaser phaser;
public FileSearch(String initPath,String end,Phaser phaser){
this.initPath=initPath;
this.end=end;
this.results=new ArrayList<String>();
this.phaser=phaser;
}
private void directoryProcess(File file){
File[] files=file.listFiles();
if(files!=null){
for(int i=0;i<files.length;i++){
if(files[i].isDirectory()){
directoryProcess(files[i]);
}
else{
fileProcess(files[i]);
}
}
}
}
private void fileProcess(File file){
if(file.getName().endsWith(end)){
results.add(file.getAbsolutePath());
}
}
private void filterResults(){
List<String> newResults=new ArrayList<String>();
long actualDate=new Date().getTime();
for(int i=0;i<results.size();i++){
File file=new File(results.get(i));
long fileDate=file.lastModified();
if(actualDate-fileDate<TimeUnit.MILLISECONDS.convert(1,TimeUnit.DAYS)){
newResults.add(results.get(i));
}
}
results=newResults;
}
private boolean checkResults(){
if(results.isEmpty()){
System.out.printf("%s: Phase %d: 0 results.\n",Thread.currentThread().getName(),phaser.getPhase());
System.out.printf("%s: Phase %d: End.\n",Thread.currentThread().getName(),phaser.getPhase());
phaser.arriveAndDeregister();
}
else{
System.out.printf("%s: Phase %d: %d results.\n",Thread.currentThread().getName(),phaser.getPhase(),results.size());
phaser.arriveAndAwaitAdvance();
return true;
}
}
private void showInfo() {
for (int i=0; i<results.size(); i++){
File file=new File(results.get(i));
System.out.printf("%s: %s\n",Thread.currentThread().getName(),file.getAbsolutePath());
}
phaser.arriveAndAwaitAdvance();
}
@Override
public void run(){
File file=new File(initPath);
if(file.isDirectory()){
directoryProcess(file);
}
if(!checkResults()){
return;
}
filterResults();
if(!checkResults()){
return;
}
showInfo();
phaser.arriveAndDeregister();
System.out.printf("%s: Work completed.\n",Thread.currentThread().getName());
}
}
相关文章推荐
- JAVA中线程同步的方法(7种)总结
- java开发线程篇4:总结:线程同步的方法。
- java判断字符串是否为空的方法总结
- java发送http get请求的两种方法(总结)
- 黑马程序员-------------多线程中的(线程、线程组、线程池、以及Java的设计模式)概念及方法的总结
- Java线程总结(2):线程同步
- 线程同步的三种方法(Java 并发编程 concurrent包复习)
- Java Web 回顾总结(五) - HttpServlet - Request - 开发常用方法总结
- java 调用webservice的各种方法总结
- 详解java 单例模式及方法总结
- 【Java高级开发工程师】近一个月的面试总结
- java基础学习总结——Object类的方法
- java基础总结 面向对象之静态方法
- Java GC机制和对象Finalize方法的一点总结
- Json学习总结(1)——Java和JavaScript中使用Json方法大全
- VC线程同步方法总结
- Java类的共同父类Object之常用方法总结
- java finalize方法总结、GC执行finalize的过程
- java 调用webservice的各种方法总结
- Java线程同步方法实例总结