异步处理FutureTask实例「建议收藏」

大家好，又见面了，我是你们的朋友全栈君。

    在Web应用前端，AJAX有同步和异步处理，异步可以避免阻塞。在WEB后端一般业务应用大多为同步处理，但也有一些需要异步处理的场合，比如A系统调B系统接口I，但B系统处理时间很长，这时，A系统主线程不能一直阻塞等待，可以使用异步处理。即先调用接口I，随即做后面的处理，等B系统返回值时再进行返回后处理。

时序为：

A: invoke I

A:do otherthing

B：处理完成，返回值

A：根据接口返回值进行后处理

1、代码例子：

package com.hf.test;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

import com.xxx.framework.po.ResponseResult;

public class FutureTest {
	Map<String, Object> params = new HashMap<String,Object>(); 
	
	ExecutorService executorService = Executors.newFixedThreadPool(3);
	
	public void test1(){
		params.put("clsId","clsId-001");
		params.put("ids", "001,002,003,004");
		FutureTask<ResponseResult> futureTask = new FutureTask<>(
				new MyTask(params));  
		

		executorService.submit(futureTask); 
		executorService.submit(new MonitorTask(futureTask));
		executorService.submit(getCollectJob(futureTask));
		
		System.out.println("HERE!!!");
	} 
	
	
	/**
	 * 返回收集工作
	 * @param futureTask
	 * @return
	 */
	public Runnable getCollectJob(final FutureTask<ResponseResult> futureTask) {
		return new Runnable() {
			public void run() {
				while(true){
					try {
						Thread.sleep(100); 
						if(futureTask.isDone()){
							ResponseResult resp = futureTask.get();
							System.out.println("2返回值：" + resp.getData());
							break;
						}		
						System.out.println("2Not finished.");
					} catch (Exception e) {
						e.printStackTrace();
					} 
				} 
			}
		};
	}
	
	
	/**
	 * 关闭线程池
	 */
	public void closeThreadPool(){
		executorService.shutdown();
	}  
	
	
	/**
	 * 测试主入口
	 * @param args
	 */
	public static void main(String[] args) {
		FutureTest ft = new FutureTest();
		ft.test1();
 
		System.out.println("Do other things...");
		ft.closeThreadPool();
	}
}

程序说明：

    这里采用的模式是调用接口和收集结果2条线，分别用不同的线程来执行。调用接口线程sumit后，收集线程随即开始工作，每隔100ms查询一遍返回状态，若接口返回，则打印返回值，否则一直打印“NOT  FINISHED”。

2、例子2

    若需要处理复杂的情形，如使用回调、传入参数处理返回结果，则可参考下面的例子。

package com.hf.test;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

import com.xxx.framework.po.ResponseResult;

public class FutureTest2 {
	Map<String, Object> initParams = new HashMap<String,Object>(); 
	
	ExecutorService executorService = Executors.newFixedThreadPool(3);
	
	public void test1(){
		initParams.put("clsId","clsId-001");
		initParams.put("ids", "001,002,003,004"); 
		
		MyCallback myCallback = new MyCallback(initParams);
		FutureTask<ResponseResult> futureTask = new FutureTask<>(
				new MyTask2(myCallback));  		

		executorService.submit(futureTask);  
		
		System.out.println("HERE!!!");
	}  
	
	
	/**
	 * 关闭线程池
	 */
	public void closeThreadPool(){
		executorService.shutdown();
	}  
	
	
	/**
	 * 测试主入口
	 * @param args
	 */
	public static void main(String[] args) {
		FutureTest2 ft = new FutureTest2();
		ft.test1();
 
		System.out.println("Do other things...");
		ft.closeThreadPool();
	}
}

回调处理类：

package com.hf.test;

import java.util.HashMap;
import java.util.Map;

import com.xxx.framework.po.ResponseResult;

public class MyCallback {
	Map<String, Object> initParams = new HashMap<String, Object>();
	
	public MyCallback(Map<String, Object> initParams){
		this.initParams = initParams;
	} 
	
	public ResponseResult doSomething(Map<String,Object> params){
		System.out.println("HERE:" + params);
		String result = "Found param pwid:" + params.get("pwid") 
			+ ",state:" + params.get("state");
		System.out.println("异步线程处理完成,结果：" + result);
		return ResponseResult.successResult("异步线程处理完成,结果：" + result);
	}
}

异步任务类：

package com.hf.test;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;

import com.xxx.framework.po.ResponseResult;
 
public class MyTask2 implements Callable<ResponseResult>{ 	
 
	MyCallback myCallback = null;
	
	public MyTask2(MyCallback myCallback){
		this.myCallback = myCallback;
	}  

	@Override
	public ResponseResult call() throws Exception {
		System.out.println("Start processing...");
		
		//模拟处理过程
		Thread.sleep(10*1000); 
		Map<String, Object> params = new HashMap<String,Object>(); 
		params.put("pwid","pwid-001");
		params.put("state", "已退回");
		//模拟处理结束
		ResponseResult result = myCallback.doSomething(params);
		return result;
	}
}

程序说明：

    主线程类为FutureTest2，异步任务类为MyTask2，回调类为MyCallback。将回调处理类作为参数传给异步任务类myTask2，当任务结束时，调用回调处理类MyCallback的回调函数，完成任务结束后的“后处理”。