Java中较为高级的线程同步方法总结！

Bertha 。 2022-08-24 15:30 20阅读 0赞

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