【Java并发】Executor框架

Executor框架简介

Java的线程既是工作单元,也是执行机制。从JDK5开始,把工作单元和执行机制分离开来。

Executor框架由3大部分组成

  • 任务。

    • 被执行任务需要实现的接口:Runnable接口或Callable接口
    • 异步计算的结果。Future接口和FutureTask类。
  • 任务的执行。两个关键类ThreadPoolExecutor和SeheduledThreadPoolExecutor。

任务

Runnable接口

不含有运行结果

public interface Runnable {
    public abstract void run();
} 

Callable接口

含有运行结果

public interface Callable<V> {
    V call() throws Exception;
}

Runnable与Callable的区别

Runnable和Callable的区别是,
(1)Callable规定的方法是call(),Runnable规定的方法是run().

(2)Callable的任务执行后可返回值,而Runnable的任务是不能返回值的

(3)call方法可以抛出异常,run方法不可以

(4)运行Callable任务可以拿到一个Future对象,Future 表示异步计算的结果。

Future接口

Future接口代表异步计算的结果,通过Future接口提供的方法可以查看异步计算是否执行完成,或者等待执行结果并获取执行结果,同时还可以取消执行。

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);//取消任务
    boolean isCancelled();//是否取消了
    boolean isDone();//任务是否完成
    //获取任务执行结果,如果任务还没完成则会阻塞等待直到任务执行完成
    V get() throws InterruptedException, ExecutionException;
    //等待一段时间尝试获取执行结果,
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

一个使用Runnable的简单例子

import java.util.concurrent.*;

public class Test {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        final ExecutorService exec = Executors.newFixedThreadPool(5);
        Runnable task = new Runnable() {
            public void run() {
                try {
                    System.out.println(Thread.currentThread().getName() + " is running");
                    Thread.sleep(1000 * 10);//休眠指定的时间,此处表示该操作比较耗时
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        exec.submit(task);
        //关闭线程池
        exec.shutdown();
    }
}

一个使用Callable的简单例子

import java.util.concurrent.*;

public class Test {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        final ExecutorService exec = Executors.newFixedThreadPool(5);
        Callable<String> call = new Callable<String>() {
            public String call() throws Exception {
                Thread.sleep(1000 * 10);//休眠指定的时间,此处表示该操作比较耗时
                return "Other less important but longtime things.";
            }
        };
        Future<String> task = exec.submit(call);
        //重要的事情
        System.out.println("Let‘s do important things. start");
        Thread.sleep(1000 * 3);
        System.out.println("Let‘s do important things. end");

        //不重要的事情
        while(!task.isDone()){
            System.out.println("still waiting....");
            Thread.sleep(1000 * 1);
        }
        System.out.println("get sth....");
        String obj = task.get();
        System.out.println(obj);
        //关闭线程池
        exec.shutdown();
    }
}

运行结果

Let‘s do important things. start
Let‘s do important things. end
still waiting....
still waiting....
still waiting....
still waiting....
still waiting....
still waiting....
still waiting....
get sth....
Other less important but longtime things.

任务的执行机制

ThreadPoolExecutor

ThreadPoolExecutor是Executor框架最核心的类,是线程池的实现类。

核心配置参数包括

corePoolSize:核心线程池的大小

maximumPoolSize:最大线程池的大小

BlockingQueue:暂时保存任务的工作队列

RejectedExecutionHandler:当ThreadPoolExecutor已经饱和时(达到了最大线程池大小且工作队列已满)将执行的Handler。

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

可以创建3种类型的ThreadPoolExecutor。

  • FixedThreadPool
  • SingleThreadExecutor
  • CachedThreadPool

1.FixedThreadPool

FixedThreadPool是固定线程数的线程池,最多线程池中有nThreads个线程。

 public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads, //nThreads为固定线程数
                                      0L, TimeUnit.MILLISECONDS, //空闲线程的等待时间为0ms,表示立刻被终止
                                      new LinkedBlockingQueue<Runnable>()); //工作队列
    }

FixedThreadPool的execute()方法内部执行过程

  1. 当新任务被提交时,如果当前运行线程数小于nTheads,创建新线程执行任务
  2. 如果当前运行线程数等于设置的最大线程数nThreads,将新任务加入到工作队列LinkedBlockingQueue中
  3. 线程执行完任务后会反复从LinkedBlockingQueue中获取新任务执行
  4. LinkedBlockingQueue中没有新任务,线程空闲,线程将被终止。

