package com.sohu.hot.vis.servlet;
import java.util.concurrent.*;
/**
* 多线程学习之Callable
*
* @author liweihan
* @time 2016-12-29 14:44
*/
public class TestCallableAndFuture {
/**
* Callable 和 Future接口
* Callable是类似于Runnable的接口，实现Callable接口的类和实现Runnable的类都是可被其它线程执行的任务。
* Callable和Runnable有几点不同：
* （1）Callable规定的方法是call()，而Runnable规定的方法是run().
* （2）Callable的任务执行后可返回值，而Runnable的任务是不能返回值的。
* （3）call()方法可抛出异常，而run()方法是不能抛出异常的。
* （4）运行Callable任务可拿到一个Future对象，
* Future 表示异步计算的结果。它提供了检查计算是否完成的方法，以等待计算的完成，并检索计算的结果。
* 通过Future对象可了解任务执行情况，可取消任务的执行，还可获取任务执行的结果。
* Future的cancel方法可以取消任务的执行，它有一布尔参数，参数为 true 表示立即中断任务的执行，
* 参数为 false 表示允许正在运行的任务运行完成。Future的 get 方法等待计算完成，获取计算结果
*/
public static class MyCallable implements Callable {
private int flag = 0;
public MyCallable(int flag) {
this.flag = flag;
}
@Override
public Object call() throws Exception {
if (this.flag == 0) {
return "flag = 0";
}else if (this.flag == 1) {
try {
while (true) {
System.out.println("循环。。。");
Thread.sleep(2000);
}
} catch (InterruptedException e) {
System.out.println("Interruptered");
}
return false;
} else {
throw new Exception("Error flag value!!");
}
}
}
public static void main(String[] args) {
//定义三个Callable类型的任务
MyCallable task1 = new MyCallable(0);
MyCallable task2 = new MyCallable(1);
MyCallable task3 = new MyCallable(2);
//定义一个执行任务的服务
ExecutorService es = Executors.newFixedThreadPool(3);
try {
/**
* 提交并执行任务，任务启动时返回了一个Future对象。
* 如果想得到任务执行的结果或者是异常可对这个Future对象进行操作
*/
Future future1 = es.submit(task1);
//获得第一个任务的结果，如果调用get方法，当前线程会等待任务执行完毕后才往下执行
System.out.println("task1:" + future1.get());
Future future2 = es.submit(task2);
//等待5秒后，再停止第二个任务。因为第二个任务进行的是无限循环
Thread.sleep(5000);
System.out.println("task2 cancel:" + future2.cancel(true));
//获取第三个任务的输出，因为执行第三个任务会引起异常
//所以下面的语句将引起异常
Future future3 = es.submit(task3);
System.out.println("task3:" + future3.get());
} catch (Exception e) {
System.out.println(e.toString());
}
//停止任务执行服务
es.shutdownNow();
}
}例子2：
package com.sohu.hot.vis.servlet;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
* 多线程学习之Callable
*
* @author liweihan
* @time 2016-12-29 15:44
*/
public class TestCallable2 {
static class StarRelationThread implements Callable<Boolean> {
private Map<String, String> mapThread ;
private Map<String, String> map ;
private int threadNum;
public StarRelationThread(Map<String, String> mapThread ,Map<String, String> map,int threadNum) {
this.map = map;
this.threadNum = threadNum;
this.mapThread = mapThread;
}
@Override
public Boolean call() throws Exception {
System.out.println(" 第 " + threadNum + " 个线程处理-开始 ,此线程处理的数量 " + mapThread.size() + ",总的数量为："+map.size());
System.out.println("处理数据 ,并写入redis中");
if (threadNum > 3) {
try {
Thread.sleep(20000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
int sync = 0;
for (Map.Entry<String, String> en : mapThread.entrySet()) {
sync++;
if (sync < 2) {
System.out.println("key :" + en.getKey() + ", value :" + en.getValue());
}
}
System.out.println(" 第 " + threadNum + " 个线程执行完毕！");
return true; //true和flase，可以根据具体业务再做处理
}
}
public static void main(String[] args) {
Map<String, String> map = new HashMap<String, String>();
//测试数据
for (int i = 0; i < 300000; i++) {
map.put("key" + i, "value"+i);
}
//5.分割map+多线程
int totalSize = map.size();
System.out.println("Map totalSize : " + totalSize);
//线程的数量
int threadNum = 16;
//每个线程处理的数量
int threadSize = totalSize / threadNum;
System.out.println("每个线程处理的数量：" + threadSize);
List<StarRelationThread> threadList = new ArrayList<StarRelationThread>();
for (int i = 0; i < threadNum; i++) {
int end ;
if (i == threadNum - 1) {
//最后一个线程
end = threadSize + totalSize % threadNum;
} else {
end = threadSize;
}
int beginNum = i * threadSize;
int endNum = i * threadSize + end;
System.out.println(i + " begin : " + beginNum + " , " + endNum);
int sync = 0;
//分割map
Map<String, String> mapThread = new HashMap<String, String>();
for(Map.Entry<String, String> entry : map.entrySet()) {
sync++;
if (sync > beginNum && sync <= endNum) {
mapThread.put(entry.getKey(), entry.getValue());
}
}
StarRelationThread st = new StarRelationThread(mapThread,map,i);
threadList.add(st);
}
//执行任务
try {
/**
* 线程池的了解：http://blog.csdn.net/coding_or_coded/article/details/6856014
* http://www.cnblogs.com/yezhenhan/archive/2012/01/07/2315645.html
* http://hbiao68.iteye.com/blog/1929245
*
* https://my.oschina.net/u/1419751/blog/359263
* https://my.oschina.net/u/1419751/blog/359263
* http://blog.csdn.net/linghu_java/article/details/17123057
*/
ExecutorService executorService = Executors.newFixedThreadPool( 4 );
List<Future<Boolean>> threadFutureList = executorService.invokeAll( threadList );
executorService.shutdownNow();
boolean hasError = false;
for ( Future<Boolean> threadFuture : threadFutureList ) {
boolean optSuccess = threadFuture.get();
if ( !optSuccess ) {
hasError = true;
}
}
if (hasError) {
System.out.println(" FAIL---------------");
} else {
System.out.println(" SUCCESS ------------------");
}
} catch (Exception e) {
e.printStackTrace();
}
}
}本文出自 “我的JAVA世界” 博客，请务必保留此出处http://hanchaohan.blog.51cto.com/2996417/1888811
原文：http://hanchaohan.blog.51cto.com/2996417/1888811