java中的Executors简介与多线程在网站上逐步优化的运用案例

news/2024/7/2 23:38:18

提供Executor的工厂类
图片描述

忽略了自定义的ThreadFactory、callable和unconfigurable相关的方法

  • newFixedxxx:在任意时刻，最多有nThreads个线程在处理task；如果所有线程都在运行时来了新的任务，它会被扔入队列；如果有线程在执行期间因某种原因终止了运行，如果需要执行后续任务，新的线程将取代它

       return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());

  • newCachedxxx:新任务到来如果线程池中有空闲的线程就复用，否则新建一个线程。如果一个线程超过60秒没有使用，它就会被关闭移除线程池

     return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());

  • newSingleThreadExecutor:仅使用一个线程来处理任务，如果这线程挂了，会产生一个新的线程来代替它。每一个任务被保证按照顺序执行，而且一次只执行一个

      public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}
    使用newFixedxxx方法也能实现类似的作用，但是ThreadPoolExecutor会提供修改线程数的方法，FinalizableDelegatedExecutorService则没有修改的途径，它在DelegatedExecutorService的基础
    上仅提供了执行finalize时候去关闭线程，而DelegatedExecutorService仅暴漏ExecutorService自身的方法

  • newScheduledThreadPool:提供一个线程池来延迟或者定期执行任务

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

  • newSingleThreadScheduledExecutor:提供单个线程来延迟或者定期执行任务，如果执行的线程挂了，会生成新的。

      return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1));
    同样，它保证返回的Executor自身的线程数不可修改

从上述的实现可以看出，核心在于三个部分

  • ThreadPoolExecutor:提供线程数相关的控制
  • DelegatedExecutorService:仅暴露ExecutorService自身的方法，保证线程数不变来实现语义场景
  • ScheduledExecutorService：提供延迟或者定期执行的功能

对应的，相应也有不同的队列去实现不同的场景

  • LinkedBlockingQueue：无界阻塞队列
  • SynchronousQueue：没有消费者消费时，新的任务就会被阻塞
  • DelayQueue：队列中的任务过期之后才可以执行，否则无法查询到队列中的元素

DelegatedExecutorService

它仅仅是包装了ExecutorService的方法，交由传入的ExecutorService来执行，所谓的UnConfigurable实际也就是它没有暴漏配置各种参数调整的方法

  static class DelegatedExecutorService extends AbstractExecutorService {private final ExecutorService e;DelegatedExecutorService(ExecutorService executor) { e = executor; }public void execute(Runnable command) { e.execute(command); }public void shutdown() { e.shutdown(); }public List<Runnable> shutdownNow() { return e.shutdownNow(); }public boolean isShutdown() { return e.isShutdown(); }public boolean isTerminated() { return e.isTerminated(); }public boolean awaitTermination(long timeout, TimeUnit unit)throws InterruptedException {return e.awaitTermination(timeout, unit);}public Future<?> submit(Runnable task) {return e.submit(task);}public <T> Future<T> submit(Callable<T> task) {return e.submit(task);}public <T> Future<T> submit(Runnable task, T result) {return e.submit(task, result);}public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)throws InterruptedException {return e.invokeAll(tasks);}public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)throws InterruptedException {return e.invokeAll(tasks, timeout, unit);}public <T> T invokeAny(Collection<? extends Callable<T>> tasks)throws InterruptedException, ExecutionException {return e.invokeAny(tasks);}public <T> T invokeAny(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException {return e.invokeAny(tasks, timeout, unit);}}

ScheduledExecutorService

提供一系列的schedule方法，使得任务可以延迟或者周期性的执行，对应schedule方法会返回ScheduledFuture以供确认是否执行以及是否要取消。它的实现ScheduledThreadPoolExecutor也支持立即执行由submit提交的任务

仅支持相对延迟时间，比如距离现在5分钟后执行。类似Timer也可以管理延迟任务和周期任务，但是存在一些缺陷：

  • 所有的定时任务只有一个线程，如果某个任务执行时间长，将影响其它TimerTask的精确性。ScheduledExecutorService的多线程机制可弥补
  • TimerTask抛出未检查的异常，将终止线程执行，此时会错误的认为任务都取消了。1：可以使用try-catch-finally对相应执行快处理；2：通过execute执行的方法可以设置UncaughtExceptionHandler来接收未捕获的异常，并作出处理;3:通过submit执行的，将被封装层ExecutionException重新抛出

