专栏首页IT云清线程池的创建和使用

线程池的创建和使用

几种线程池的创建和使用

目录:

  • 1.newFixedThreadPool固定线程池
  • 2.newSingleThreadExecutor一个线程的线程池
  • 3.newCachedThreadPool缓存线程池
  • 4.ThreadPoolExecutor
  • 5.Future获取返回结果

1.newFixedThreadPool固定线程池

示例:

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(4);

源码:

    /**
     * Creates a thread pool that reuses a fixed number of threads
     * operating off a shared unbounded queue.  At any point, at most
     * {@code nThreads} threads will be active processing tasks.
     * If additional tasks are submitted when all threads are active,
     * they will wait in the queue until a thread is available.
     * If any thread terminates due to a failure during execution
     * prior to shutdown, a new one will take its place if needed to
     * execute subsequent tasks.  The threads in the pool will exist
     * until it is explicitly {@link ExecutorService#shutdown shutdown}.
     *
     * @param nThreads the number of threads in the pool
     * @return the newly created thread pool
     * @throws IllegalArgumentException if {@code nThreads <= 0}
     */
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

解读一下源码注释:

创建一个线程池,该线程池复用固定数量的线程去操作一个共享的无界队列;
在任何时刻,最多只有nThreads的线程是处于可处理任务的活跃状态。
当所有的线程都处于活跃状态(在处理任务),如果提交了额外的任务,它将会在队列中等待,直到有线程可用。
如果线程在执行期间由于失败而终止,如果需要的话,一个新的线程将会取代它执行后续任务。
线程池中的线程将会一直存在,直到显示的关闭。

这里需要注意,线程的数量是固定的,但是队列大小是无界的(Integer.MAX_VALUE足够大,大到可以任务无界。)

注意这里用的队列:LinkedBlockingQueue,默认队列大小Integer的最大值。

    /**
     * Creates a {@code LinkedBlockingQueue} with a capacity of
     * {@link Integer#MAX_VALUE}.
     */
    public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

代码测试:

package com.java4all.test11;

import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;


/**
 * Author: yunqing
 * Date: 2018/9/19
 * Description:
 */
@RestController
@RequestMapping(value = "testThread")
public class TestThread {


    /**固定大小的线程池*/
    static ExecutorService fixedThreadPool = Executors.newFixedThreadPool(4);
    
    @RequestMapping(value = "parse",method = RequestMethod.GET)
    public String parse(){
        Future<?> submit = fixedThreadPool.submit(() -> {
            System.out.println("线程名称:" + Thread.currentThread().getName());
        });
        return null;
    }
}

我们设定固定大小为4的线程池,然后用20个并发请求:

线程名称:pool-1-thread-1
线程名称:pool-1-thread-3
线程名称:pool-1-thread-4
线程名称:pool-1-thread-1
线程名称:pool-1-thread-3
线程名称:pool-1-thread-4
线程名称:pool-1-thread-4
线程名称:pool-1-thread-3
线程名称:pool-1-thread-4
线程名称:pool-1-thread-1
线程名称:pool-1-thread-1
线程名称:pool-1-thread-4
线程名称:pool-1-thread-1
线程名称:pool-1-thread-3
线程名称:pool-1-thread-4
线程名称:pool-1-thread-1
线程名称:pool-1-thread-3
线程名称:pool-1-thread-2
线程名称:pool-1-thread-2
线程名称:pool-1-thread-1

由是于队列无界,200 000个也是可以的,但是处理复杂任务时,无界队列可能会让内存爆掉。

2.newSingleThreadExecutor单线程

示例:

static ExecutorService service = Executors.newSingleThreadExecutor();

源码:

    /**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue. (Note however that if this single
     * thread terminates due to a failure during execution prior to
     * shutdown, a new one will take its place if needed to execute
     * subsequent tasks.)  Tasks are guaranteed to execute
     * sequentially, and no more than one task will be active at any
     * given time. Unlike the otherwise equivalent
     * {@code newFixedThreadPool(1)} the returned executor is
     * guaranteed not to be reconfigurable to use additional threads.
     *
     * @return the newly created single-threaded Executor
     */
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

