java多线程之线程池执行器2（callable）

拼搏现实的明天。 2022-05-29 03:41 275阅读 0赞

一、线程池执行器可以执行有结果返回的任务，称为callable任务。执行方式有三种：

(1) executor.submit();

(2) executor.invokeAny();

(3) executor.invokeAll();

二、executor.submit();

package com.np.ota.test.executor.callable;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 可以返回执行结果的任务Callable
     * @author luke
     *
     */
    public class FactorialCalculator implements Callable<Integer>{
    
    	private Integer number;
    	
    	public FactorialCalculator(Integer number) {
    		this.number = number;
    	}
    
    	@Override
    	public Integer call() throws Exception {
    		int result = 1;
    		if(number == 0 || number==1){
    			return 1;
    		}else{
    			for(int i = 2; i <= number; i++){
    				result *= i;
    				TimeUnit.MICROSECONDS.sleep(20);
    			}
    		}
    		System.out.printf("%s: %d\n",Thread.currentThread().getName(),result);
    		return result;
    	}
    
    }

package com.np.ota.test.executor.callable;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    import java.util.concurrent.ThreadPoolExecutor;
    
    public class Main {
    
    	public static void main(String[] args) {
    		
    		ThreadPoolExecutor poolExecutor = (ThreadPoolExecutor)Executors.newFixedThreadPool(2);
    		
    		List<Future<Integer>> resultList = new ArrayList<>();
    		Random random = new Random();
    		for(int i = 0; i < 10; i++) {
    			Integer number = random.nextInt(10);
    			FactorialCalculator calculator = new FactorialCalculator(number);
    			Future<Integer> result = poolExecutor.submit(calculator);//非阻塞（立即返回）
    			//If you would like to immediately block waiting for a task, 
    			//you can use constructions of the form result = exec.submit(aCallable).get();
    			resultList.add(result);
    		}
    
    		//循环判断任务结束
    		do {
    			for(int i = 0; i < resultList.size(); i++){
    				Future<Integer> result = resultList.get(i);
    				System.out.println("Task"+i+","+result.isDone());
    			}
    		} while (poolExecutor.getCompletedTaskCount() < resultList.size());
    		
    		//输出任务执行结果
    		for(int i = 0; i < resultList.size(); i++){
    			try {
    				Future<Integer> result = resultList.get(i);
    				System.out.println(result.get());
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			} catch (ExecutionException e) {
    				e.printStackTrace();
    			}
    		}
    		
    		//关闭线程池
    		poolExecutor.shutdown();
    		
    	}
    
    }

三、executor.invokeAny();和executor.invokeAll();

package com.np.ota.test.executor.callable2;
    
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    public class UserValidator {
    	
    	private String name;
    
    	public UserValidator(String name) {
    		this.name = name;
    	}
    	
    	public String getName() {
    		return name;
    	}
    
    	public void setName(String name) {
    		this.name = name;
    	}
    
    	public boolean validate(String name,String password){
    		Random random = new Random();
    		try {
    			long duration = (long)(Math.random()*10);
    			TimeUnit.SECONDS.sleep(duration);
    			System.out.println("sleep:"+duration+"--"+this.getName());
    		} catch (InterruptedException e) {
    			return false;
    		}
    		return random.nextBoolean();
    	}
    
    }

package com.np.ota.test.executor.callable2;
    
    import java.util.concurrent.Callable;
    
    public class ValidatorTask implements Callable<String>{
    	
    	private UserValidator validator;
    	private String user;
    	private String password;
    	
    	public ValidatorTask(UserValidator validator, String user, String password) {
    		this.validator = validator;
    		this.user = user;
    		this.password = password;
    	}
    
    	@Override
    	public String call() throws Exception {
    		
    		if(!validator.validate(user, password)){//校验用户，结果是随机的
    			System.out.println("validator:"+this.validator.getName()+" not found user");
    			throw new Exception("Error validating user,"+this.validator.getName());//校验不通过抛异常
    		}
    		System.out.println("validator:"+this.validator.getName()+" found user");
    		return this.validator.getName();
    	}
    
    }

package com.np.ota.test.executor.callable2;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /**
     * ExecutorService调用callable任务有三种方法
     * 1.submit	Future<V> result = poolExecutor.submit(calculator); 执行单个callable任务，
     * 非阻塞立即返回，任务执行完成后result才能获取返回结果。如果调用了result.get()方法，那么将阻塞知道获取结果（或者接收到任务抛出的异常）
     * 2.invokeAny   executor.invokeAny(taskList) 同时调用多个callable任务，返回最先成功执行完（没有抛出异常）的任务结果
     * 	  如果最先的任务抛出异常，那么将executor将会去找下一个没有抛出异常的任务结果返回。如果全部任务都抛出异常那么返回结果才会抛出异常
     * 3.invokeAll	List<Future<String>> resultList = executor.invokeAll(taskList); 返回一个结果集合，
     * 	当遍历集合获取结果时f.get()，将会获取成功返回结果或者抛出任务抛出的异常
     * 		
     * @author luke
     *
     */
    public class Main {
    
    	public static void main(String[] args) {
    
    		String user = "123";
    		String password = "123";
    		
    		UserValidator validator1 = new UserValidator("validator1");
    		UserValidator validator2 = new UserValidator("validator2");
    		
    		ValidatorTask task1 = new ValidatorTask(validator1, user, password);
    		ValidatorTask task2 = new ValidatorTask(validator2, user, password);
    		
    		List<ValidatorTask> taskList = new ArrayList<ValidatorTask>();
    		taskList.add(task1);
    		taskList.add(task2);
    		
    		ExecutorService executor = Executors.newCachedThreadPool();
    		try {
    			System.out.println("result:"+executor.invokeAny(taskList));//invokeAny阻塞执行，获取任意一个任务执行结果
    		} catch (InterruptedException e) {
    			System.out.println("+++");
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			System.out.println("xxx");
    			e.printStackTrace();
    		}
    		
    		/*try {
    			List<Future<String>> resultList = executor.invokeAll(taskList);//invokeAll阻塞获，取所有任务执行结果返回值
    			for(Future<String> f : resultList){
    				try {
    					System.out.println("result:"+f.get());
    				} catch (ExecutionException e) {
    					e.printStackTrace();
    				}
    			}
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}*/
    		
    		executor.shutdown();
    		System.out.println("------------------");
    		
    	}
    
    }

四、实际开发中的使用使用

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3Mzk1NTE_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3Mzk1NTE_size_16_color_FFFFFF_t_70]: /images/20220529/de74cc43a5e7427b8062b6ec4be14d11.png