注意点:

由于工作队列使用的是无界队列LinkedBlockingQueue,FixedThreadPool不会拒绝任务(不会调用RejectedExecutionHandler.rejectedExecution()方法)。

2.SingleThreadExecutor

SingleThreadExecutor是FixedThreadPool的特例,线程池中线程的固定数量为1,即最多有一个线程。

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,  //nThreads为固定线程数
                                    0L, TimeUnit.MILLISECONDS, //空闲线程的等待时间为0ms,表示立刻被终止
                                    new LinkedBlockingQueue<Runnable>())); //工作队列
    }

SingleThreadExecutor的execute()方法内部执行过程与注意事项可参考FixedThreadPool的。

3.CachedThreadPool

CachedThreadPool是根据需要创建新线程的线程池。

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE, //coolPoolSize为0,maxinumPoolSize为Integer.MAX_VALUE
                                      60L, TimeUnit.SECONDS, //空闲线程的等待时间,空闲60s后被终止
                                      new SynchronousQueue<Runnable>()); //工作队列
    }

CachedThreadPool的execute()方法的内部运行过程

  1. 当新任务被提交时,主线程将任务插入到工作队列中(SynchronousQueue的offer()方法),如果线程池有空闲线程在等待任务,新任务交给空闲线程处理。
  2. 如果线程池中没有在等待任务的空闲线程,创建新线程执行任务
  3. 线程执行完任务后,等待60s(SynchronousQueue.poll(60, TimeUnit.SECONDS)方法),如果没有等待新任务,线程终止

SynchronousQueue是一个没有容量的BlockingQueue。每一个插入操作必须等待另一个线程的移除操作。

CachedThreadPool使用SynchronousQueue,把主线程提交的任务传递给空闲线程。

注意点:

CachedThreadPool的线程池是无解的,没有限制数量,如果主线程提交任务的速度高于线程处理任务的速度,将不断创建新线程。

极端情况下,会因创建过多线程耗尽CPU和内存。

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor主要用于定时任务(定期执行,给定延迟后执行)

执行方式

1.scheduleAtFixedRate

任务按照固定周期执行,比如设定每10分钟执行一次,第8分钟时候执行了第一次,后续执行时间点为第18分钟,第28分钟,第38分钟

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, //任务
                                                  long initialDelay, //初始延迟
                                                  long period, //任务执行周期
                                                  TimeUnit unit)  

2.scheduleWithFixedDelay

任务按照固定延迟执行,比如设定延迟时间为10是分钟,第8分钟时候执行了第一次,任务执行完成后,再等待10分钟,执行下一次。如果任务执行了2分钟,则下一次为第20分钟

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,//任务
                                                     long initialDelay, //初始延迟
                                                     long delay, //任务执行延迟
                                                     TimeUnit unit) 

实现原理

数据结构

ScheduledThreadPoolExecutor:定时任务执行器

DelayQueue:使用DelayQueue作为任务队列,保存待调度的任务,任务按照执行的时间点排序。DelayQueue内部是用PriorityQueue实现。

ScheduledFutureTask:待调度任务。

ScheduledFutureTask的成员变量:

  • time:任务将被执行的具体时间
  • sequenceNumber:任务序号,time相同时,序号小的先执行
  • period:任务执行的间隔周期

运行机制

ScheduledThreadPoolExecutor内部运行过程

  1. 线程从DelayQueue中获取到到期的ScheduledFutureTask,到期是指time大于等于当前时间。
  2. 执行任务ScheduledFutureTask
  3. 修改ScheduledFutureTask的time为下次要执行的时间,放回到DelayQueue中

原文地址:https://www.cnblogs.com/yeyang/p/9357382.html

时间: 2024-08-28 09:36:40

【Java并发】Executor框架的相关文章

Java并发Executor框架

1 Executor框架简介 从JDK5开始,工作单元和执行机制隔离开来,工作单元包括Runnable和Callable,执行机制由Executor提供. 调用关系:Java线程一对一映射到本地操作系统的系统线程,当多线程程序分解若干任务,使用用户级的调度器(Executor框架)将任务映射为固定数量的线程,底层,操作系统吧.内核将这些线程映射到硬件处理器上. 2.EXecutor结构成员 Executor是一个接口,它将任务的提交与任务的执行分离开来. ThreadPoolExecutor是线

