多线程之Callable的简单学习

时间  2020-01-08
标签 多线程 callable 简单 学习 栏目 Java 繁體版
原文   原文链接 
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：html

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
             *        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();
        }

    }
}

当用完一个线程池后，应该调用该线程池的shutdown()方法，该方法将启动线程池的关闭序列，调用shutdown()方法后的线程池再也不接受新任务，但将之前全部已提交任务执行完。当线程池中的全部任务都执行完成后，池中的全部线程都会死亡；java

 void shutdown();c++


另外也能够调用线程池中的shutdownNow()方法来关闭线程池，该方法试图中止全部正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行任务列表。redis

List<Runnable> shutdownNow();多线程


例子3：Semaphoredom

一个计数信号量。从概念上讲，信号量维护了一个许可集。若有必要，在许可可用前会阻塞每个 acquire()，而后再获取该许可。每一个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。可是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采起相应的行动。拿到信号量的线程能够进入代码，不然就等待。经过acquire()和release()获取和释放访问许可。异步

<1.>
ide

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();
	}
	
}
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程