解读一下源码注释:

创建一个线程执行者,它使用单个线程去操作一个无界队列。
(需要注意:如果一个线程由于执行过程中失败导致线程终止,一个新的线程将会取代他,如果需要执行后续任务)

这里使用的队列,也是LinkedBlockingQueue,需要注意。

3.newCachedThreadPool缓存线程池

示例

static ExecutorService service1 = Executors.newCachedThreadPool();

源码:

    /**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available.  These pools will typically improve the performance
     * of programs that execute many short-lived asynchronous tasks.
     * Calls to {@code execute} will reuse previously constructed
     * threads if available. If no existing thread is available, a new
     * thread will be created and added to the pool. Threads that have
     * not been used for sixty seconds are terminated and removed from
     * the cache. Thus, a pool that remains idle for long enough will
     * not consume any resources. Note that pools with similar
     * properties but different details (for example, timeout parameters)
     * may be created using {@link ThreadPoolExecutor} constructors.
     *
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

源码解读:

这里,使用的是异步队列SynchronousQueue,而且是非公平的。

查看下源码:

    /**
     * Creates a {@code SynchronousQueue} with nonfair access policy.
     */
    public SynchronousQueue() {
        this(false);
    }
    /**
     * Creates a {@code SynchronousQueue} with the specified fairness policy.
     *
     * @param fair if true, waiting threads contend in FIFO order for
     *        access; otherwise the order is unspecified.
     如果为真,等待线程按照FIFO顺序,否则,顺序是不确定的。
     */
    public SynchronousQueue(boolean fair) {
        transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
    }

测试一下:

/**
 * Author: yunqing
 * Date: 2018/9/19
 * Description:
 */
@RestController
@RequestMapping(value = "testThread")
public class TestThread {


    /***/
    static ExecutorService service1 = Executors.newCachedThreadPool();
    @RequestMapping(value = "parse",method = RequestMethod.GET)
    public String parse(){
        Future<?> submit = service1.submit(() -> {
            System.out.println("线程名称:" + Thread.currentThread().getName());
        });
        return null;
    }
}

20个并发请求:

线程名称:pool-1-thread-13
线程名称:pool-1-thread-15
线程名称:pool-1-thread-13
线程名称:pool-1-thread-2
线程名称:pool-1-thread-15
线程名称:pool-1-thread-17
线程名称:pool-1-thread-4
线程名称:pool-1-thread-14
线程名称:pool-1-thread-12
线程名称:pool-1-thread-4
线程名称:pool-1-thread-8
线程名称:pool-1-thread-7
线程名称:pool-1-thread-1
线程名称:pool-1-thread-6
线程名称:pool-1-thread-3
线程名称:pool-1-thread-5
线程名称:pool-1-thread-11
线程名称:pool-1-thread-10
线程名称:pool-1-thread-9
线程名称:pool-1-thread-16

注意看,是有线程复用的。

4.ThreadPoolExecutor

4.1这种方式创建线程池,参数很多,由于可以显示指定队列的大小,所以可以合理避免OOM;
4.2拒绝策略
  • AbortPolicy:抛出RejectedExecutionException异常。这种策略时,注意捕获异常。
  • DiscardPolicy:什么也不做,直接忽略
  • DiscardOldestPolicy:丢弃执行队列中最老的任务,尝试为当前提交的任务腾出位置
  • CallerRunsPolicy:直接由提交任务者执行这个任务
/**
 * Author: yunqing
 * Date: 2018/9/19
 * Description:
 */
@RestController
@RequestMapping(value = "testThread")
public class TestThread {


