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 { private Map mapThread ; private Map map ; private int threadNum; public StarRelationThread(Map mapThread ,Map 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 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 map = new HashMap (); //测试数据 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 threadList = new ArrayList (); 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 mapThread = new HashMap (); for(Map.Entry 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 * http://blog.csdn.net/linghu_java/article/details/17123057 */ ExecutorService executorService = Executors.newFixedThreadPool( 4 ); List > threadFutureList = executorService.invokeAll( threadList ); executorService.shutdownNow(); boolean hasError = false; for ( Future 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(); } }} 当用完一个线程池后，应该调用该线程池的shutdown()方法，该方法将启动线程池的关闭序列，调用shutdown()方法后的线程池不再接受新任务，但将以前所有已提交任务执行完。当线程池中的所有任务都执行完成后，池中的所有线程都会死亡；
void shutdown();
另外也可以调用线程池中的shutdownNow()方法来关闭线程池，该方法试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行任务列表。
List<Runnable> shutdownNow();
例子3：Semaphore
一个计数信号量。从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。拿到信号量的线程可以进入代码，否则就等待。通过acquire()和release()获取和释放访问许可。
<1.>
public void acquire() throws InterruptedException
从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被。获取一个许可（如果提供了一个）并立即返回，将可用的许可数减 1。
<2.>
public void release()
释放一个许可，将其返回给信号量。释放一个许可，将可用的许可数增加 1。如果任意线程试图获取许可，则选中一个线程并将刚刚释放的许可给予它。然后针对线程安排目的启用（或再启用）该线程。
package com.book.admin.test;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;public class SemaphoreTest { public static void main(String[] args) { //线程池 ExecutorService exec = Executors.newCachedThreadPool(); //只能5个线程同时访问 final Semaphore semp = new Semaphore(5); for (int i = 0; i < 20; i++) { final int no = i; Runnable runnable = new Runnable() { @Override public void run() { try { //获取许可 semp.acquire(); System.out.println("Accessing: " + no); Thread.sleep((long) Math.random() * 10000); //访问完后,释放许可，如果注释掉下面的语句,则控制台只能打印5条记录,之后线程一直阻塞 semp.release(); } catch (InterruptedException e) { e.printStackTrace(); } } }; //执行线程 exec.execute(runnable); } //退出线程池 exec.shutdown(); } }