How to create a Java thread pool

Benefits of thread pool

• It can realize the reuse of threads and avoid re-creating and destroying threads. Creating threads and destroying threads is very expensive for the CPU.

• You can limit the maximum number of threads that can be created, and dynamically adjust thread pool parameters according to your own machine performance to improve application performance.

• Provides functions such as scheduled execution and concurrency control.

• Unified management of threads.

Five ways to create a thread pool

1: Cache thread pool (not recommended)

2: Fixed capacity thread pool (not recommended)

3: Single thread pool (not recommended)

4: Scheduled task thread pool (not recommended)

5: Create a thread pool through the ThreadPoolExecutor construction method (developed by Alibaba The manual is highly recommended)

The first four ways to create a thread pool are all created through the static methods of Executors.

Cached Thread Pool CachedThreadPool

	ExecutorService executorService = Executors.newCachedThreadPool();

        for (int i = 0; i < 10; i++) {
            final int finalI = i;
            executorService.execute(new Runnable() {
                public void run() {
                    System.out.println(Thread.currentThread().getName()+"<thread->run>"+ finalI);
                }
            });
        }

Why is the cache thread pool not recommended?

Source code analysis

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, 2147483647, 60L, TimeUnit.SECONDS, new SynchronousQueue());
    }

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

Through the above two code snippets, we can see that the maximumPoolSize of CachedThreadPool is the maximum value of Integer 2147483647, which is equivalent to unlimited thread creation, and creating threads requires memory Yes, this will cause memory overflow, and ordinary machines do not use such a large memory to create such a large number of threads.

Fixed capacity thread pool FixedThreadPool

newFixedThreadPool(int num), num is the number of fixed threads we want to specify

	ExecutorService executorService = Executors.newFixedThreadPool(5);

      for (int i = 0; i < 10; i++) {
          final int finalI = i;
          executorService.execute(new Runnable() {
              public void run() {
                  System.out.println(Thread.currentThread().getName()+"<thread->run>"+ finalI);
              }
          });
      }

Output:

pool-1-thread-5run>4
pool-1-thread-4run>3
pool-1-thread-5run> 5
pool-1-thread-3run>2
pool-1-thread-3run>8
pool-1-thread-3run>9
pool-1-thread-2run>1
pool-1-thread-1run>0
pool-1-thread- 5run>7
pool-1-thread-4run>6

It can be seen that thread reuse is achieved.

Why is FixedThreadPool a fixed thread pool?

Source code analysis

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());
    }

It can be seen from this source code that the number of core threads (corePoolSize) and the maximum number of threads (maximumPoolSize) are both nThreads, because only in this way, the thread pool will not be expanded. , the number of threads is fixed.

Single thread pool SingleThreadExecutor

	ExecutorService executorService = Executors.newSingleThreadExecutor();

      for (int i = 0; i < 10; i++) {
          final int finalI = i;
          executorService.execute(new Runnable() {
              public void run() {
                  System.out.println(Thread.currentThread().getName()+"<thread->run>"+ finalI);
              }
          });

      }

Why does SingleThreadExecutor only contain one thread?

Source code analysis

public static ExecutorService newSingleThreadExecutor() {
        return new Executors.FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()));
    }

It can be seen from this source code that the number of core threads (corePoolSize) and the maximum number of threads (maximumPoolSize) are both 1, so it only contains one thread.

Scheduled ThreadPool

	  int initDelay=10; //初始化延时
      int period=1;//初始化延迟过了之后，每秒的延时

      ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);

      scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
          @Override
          public void run() {
              System.out.println(Thread.currentThread().getName()+"<thread->run>");
          }
      },initDelay,period, TimeUnit.SECONDS);

The effect of this code is: wait 10 seconds after the program runs, then output the first result, and then output the result every 1 second.

Why is ScheduledThreadPool not recommended?

Source code analysis

public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, 2147483647, 10L, TimeUnit.MILLISECONDS, new ScheduledThreadPoolExecutor.DelayedWorkQueue());
    }

It can be seen that the maximum number of threads (maximumPoolSize) of ScheduledThreadPool is the maximum value of Integer 2147483647, which is equivalent to unlimited creation of threads, and creating threads requires memory, so It will cause memory overflow, and ordinary machines do not use such a large memory to create such a large number of threads.

