Saul's blog
SaulJWu
2020-12-21

Callable和线程池

JDK5.0新增线程创建方式有2个：

  • 1、实现Callable接口
  • 2、使用线程池

# Callable

与使用Runnable相比，Callable功能更强大些

  • 相比run()方法，call()可以有返回值
  • 方法可以抛出异常
  • 支持泛型的返回值
  • 需要借助FutureTask类，比如获取返回结果

Future接口：FutrueTask类是Futrue接口的唯一的实现类

  • 可以对具体Runnable、Callable任务的执行结果进行取消、查询是否完成、获取结果等。
  • FutureTask 同时实现了Runnable, Future接口。它既可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值

实现步骤

1、创建一个实现Callable的实现类。

2、重写call()方法，将线程需要操作的代码放入方法中。

3、创建Callable接口实现类的对象。

4、创建FutureTask对象，将Callable接口实现类的对象作为传递参数放入FutureTask构造器中。

5、创建Thread对象，将FutureTask的对象作为参数传递到Thread构造器中，并启动线程。

6、获取Callable中call方法的返回值

遍历1-100以内的偶数，并求和。

//1、创建一个实现`Callable`的实现类。
class NumThread implements Callable {

    //2、重写`call()`方法，将线程需要操作的代码放入方法中。
    @Override
    public Object call() throws Exception {
        int sum = 0;
        for (int i = 1; i <= 100; i++) {
            if (i % 2 == 0) {
                System.out.println(i);
                sum += i;
            }
        }
        return sum;
    }
}