    static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            4,  //corePoolSize:线程核心数量,及时处于idle状态,也不会回收
            10, //maximumPoolSize:线程数的上限
            60, //keepAliveTime:超过这个时间,超过corePoolSize的线程,多余的线程将会被回收
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(300), //任务的排队队列  不显示指定大小,会默认Integer.MAX_VALUE,设置不当容易OOM
            new ThreadPoolExecutor.AbortPolicy()  //拒绝策略
    );
    @RequestMapping(value = "parse",method = RequestMethod.GET)
    public String parse(){
        Future<?> submit = threadPoolExecutor.submit(() -> {
            System.out.println("线程名称:" + Thread.currentThread().getName());
        });
        return "处理完了";
    }
}

5.Future获取返回结果

package com.java4all.test11;

import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.*;


/**
 * Author: yunqing
 * Date: 2018/9/19
 * Description:
 */
@RestController
@RequestMapping(value = "testThread")
public class TestThread {


    static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            4,  //corePoolSize:线程核心数量,及时处于idle状态,也不会回收
            10, //maximumPoolSize:线程数的上限
            60, //keepAliveTime:超过这个时间,超过corePoolSize的线程,多余的线程将会被回收
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(200), //任务的排队队列  不显示指定大小,会默认Integer.MAX_VALUE,设置不当容易OOM
            new ThreadPoolExecutor.AbortPolicy()  //拒绝策略
    );
    @RequestMapping(value = "parse",method = RequestMethod.GET)
    public String parse(){
        Future<?> future = threadPoolExecutor.submit(() -> {
            Thread.sleep(10000);
            System.out.println("线程名称:" + Thread.currentThread().getName());
            return Thread.currentThread().getName()+":处理完了数据";
        });

        Object o = null;
        try {
            //获取执行结果  如果设置了时间,规定时间内没有处理完,返回结果可能为空
            o = future.get(4,TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        System.out.println(o.toString());
        return "处理完了";
    }
}

future.get()方法,是一个阻塞方法,会等到线程任务执行完了,返回结果;如果设置了时间,会等待到设定的时间,如果超市了还没有计算完成,返回结果可能为null;所以,这里需要注意NPE;

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Java多线程系列(三)-----线程池

    https://blog.csdn.net/weixin_39800144/article/details/77803751

    IT云清
  • java 主线程等待子线程执行完后再执行

    示例如下,我们初始化一个CountDownLatch,值为10(子线程个数),然后每次一个子线程执行完后执行一下countDown(),代码示例如下:

    IT云清
  • 一个线程start多次

    IT云清
  • 线程池

    创建一个可缓存线程池,如果线程池长度超过处理需要,可灵活回收空闲线程,若无可回收,则新建线程。示例代码如下:

    希希里之海
  • 初学者第68节多线程之线程池(十一)

    线程池是一种多线程处理形式,处理过程中将任务添加到队列,然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小,以默认的优先级运...

    用户5224393
  • java 基础

    从上面代码中可以看出LinkedList迭代器的next函数只是通过next指针快速得到下一个元素并返回。而get方法会从头遍历直到index下标,查找一个元素...

    大学里的混子
  • Java线程(四):线程中断、线程让步、线程睡眠、线程合并

    最近在Review线程专栏,修改了诸多之前描述不够严谨的地方,凡是带有Review标记的文章都是修改过了。本篇文章是插进来的,因为原来没有写,现在...

    高爽
  • FutureTask——另一种闭锁的实现

    关于多线程,我们接触对多的,最基础,入门的可能就是实现Runnable接口继承Thead类,因为Java单继承的原因,通常建议是实现Runnable接口。但这种...

    用户1148394
  • java多线程系列:Executors框架

    Executor是一个接口,里面提供了一个execute方法,该方法接收一个Runable参数,如下

    云枭
  • 【RT-Thread笔记】临界区问题及IPC机制

    在多线程实时系统中,多个线程操作/访问同一块区域(代码),这块代码就称为临界区。 例如一项工作中的两个线程:一个线程从传感器中接收数据并且将数据写到共享内存中...

    正念君

扫码关注云+社区

领取腾讯云代金券