ThreadPoolExecutor creates a thread pool (highly recommended)

	ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 20,
              2L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5),
              Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());

      for (int i = 0; i < 12; i++) {
          final int finalI = i;
          threadPoolExecutor.execute(new Runnable() {
              public void run() {
                  System.out.println(Thread.currentThread().getName()+"<thread->run>"+ finalI);
              }
          });
      }

Detailed explanation of the seven parameters of ThreadPoolExecutor

	public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        
    }

• corePoolSize: the number of core threads. Once these threads are created, they will not be destroyed and will always exist. The thread pool has no threads by default. When a task arrives, a thread will be created through ThreadFactory and will always exist.

• maximumPoolSize: Maximum number of threads. The number of non-core threads = maximumPoolSize-corePoolSize. The number of non-core threads is actually the number of scalable threads and may be destroyed.

• keepAliveTime: The idle survival time of non-core threads. When the number of non-core threads generated through expansion is still idle after keepAliveTime, these non-core threads will be destroyed.

• unit: the time unit of keepAliveTime, for example: seconds

• workQueue: waiting area. When a task of >corePoolSize comes, the task will be stored in the blocking queue of workQueue, waiting for other threads to process it.

• threadFactory: Thread factory. A way to create threads.

• handler: Rejection strategy. When the capacity of >maximum number of threads workQueue is reached, the rejection policy will be executed

workQueue

ArrayBlockingQueue: bounded blocking queue. The queue has a size limit. When the capacity is exceeded, the expansion or rejection policy will be triggered.

	public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }

LinkedBlockingQueue: Unbounded blocking queue, the queue has no size limit, and may cause memory overflow.

	 public LinkedBlockingQueue() {
        this(2147483647);
    }

handler

AbortPolicy: Throw an exception directly

	public static class AbortPolicy implements RejectedExecutionHandler {
        public AbortPolicy() {
        }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString());
        }
    }

DiscardPolicy: No operation is performed. Silently discard the task

	public static class DiscardPolicy implements RejectedExecutionHandler {
        public DiscardPolicy() {
        }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

DiscardOldestPolicy: Discard the longest existing task

	public static class DiscardOldestPolicy implements RejectedExecutionHandler {
        public DiscardOldestPolicy() {
        }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }

        }
    }

CallerRunsPolicy: Let the thread that submitted the task process the task

	public static class CallerRunsPolicy implements RejectedExecutionHandler {
        public CallerRunsPolicy() {
        }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }

        }
    }

threadFactory

	ThreadFactory threadFactory = Executors.defaultThreadFactory();

      threadFactory.newThread(new Runnable() {
          @Override
          public void run() {
              System.out.println("threadFactory");
          }
      }).start();

如何触发拒绝策略和线程池扩容？

	ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 20,
              2L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5),
              Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());

      for (int i = 0; i < 26; i++) { //并发数26
          final int finalI = i;
          threadPoolExecutor.execute(new Runnable() {
              public void run() {
                  System.out.println(Thread.currentThread().getName()+"<thread->run>"+ finalI);
              }
          });
      }
      /**
       * 核心线程数=10，最大线程数=20，故可扩容线程数=20-10
       * BlockingQueue的大小为5，故等待区的大小为5，也就是当并发数<=核心线程数+5不会扩容，并发数大于16才会扩容
       *
       * 触发扩容：并发数>核心线程数+阻塞队列的大小
       * 对于这段代码，如果来了26个并发，10个并发会被核心线程处理，5个会在等待区，剩下11个会因为等待区满了而触发扩容
       * 因为这里最多能够扩容10个，这里却是11个，所以会触发拒绝策略
       */

• 为什么这段代码会触发拒绝策略

对于这段代码，如果来了26个并发，10个并发会被核心线程处理，5个会在等待区，剩下11个会因为等待区满了而触发扩容，但是又因为因为这里最多能够扩容10个，这里却是11个，所以会触发拒绝策略。

• 怎么触发扩容

触发扩容：并发数>核心线程数（corePoolSize）+阻塞队列（workQueue）的大小

• 使用Java纯手写一个线程池

The above is the detailed content of How to create a Java thread pool. For more information, please follow other related articles on the PHP Chinese website!