public class ThreadNew {
    public static void main(String[] args) {
        //3、创建`Callable`接口实现类的对象
        NumThread numThread = new NumThread();
        //4、创建FutureTask对象，将`Callable`接口实现类的对象作为传递参数放入FutureTask构造器中。
        FutureTask futureTask = new FutureTask(numThread);
        //5、创建Thread对象，将FutureTask的对象作为参数传递到Thread构造器中，并启动线程
        new Thread(futureTask).start();
        try {
            //6、获取Callable中call方法的返回值
            //get()返回值即为FutureTask构造器参数Callable实现类重写的calll()方法的返回值
            Object sum = futureTask.get();
            System.out.println("sum = " + sum);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

# 线程池

在开发中不会自己一个个去创建线程，其实是创建线程池。

  • 背景：经常创建和销毁、使用量特别大的资源，比如并发情况下的线程，对性能影响很大。
  • 思路：提前创建好多个线程，放入线程池中，使用时直接获取，使用完放回池中。可以避免频繁创建销毁、实现重复利用。类似生活中的公共交通工具。
  • 好处：
    • 1、提高响应速度（减少了创建新线程的时间）
    • 2、降低资源消耗（重复利用线程池中线程，不需要每次都创建）
    • 3、便于线程管理
      • corePoolSize：核心池的大小
      • maximumPoolSize：最大线程数
      • keepAliveTime：线程没有任务时最多保持多长时间后会终止
      • ……

# 线程池相关API

  • JDK 5.0起提供了线程池相关API：ExecutorService和Executors
  • ExecutorService：真正的线程池接口。常见子类ThreadPoolExecutor
    • void execute(Runnable command) ：执行任务/命令，没有返回值，一般用来执行Runnable
      • 需要提供实现Runnable接口的类
    • <T> Future<T> submit(Callable<T> task)：执行任务，有返回值，一般又来执行Callable
      • 需要提供实现Callable接口的类
    • void shutdown() ：关闭线程池
  • Executors：工具类、线程池的工厂类，用于创建并返回不同类型的线程池
    • Executors.newCachedThreadPool()：创建一个可根据需要创建新线程的线程池
    • Executors.newFixedThreadPool(n); 创建一个可重用固定线程数的线程池
    • Executors.newSingleThreadExecutor() ：创建一个只有一个线程的线程池
    • Executors.newScheduledThreadPool(n)：创建一个线程池，它可安排在给定延迟后运行命令或者定期地执行。

示例代码

class NumberThread implements Runnable {
    @Override
    public void run() {
        for (int i = 0; i <= 100; i++) {
            if (i % 2 == 0) {
                System.out.println(Thread.currentThread().getName()+":"+i);
            }
        }
    }
}

class NumberThread2 implements Runnable {
    @Override
    public void run() {
        for (int i = 0; i <= 100; i++) {
            if (i % 2 != 0) {
                System.out.println(Thread.currentThread().getName()+":"+i);
            }
        }
    }
}

public class ThreadPool {
    public static void main(String[] args) {java
         //创建一个可重用固定线程数的线程池
        ExecutorService service = Executors.newFixedThreadPool(10);
        //获取实现类
        System.out.println(service.getClass());
        ThreadPoolExecutor serviceImpl = (ThreadPoolExecutor) service;
        //设置线程池的属性
        //执行任务/命令
        service.execute(new NumberThread());
        service.execute(new NumberThread2());
        //关闭线程池
        service.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# 线程池不允许使用Executors去创建

线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。

Executors各个方法的弊端：

1、newFixedThreadPool和newSingleThreadExecutor:

  • 主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM（Out Of Memory，内存溢出）。

newSingleThreadExecutor：

  • 创建一个单线程的线程池。这个线程池只有一个线程在工作，也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。
  • 此线程池保证所有任务的执行顺序，按照任务的提交顺序(FIFO, LIFO, 优先级)执行。
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
1
2
3
4
5
6

newFixedThreadPool：

  • 创建固定大小的线程池。每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小。
  • 线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。
  • 可控制线程最大并发数，超出的线程会在队列中等待
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
1
2
3
4
5

2、newCachedThreadPool和newScheduledThreadPool:

  • 主要问题是线程数最大数是Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM（Out Of Memory，内存溢出）。

newCachedThreadPool：

  • 创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程，
  • 那么就会回收部分空闲（60秒不执行任务）的线程，当任务数增加时，此线程池又可以智能的添加新线程来处理任务。
  • 此线程池不会对线程池大小做限制
  • 线程池大小完全依赖于操作系统（或者说JVM）能够创建的最大线程大小。
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,java
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}
1
2
3
4
5

newScheduledThreadPool：

  • 创建一个定时线程池，支持定时及周期性任务执行
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}


public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}


public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 线程池创建方式（推荐）

参考资料：https://blog.csdn.net/fly910905/article/details/81584675

根据阿里巴巴java开发规范，推荐了3种线程池创建方式 ：

  • 1、commons-lang3包
  • 2、com.google.guava包
  • 3、 spring配置线程池方式：自定义线程工厂bean需要实现ThreadFactory，可参考该接口的其它默认实现类，使用方式直接注入bean，调用execute(Runnable task)方法即可。

方式一：引入：commons-lang3包

ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,new BasicThreadFactory.Builder().namingPattern("example-schedule-pool-%d").daemon(true).build());
1

方式二：引入：com.google.guava包

ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
    .setNameFormat("demo-pool-%d").build();

//Common Thread Pool
ExecutorService pool = new ThreadPoolExecutor(5, 200,
                                              0L, TimeUnit.MILLISECONDS,
                                              new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

pool.execute(()-> System.out.println(Thread.currentThread().getName()));
pool.shutdown();//gracefully shutdown
1
2
3
4
5
6
7
8
9
10

方式三：spring配置线程池方式：自定义线程工厂bean需要实现ThreadFactory，可参考该接口的其它默认实现类，使用方式直接注入bean，调用execute(Runnable task)方法即可。

<bean id="userThreadPool"
        class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
        <property name="corePoolSize" value="10" />
        <property name="maxPoolSize" value="100" />
        <property name="queueCapacity" value="2000" />
 
    <property name="threadFactory" value= threadFactory />
        <property name="rejectedExecutionHandler">
            <ref local="rejectedExecutionHandler" />
        </property>
    </bean>
    //in code
    userThreadPool.execute(thread);
1
2
3
4
5
6
7
8
9
10
11
12
13

# 总结

多线程的创建方式：

  • 1、继承Thread类
  • 2、实现Runnable接口
  • 3、实现Callable接口
  • 4、使用线程池

同步机制有几种方式：

  • 1、同步代码块
  • 2、同步方法
  • 3、使用Lock锁(ReentrantLock的lock()和unlock())

同步方法的同步监视器是什么？

  • 非静态方法，this
  • 静态方法，类的对象(类名.Class)
帮我改善此页面 (opens new window)
#Callable#线程池
上次更新: 2020/12/23, 05:23:03
线程的通信
字符串相关的类

← 字符串相关的类→

最近更新
01
zabbix学习笔记二
02-28
02
zabbix学习笔记一
02-10
03
Linux访问不了github
12-08
更多文章>
Theme by Vdoing | Copyright © 2020-2022 Saul.J.Wu | MIT License