ThreadPoolExecutor

 public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

  • corePoolSize、maximumPoolSize：ThreadPoolExecutor会根据这两自动调整线程池的大小，当一个新任务通过execute提交的时候：
    如果当前运行的线程数小于corePoolSize就新建线程；
    如果当前线程数在corePoolSize与maximumPoolSize之间，则只有在队列满的时候才会创建新的线程；
    如果已经达到最大线程数，并且队列都满了，在这种饱和状态下就会执行拒绝策略

    默认情况下，只有新任务到达的时候才会启动线程，可通过prestartCoreThread方法实现事先启动

    1. corePoolSize：默认线程池所需要维护的最小的worker的数量，就算是worker过期了也会保留。如果想要不保留，则需要设置allowCoreThreadTimeOut，此时最小的就是0
    2. maximumPoolSize：线程池最大的线程数。java限制最多为 2^29-1,大约5亿个
  • keepAliveTime、unit：如果当前线程池有超过corePoolSize的线程数，只要有线程空闲时间超过keepAliveTime的设定，就会被终止；unit则是它的时间单位

  • workQueue：任何BlockingQueue都可以使用，基本上有三种

    1. Direct handoffs，直接交付任务。比如 SynchronousQueue，如果没有线程消费，提交任务会失败，当然可以新建一个线程来处理。它适合处理有依赖关系的任务，一般它的maximumPoolSizes会被设置成最大的
    2. Unbounded queues，无界队列。比如LinkedBlockingQueue，这意味着如果有corePoolSize个线程在执行，那么其他的任务都只能等待。它适合于处理任务都是互相独立的，
    3. Bounded queues，有界队列。比如ArrayBlockingQueue，需要考虑队列大小和最大线程数之间的关系，来达到更好的资源利用率和吞吐量
  • threadFactory：没有指定的时候，使用Executors.defaultThreadFactory

  • RejectedExecutionHandler：通过execute添加的任务，如果Executor已经关闭或者已经饱和了(线程数达到了maximumPoolSize,并且队列满了)，就会执行，java提供了4种策略：

    1. AbortPolicy，拒绝的时候抛出运行时异常RejectedExecutionException；
    2. CallerRunsPolicy，如果executor没有关闭，那么由调用execute的线程来执行它；
    3. DiscardPolicy，直接扔掉新的任务；
    4. DiscardOldestPolicy，如果executor没有关闭，那么扔掉队列头部的任务，再次尝试；
ThreadPoolExecutor可自定义beforeExecutor、afterExecutor可以用来添加日志统计、计时、件事或统计信息收集功能，无论run是正常返回还是抛出异常，afterExecutor都会被执行。如果beforeExecutor抛出RuntimeException,任务和afterExecutor都不会被执行。terminated在所有任务都已经完成，并且所有工作者线程关闭后会调用，此时也可以用来执行发送通知、记录日志等等。

如何估算线程池的大小

  1. 计算密集型，通常在拥有$N_{cpu}$个处理器的系统上，线程池大小设置为$N_{cpu}+1$能够实现最优的利用率；

    $N_{cpu}$ cpu的个数
  2. I/O密集型或者其它阻塞型的任务，定义 $N_{cpu}$为CPU的个数，$U_{cpu}$为CPU的利用率，$W/C$为等待时间与计算时间的比率，此时线程池的最优大小为

$$N_{threads}=N_{cpu}*U_{cpu}*(1+W/C)$$

场景说明

将一个网站的业务抽象成如下几块

  • 接收客户端请求与处理请求
  • 页面渲染返回的文本和图片
  • 获取页面的广告

接收请求与处理请求

理论模型

理论上，服务端通过实现约定的接口就可以实现接收请求和处理连续不断的请求过来

ServerSocket socket = new ServerSocket(80);
while(true){Socket conn = socket.accept();handleRequest(conn)
}

缺点：每次只能处理一个请求，新请求到来时，必须等到正在处理的请求处理完成，才能接收新的请求

显示的创建多线程

为每个请求创建新的线程提供服务