Java的Executor框架和线程池实现原理

一,Java的Executor框架 1,Executor接口 public interface Executor { void execute(Runnable command); } Executor接口是Executor框架中最基础的部分,定义了一个用于执行Runnable的execute方法,它没有实现类只有另一个重要的子接口ExecutorService 2,ExecutorService接口 //继承自Executor接口 public interface ExecutorServic

JAVA的Executor框架

Executor框架分离了任务的创建和执行.JAVA SE5的java.util.concurrent包中的执行器(Executor)管理Thread对象,从而简化了并发编程.Executor引入了一些功能类来管理和使用线程Thread,其中包括线程 池,Executor,Executors,ExecutorService,CompletionService,Future,Callable等. 1.Executor接口 public interface Executor{ void execut

Java并发基础框架AbstractQueuedSynchronizer初探(ReentrantLock的实现分析)

AbstractQueuedSynchronizer是实现Java并发类库的一个基础框架,Java中的各种锁(RenentrantLock, ReentrantReadWriteLock)以及同步工具类(Semaphore, CountDownLatch)等很多都是基于AbstractQueuedSynchronizer实现的.AbstractQueuedSynchronizer 一般简称AQS,Abstract表示他是一个抽象类,Queued表示他是基于先进先出 FIFO 等待队列实现的,Sy

Java多线程—Executor框架概述

1. 任务Task相关的接口与类 1.1 Runnable 表示一个可被执行的命令,通常用于在不同线程中执行任务. package java.lang; public interface Runnable { public void run(); } 1.2 Callable<V> 表示一个有返回结果的任务 package java.util.concurrent; public interface Callable<V> { V call() throws Exception;

java并发编程框架 Executor ExecutorService invokeall

首先介绍两个重要的接口,Executor和ExecutorService,定义如下: Java代码   public interface Executor { void execute(Runnable command); } Java代码   public interface ExecutorService extends Executor { //不再接受新任务,待所有任务执行完毕后关闭ExecutorService void shutdown(); //不再接受新任务,直接关闭Execut

Java并发---- Executor并发框架--ThreadToolExecutor类详解(execute方法)

1.构造方法 请参考上篇文章:http://blog.csdn.net/ochangwen/article/details/53044733 2.源码详解 线程池内部有一些状态,先来了解下这些状态的机制.以下用代码注释的方式来解释其中的含义. /* 这个是用一个int来表示workerCount和runState的,其中runState占int的高3位, 其它29位为workerCount的值. workerCount:当前活动的线程数: runState:线程池的当前状态. 用AtomicIn

Java并发包下锁学习第二篇Java并发基础框架-队列同步器介绍

Java并发包下锁学习第二篇队列同步器 还记得在第一篇文章中,讲到的locks包下的类结果图吗?如下图: ? 从图中,我们可以看到AbstractQueuedSynchronizer这个类很重要(在本文中,凯哥就用AQS来代替这个类).我们先来了解这个类.对这个类了解之后,学习后面的会更容易了. 本篇是<凯哥(凯哥Java:kagejava)并发编程学习>系列之<Lock系列>教程的第一篇:<Java并发包下锁学习第二篇:队列同步器>. 本文主要内容:同步器介绍:同步器

Java并发编程框架Disruptor

Disruptor是什么? Disruptor是一个高性能的异步处理框架,一个轻量级的JMS,和JDK中的BlockingQueue有相似处,但是它的处理速度非常快,获得2011年程序框架创新大奖,号称"一个线程一秒钟可以处理600W个订单"(这有点吓人吧),并且Disruptor不仅仅只有buffer,它提供的功能非常强大,比如它可以帮助我们轻松构建数据流处理(比如一个数据先交给A和B这2个消费者并行处理后在交给C处理,是不是有点想起storm这种流处理,实际上strom的底层就是应

Java并发和多线程(二)Executor框架

Executor框架 1.Task?Thread? 很多人在学习多线程这部分知识的时候,容易搞混两个概念:任务(task)和线程(thread). 并发编程可以使我们的程序可以划分为多个分离的.独立运行的任务.而这些任务具体得由线程来驱动.Java中,Thread类自身不执行任何操作,它只是驱动赋予它的任务,任务由Runnable接口提供. 2.executor Executor是个简单的接口,但它却提供了一种标准的方法将任务的提交过程与任务的执行过程解耦开来,从而无须太大困难就可以为某种类型的