Java并发学习之十九——线程同步工具之Phaser

刺骨的言语ヽ痛彻心扉 2022-08-13 00:53 197阅读 0赞

本文是学习网络上的文章时的总结，感谢大家无私的分享。

JDK 1.7 添加了一个新的工具Phaser，Phaser的在功能上与CountDownLatch有部分重合。

下面使用Phaser类来同步3个并发任务。这3个任务会在3个不同的文件夹和它们的子文件夹中搜索扩展名是.log的文件。这个任务被分成3个步骤：

1. 在指定的文件夹和子文件夹中获得文件扩展名为.log的文件列表。  
       2. 在操控台打印结果。

在步骤1和步骤2的结尾我们要检查列表是否为空。如果为空，那么线程直接结束运行并从phaser类中淘汰。

package chapter3;
    
    import java.io.File;
    import java.util.ArrayList;
    import java.util.Date;
    import java.util.List;
    import java.util.concurrent.Phaser;
    import java.util.concurrent.TimeUnit;
    
    public 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.phaser = phaser;
    		this.results = new ArrayList<String>();
    	}
    	
    	private void directoryProcess(File file){
    		File list[] = file.listFiles();
    		if(list != null){
    			for(int i = 0;i< list.length;i++){
    				if(list[i].isDirectory()
    						){
    					directoryProcess(list[i]);
    				}else{
    					fileProcess(list[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();
    			return false;
    
    		}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() {
    		phaser.arriveAndAwaitAdvance();
    		System.out.printf("%s: Starting.\n", Thread.currentThread().getName());
    
    		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());
    
    	}
    	
    	
    	
    }

package chapter3;
    
    import java.util.concurrent.Phaser;
    
    public class Main5 {
    
    	/**
    	 * <p>
    	 * </p>
    	 * @author zhangjunshuai
    	 * @date 2014-9-29 下午4:31:46
    	 * @param args
    	 */
    	public static void main(String[] args) {
    		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 (Exception e) {
    			e.printStackTrace();
    		}
    
    		System.out.println("Terminated: " + phaser.isTerminated());
    
    	}
    
    }

运行结果

![SouthEast][]

[SouthEast]: /images/20220810/65550b7a4a3848a9909c3bf993a4c2e9.png