ServerSocket socket = new ServerSocket(80);
while(true){final Socket conn = socket.accept();Runnable task = new Runnable(){public void run(){handleRequest(conn);        }}new Thread(task).start();
}

缺点：

  • 线程的创建和销毁都有一定的开销，延迟对请求的处理；
  • 创建后的线程多于可用处理器的数量，造成线程闲置，这会给垃圾回收带来压力
  • 存活的大量线程竞争CPU资源会产生很多性能开销
  • 系统上对可创建的线程数存在限制

使用线程池

使用java自带的Executor框架。

private static final Executor exec = Executors.newFixedThreadPool(100);
...
ServerSocket socket = new ServerSocket(80);
while(true){final Socket conn = socket.accept();Runnable task = new Runnable(){public void run(){handleRequest(conn);        }}exec.execute(task);
}
...

线程池策略通过实现预估好的线程需求，限制并发任务的数量，重用现有的线程，解决每次创建线程的资源耗尽、竞争过于激烈和频繁创建的问题，也囊括了线程的优势，解耦了任务提交和任务执行。

页面渲染返回的文本和图片

串行渲染

renderText(source);
List<ImageData> imageData = new ArrayList<ImageData>();
for(ImageInfo info:scaForImageInfo(source)){imageData.add(info.downloadImage());
}
for(ImageData data:imageData){renderImage(data);
}

缺点：图像的下载大部分时间在等待I/O操作执行完成，这期间CPU几乎不做任何工作，使得用户看到最终页面之前要等待过长的时间

并行化

渲染过程可以分成两个部分，1是渲染文本，1是下载图像

private static final ExecutorService exec = Executors.newFixedThreadPool(100);
...
final List<ImageInfo> infos=scaForImageInfo(source);
Callable<List<ImageData>> task=new Callable<List<ImageData>>(){public List<ImageData> call(){List<ImageData> r = new ArrayList<ImageData>();for(ImageInfo info:infos){r.add(info.downloadImage());}return r;}
};
Future<List<ImageData>> future = exec.submit(task);
renderText(source);
try{List<ImageData> imageData = future.get();for(ImageData data:imageData){renderImage(data);}
}catch(InterruptedException e){Thread.currentThread().interrupt();future.cancel(true);
}catche(ExecutionException e){throw launderThrowable(e.getCause());
}

使用Callable来返回下载的图片结果，使用future来获得下载的图片，这样将减少用户所需要的等待时间。
缺点：图片的下载很明显时间要比文本要慢，这样的并行化很可能速度可能只提升了1%

并行性能提升

使用CompletionService。

private static final ExecutorService exec;
...
final List<ImageInfo> infos=scaForImageInfo(source);
CompletionService<ImageData> cService =  new ExecutorCompletionService<ImageData>(exec);
for(final ImageInfo info:infos){cService.submit(new Callable<ImageData>(){public ImageData call(){return info.downloadImage();}});
}
renderText(source);
try{for(int i=0,n=info.size();t<n;t++){Future<ImageData> f = cService.take();ImageData imageData=f.get();renderImage(imageData)}
}catch(InterruptedException e){Thread.currentThread().interrupt();
}catche(ExecutionException e){throw launderThrowable(e.getCause());
}

核心思路为为每一幅图像下载都创建一个独立的任务，并在线程池中执行他们，从而将串行的下载过程转换为并行的过程

获取页面的广告

广告展示如果在一定的时间以内没有获取，可以不再展示，并取消超时的任务。

 ExecutorService exe = Executors.newFixedThreadPool(3);List<MyTask> myTasks = new ArrayList<>();for (int i=0;i<3;i++){myTasks.add(new MyTask(3-i));}try {List<Future<String>> futures = exe.invokeAll(myTasks, 1, TimeUnit.SECONDS);for (int i=0;i<futures.size();i++){try {String s = futures.get(i).get();System.out.println("task execut "+myTasks.get(i).getSleepSeconds()+" s");} catch (ExecutionException e) {System.out.println("task sleep "+myTasks.get(i).getSleepSeconds()+" not execute ");}catch (CancellationException e){System.out.println("task sleep "+myTasks.get(i).getSleepSeconds()+" not execute ,because "+e);}}} catch (InterruptedException e) {e.printStackTrace();}exe.shutdown();

