Java多线程Future模式

Posted 2020-09-18

Java多线程Future模式有些类似于Ajax的异步请求
Future模式的核心在于：去除了主函数的等待时间，并使得原本需要等待的时间段可以用于处理其他业务逻辑

假设服务器的处理某个业务，该业务可以分成AB两个过程，并且AB两个过程之间不需要彼此的返回结果
A过程需要1秒钟，B过程需要2秒钟，主线程其他操作2秒钟
按照正常编写，程序大概需要执行5秒
如果按照Future模式只需要执行2秒（取其中运行时间最久的线程的运行时间）

Future模式的核心实现在于两个方面

1.多线程运行
　　主线程采用多线的方式，运行几个业务无关的任务来节省主线的等待时间。
2.锁的锁定和释放
　　子线程执行指定的任务时，需要保证主线程能正确的获取子线程的返回数据
　　这里分两种情况
　　（1）子线程的运行时间要大于主线程处理其他业务的时间，此时主线程需要获取子线程的返回值，而子线程却还没有执行完毕，
　　　　所以在这个时候需要让主线程进入等待。子线程执行完毕以后立刻唤醒主线程。
　　（2）子线程的运行时间小于主线处理其他业务的时间，此时无需要等待。

知识要点：
　　1、wait(),notify()需要和synchronized一块使用，去掉会报 java.lang.IllegalMonitorStateException

　　1>当前线程不含有当前对象的锁资源的时候，调用obj.wait()方法;
　　2>当前线程不含有当前对象的锁资源的时候，调用obj.notify()方法。
　　3>当前线程不含有当前对象的锁资源的时候，调用obj.notifyAll()方法。

　　2、wait()方法会释放锁

Service模拟服务器处理业务的过程

package future;

import java.util.Date;

/**
 * 服务器
 * 
 * @author wpy
 * 
 */
public class Service {
	/**
	 * 1.服务器的处理某个业务，该业务可以分成AB两个过程，并且AB两个过程之间不需要彼此的返回结果
	 * 2.A过程需要1秒钟，B过程需要2秒钟，主线程其他操作2秒钟
	 * 
	 * @param args
	 * @throws InterruptedException
	 */
	public static void main(String[] args) throws InterruptedException {
		Service service = new Service();
		long notUseFuture = service.notUseFuture();
		
		System.out.println("==============================");
		long useFuture = service.useFuture();
		
		System.out.println("==============================");
		System.out.println("notUseFuture整个业务耗时"+notUseFuture);
		System.out.println("useFuture整个业务耗时"+useFuture);
	}

	public long useFuture() throws InterruptedException {
		Date startOn = new Date();

		String name = Thread.currentThread().getName();
		final FutureDate<String> futureDateA = new FutureDate<>();
		final FutureDate<String> futureDateB = new FutureDate<>();

		Thread a = new Thread(new Runnable() {

			@Override
			public void run() {
				String name = Thread.currentThread().getName();
				System.out.println(name + ":任务A开始执行");
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				futureDateA.setData(name + ":任务A执行结果");
				System.out.println(name + ":任务A执行结束");
			}
		}, "线程A");

		Thread b = new Thread(new Runnable() {

			@Override
			public void run() {
				String name = Thread.currentThread().getName();
				System.out.println(name + ":任务B开始执行");
				try {
					Thread.sleep(3000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				futureDateB.setData(name + ":任务B执行结果");
				System.out.println(name + ":任务B执行结束");
			}
		}, "线程B");

		Date before = new Date();
		a.start();
		b.start();
		Date after = new Date();
		System.out.println(name + ":a,b阻塞主线程时间:"
				+ (after.getTime() - before.getTime()));

		// 假设其他业务执行两秒钟
		Thread.sleep(2000);

		before = new Date();
		String dataA = futureDateA.getData();
		after = new Date();
		System.out.println(name + ":获取A线程结果时间:"
				+ (after.getTime() - before.getTime()));

		before = new Date();
		String dataB = futureDateB.getData();
		after = new Date();
		System.out.println(name + ":获取线程结果时间:"
				+ (after.getTime() - before.getTime()));

		System.out.println(name + ":A线程结果：" + dataA);
		System.out.println(name + ":B线程结果：" + dataB);
		Date endOn = new Date();

		/*System.out.println(name + "整个业务耗时"
				+ (endOn.getTime() - startOn.getTime()));*/
		return endOn.getTime() - startOn.getTime();
	}

	public long notUseFuture() throws InterruptedException {
		Date startOn = new Date();

		// 任务A
		String name = Thread.currentThread().getName();
		System.out.println(name + ":任务A开始执行");
		Thread.sleep(1000);
		System.out.println(name + ":任务A执行结束");

		// 任务B
		System.out.println(name + ":任务B开始执行");
		Thread.sleep(3000);
		System.out.println(name + ":任务B执行结束");

		// 主线程其他操作
		Thread.sleep(2000);

		Date endOn = new Date();
		return endOn.getTime() - startOn.getTime();
	}
}

　　

封装的FutrueData类

package future;

/**
 * 用户获取异步任务执行结果
 * @author wpy
 *
 */
public class FutureDate<T> {
	private boolean isReady = false;
	private T data;
	
	/**
	 * 异步任务执行完毕后会通过此方法将执行结果传递给data;
	 * @param data
	 */
	public synchronized void setData(T data){
		if(isReady){
			return;
		}
		this.data = data;
		isReady = true;
		notify();
	}
	
	/**
	 * 如果数据没有加载完毕，线程积蓄等待
	 * wait()会释放锁
	 * @return
	 * @throws InterruptedException
	 */
	public synchronized T getData() throws InterruptedException{
		if(!isReady){
			wait();
		}
		return data;
	}
	
	
}

　执行结果

main:任务A开始执行
main:任务A执行结束
main:任务B开始执行
main:任务B执行结束
==============================
main:a,b阻塞主线程时间:0
线程A:任务A开始执行
线程B:任务B开始执行
线程A:任务A执行结束
main:获取A线程结果时间:0
线程B:任务B执行结束
main:获取线程结果时间:1001
main:A线程结果：线程A:任务A执行结果
main:B线程结果：线程B:任务B执行结果
==============================
notUseFuture整个业务耗时6001
useFuture整个业务耗时3006

 

 

JDK封装的Future简单使用

package test;

import java.util.Date;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;


public class TestFuture {
	
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		
		Callable<Object> callable = new Callable<Object>() {
			
			@Override
			public Object call() throws Exception {
				System.out.println(Thread.currentThread().getName()+":正在构造数据");
				Thread.sleep(1000);
				System.out.println(Thread.currentThread().getName()+":构造数据数据完毕");
				return "结果";
			}
		};
		Date before = new Date();
		FutureTask<Object> futureTask = new FutureTask<>(callable);
		new Thread(futureTask).start();
		
		Thread.sleep(1000);
		Date after = new Date();
		System.out.println((after.getTime() - before.getTime())/1000);
		
		
		
		before = new Date();
		System.out.println(Thread.currentThread().getName()+":主程序准备获取结果");
		Object object = futureTask.get();
		after = new Date();
		System.out.println((after.getTime() - before.getTime())/1000);
		
		System.out.println(object);
	}
}