invokeAll方法对于没有完成的任务会被取消，通过CancellationException可以捕获，invokeAll返回的序列顺序和传入的task保持一致。结果如下：

task sleep 3 not execute ,because java.util.concurrent.CancellationException
task sleep 2 not execute ,because java.util.concurrent.CancellationException
task execut 1 s

http://lihuaxi.xjx100.cn/news/269856.html

相关文章

ASP.NET将原始图片按照指定尺寸等比例缩放显示图片

ASP.NET将原始图片按照指定尺寸等比例缩放显示图片

网站上可能会有很多图片&#xff0c;比如产品图片等&#xff0c;而且他们可能大小不一&#xff0c;宽度和高度也不一定一样&#xff0c;有的很大有的很小。如果放在一张网页上&#xff0c;可能会破坏版面&#xff0c;但是如果强制让他们按照指定的宽度和高度显示&#xff0c;因…
阅读更多...
2011-11-27

2011-11-27

大三&#xff0c;这个关键时刻&#xff0c;目前自己正在写软件课程设计程序-教材订购系统&#xff0c;加油&#xff01;努力&#xff01;转载于:https://blog.51cto.com/3122770/728090
阅读更多...
SpringBoot最最最常用的注解梳理

SpringBoot最最最常用的注解梳理

点击上方蓝色“方志朋”&#xff0c;选择“设为星标”回复“666”获取独家整理的学习资料&#xff01;来源&#xff1a;https://s.yam.com/w6mfR一、注解(annotations)列表SpringBootApplication&#xff1a; 包含了ComponentScan、Configuration和EnableAutoConfiguration注解…
阅读更多...
php开发之登录注册教程,PHP开发登录注册完整代码之注册PHP页面

php开发之登录注册教程,PHP开发登录注册完整代码之注册PHP页面

创建我们的 reg.php 文件此页面完成的功能如下代码如下<?php session_start();header("Content-type:text/html;charsetutf-8");$link mysqli_connect(localhost,root,root,login);mysqli_set_charset($link,utf8); //设定字符集$name$_POST[name];$pwd$_POST[p…
阅读更多...
抖音、快手和直播行业的火爆究竟给了谁机会？

抖音、快手和直播行业的火爆究竟给了谁机会？

经常收到一些CSDN小伙伴的留言&#xff0c;反馈如下这样的困惑“短视频这么火爆&#xff0c;我该学些什么技术才能入行&#xff1f;”“我想从事音视频开发&#xff0c;该如何入门和进阶&#xff1f;真的像坊间传闻的那么难吗&#xff1f;”音视频的开发前景做一个不恰当的比喻…
阅读更多...
第一视角：深度学习框架这几年

第一视角：深度学习框架这几年

点击上方“视学算法”&#xff0c;选择加"星标"或“置顶”重磅干货&#xff0c;第一时间送达‍‍‍‍‍‍‍‍作者 | Peter潘欣转自 | 夕小瑶的卖萌屋和深度学习框架打交道已有多年时间。从Google的TensorFlow, 到百度的PaddlePaddle&#xff0c;再到现在腾讯的无量。…
阅读更多...
教程 | 基于LSTM实现手写数字识别

教程 | 基于LSTM实现手写数字识别

点击上方“小白学视觉”&#xff0c;选择加"星标"或“置顶”重磅干货&#xff0c;第一时间送达基于tensorflow&#xff0c;如何实现一个简单的循环神经网络&#xff0c;完成手写数字识别&#xff0c;附完整演示代码。01 LSTM网络构建基于tensorflow实现简单的LSTM网络…
阅读更多...
C++11 带来的新特性 （2）—— 统一初始化（Uniform Initialization）

C++11 带来的新特性 （2）—— 统一初始化（Uniform Initialization）

1 统一初始化&#xff08;Uniform Initialization&#xff09; 在C 11之前&#xff0c;所有对象的初始化方式是不同的&#xff0c;经常让写代码的我们感到困惑。C 11努力创造一个统一的初始化方式。 其语法是使用{}和std::initializer_list&#xff0c;先看示例。 int values[]…
阅读更多...
最新文章

Copyright @ 2022~2023