Einführung in die Verwendung des ThreadPoolExecutor-Thread-Pools in Java

Dieser Artikel bietet Ihnen eine Einführung in die Verwendung des ThreadPoolExecutor-Thread-Pools in Java. Ich hoffe, dass er für Freunde hilfreich ist.

Executors

Executors ist eine Toolklasse in Java. Bietet Factory-Methoden zum Erstellen verschiedener Arten von Thread-Pools.

Wie aus der obigen Abbildung ersichtlich ist, implementiert die Executors-Methode zum Erstellen von Thread-Pools alle die ExecutorService-Schnittstelle. Häufig verwendete Methoden sind wie folgt:

newFixedThreadPool(int Threads) : Erstellen Ein Thread-Pool mit einer festen Anzahl von Threads, und überschüssige Threads warten in der Warteschlange.

newCachedThreadPool(): Erstellen Sie einen zwischenspeicherbaren Thread-Pool. Wenn die Länge des Thread-Pools den Verarbeitungsbedarf übersteigt, können inaktive Threads vorhanden sein flexibel recycelt (60 Sekunden), wenn kein Recycling erfolgt, erstellen Sie einen neuen Thread

newSingleThreadExecutor(): Erstellen Sie einen Single-Threaded-Thread-Pool, der nur den einzigen Worker-Thread zum Ausführen von Aufgaben verwendet, um dies sicherzustellen Alle Aufgaben werden in der angegebenen Reihenfolge (FIFO, LIFO, Priorität) ausgeführt. Wenn ein Fehler bei der Aufgabenausführung auftritt, wird die Ausführung durch einen anderen Thread fortgesetzt.

newScheduledThreadPool(int corePoolSize): Erstellen Sie einen Thread-Pool, der geplante und periodische Aufgaben unterstützt Aufgabenausführung kann in den meisten Fällen verwendet werden, um die Timer-Klasse zu ersetzen Wir fügen n Aufgaben zum Thread-Pool hinzu. Diese n Aufgaben werden alle zusammen ausgeführt nach dieser Ausführung ausgeführt werden.

        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        cachedThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        Thread.currentThread().sleep(1000);
                        System.out.println(Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        cachedThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        Thread.currentThread().sleep(1000);
                        System.out.println(Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

newSingleThreadExecutor

Führen Sie jede Aufgabe nacheinander aus, bevor die nächste ausgeführt wird.

        ExecutorService cachedThreadPool = Executors.newFixedThreadPool(1);
        cachedThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        Thread.currentThread().sleep(1000);
                        System.out.println(Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        cachedThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        Thread.currentThread().sleep(1000);
                        System.out.println(Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

Was sind die Probleme? mit Executors

Im Alibaba Java Development Manual wird erwähnt, dass die Verwendung von Executors zum Erstellen eines Thread-Pools zu OOM (OutOfMemory, Speicherüberlauf) führen kann, es wird jedoch nicht erklärt, warum Schauen wir uns also das Ende an. Warum sind Executors nicht erlaubt?

Beginnen wir mit einem einfachen Beispiel, um die Situation zu simulieren, in der die Verwendung von Executoren OOM verursacht.

        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
        scheduledExecutorService.schedule(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        Thread.currentThread().sleep(2000);
                        System.out.println(Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }, 3, TimeUnit.SECONDS);

Führen Sie den obigen Code aus, indem Sie JVM-Parameter angeben: -Xmx8m -Xms8m, OOM wird ausgelöst:

        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        System.out.println(Thread.currentThread().getName());
                        Thread.currentThread().sleep(10000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        executorService.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        System.out.println(Thread.currentThread().getName());
                        Thread.currentThread().sleep(2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

Der obige Code weist darauf hin, dass Zeile 16 von ExecutorsDemo.java im Code executor.execute(new SubThread()) ist;

Es gibt zwei Hauptimplementierungen von BlockingQueue in Java, nämlich ArrayBlockingQueue und LinkedBlockingQueue.

ArrayBlockingQueue ist eine begrenzte Blockierungswarteschlange, die mit einem Array implementiert ist, und die Kapazität muss festgelegt werden.

/**
 * @author Hollis
 */
public class ExecutorsDemo {
    private static ExecutorService executor = Executors.newFixedThreadPool(15);
    public static void main(String[] args) {
        for (int i = 0; i LinkedBlockingQueue ist eine begrenzte Blockierungswarteschlange, die mit einer verknüpften Liste implementiert ist, und die Kapazität kann festgelegt werden optional festgelegt. Wenn nicht festgelegt, handelt es sich um eine unbegrenzte Blockierungswarteschlange mit einer maximalen Länge von Integer.MAX_VALUE.<p></p><pre class="brush:php;toolbar:false">Exception in thread "main" java.lang.OutOfMemoryError: GC overhead limit exceeded
    at java.util.concurrent.LinkedBlockingQueue.offer(LinkedBlockingQueue.java:416)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1371)
    at com.hollis.ExecutorsDemo.main(ExecutorsDemo.java:16)

Das Problem hierbei ist, dass, wenn wir die Kapazität von LinkedBlockingQueue nicht festlegen, die Standardkapazität beträgt Ganzzahl. MAX_VALUE.

Beim Erstellen von LinkedBlockingQueue wird keine Kapazität angegeben. Bei einer unbegrenzten Warteschlange können Aufgaben kontinuierlich zur Warteschlange hinzugefügt werden. Aufgrund zu vieler Aufgaben liegt möglicherweise ein Speicherüberlaufproblem vor.

Die maximale Anzahl der von newCachedThreadPool und newScheduledThreadPool erstellten Threads kann Integer.MAX_VALUE sein, und das Erstellen so vieler Threads ist zwangsläufig möglich Ursache OOM.

ThreadPoolExecutor erstellt einen Thread-Pool

Vermeiden Sie die Verwendung von Executoren zum Erstellen eines Thread-Pools, hauptsächlich um die Verwendung der Standardimplementierung zu vermeiden. Dann können wir den Konstruktor von ThreadPoolExecutor direkt aufrufen, um den Thread-Pool selbst zu erstellen Geben Sie beim Erstellen einfach die Kapazität für BlockQueue an.

public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity  In diesem Fall wird java.util.concurrent.RejectedExecutionException ausgelöst, weil die Warteschlange Derzeit wird vom Thread-Pool eine begrenzte Warteschlange verwendet. Wenn die Warteschlange voll ist, können keine neuen Anforderungen mehr verarbeitet werden.<p></p>Zusätzlich zur Definition von ThreadPoolExecutor gibt es noch andere Methoden wie Apache und Guava.<p></p>Vier Konstruktoren<p></p><pre class="brush:php;toolbar:false">public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}

int corePoolSize => Die maximale Anzahl von Kernthreads im Thread-Pool

Wenn der Thread-Pool einen neuen Thread erstellt, wenn die aktuelle Gesamtzahl der Threads geringer ist Als corePoolSize wird der neue Thread erstellt. Wenn er corePoolSize überschreitet, bleibt der neu erstellte Nicht-Kern-Thread standardmäßig immer im Thread-Pool erhalten, auch wenn der Kern vorhanden ist Thread macht nichts (Ruhezustand).

如果指定 ThreadPoolExecutor 的 allowCoreThreadTimeOut 这个属性为 true, 那么核心线程如果不干活(闲置状态)的话, 超过一定时间(时长下面参数决定), 就会被销毁掉

很好理解吧, 正常情况下你不干活我也养你, 因为我总有用到你的时候, 但有时候特殊情况(比如我自己都养不起了), 那你不干活我就要把你干掉了

int maximumPoolSize
该线程池中线程总数最大值

线程总数 = 核心线程数 + 非核心线程数.

long keepAliveTime
该线程池中非核心线程闲置超时时长

一个非核心线程, 如果不干活(闲置状态)的时长超过这个参数所设定的时长, 就会被销毁掉

如果设置 allowCoreThreadTimeOut = true, 则会作用于核心线程

TimeUnit unit

keepAliveTime的单位, TimeUnit是一个枚举类型, 其包括:

TimeUnit.DAYS;               //天
TimeUnit.HOURS;             //小时
TimeUnit.MINUTES;           //分钟
TimeUnit.SECONDS;           //秒
TimeUnit.MILLISECONDS;      //毫秒
TimeUnit.MICROSECONDS;      //微妙
TimeUnit.NANOSECONDS;       //纳秒

BlockingQueue workQueue

一个阻塞队列, 用来存储等待执行的任务. 也就是说现在有10个任务, 核心线程 有四个, 非核心线程有六个, 那么这六个线程会被添加到 workQueue 中, 等待执行.

这个参数的选择也很重要, 会对线程池的运行过程产生重大影响, 一般来说, 这里的阻塞队列有以下几种选择:

SynchronousQueue: 这个队列接收到任务的时候, 会直接提交给线程处理, 而不保留它, 如果所有线程都在工作怎么办？ 那就*新建一个线程来处理这个任务！所以为了保证不出现的错误, 使用这个类型队列的时候, maximumPoolSize 一般指定成 Integer.MAX_VALUE, 即无限大.

LinkedBlockingQueue: 这个队列接收到任务的时候, 如果当前线程数小于核心线程数, 则核心线程处理任务; 如果当前线程数等于核心线程数, 则进入队列等待. 由于这个队列最大值为 Integer.MAX_VALUE , 即所有超过核心线程数的任务都将被添加到队列中，这也就导致了 maximumPoolSize 的设定失效, 因为总线程数永远不会超过 corePoolSize.

ArrayBlockingQueue: 可以限定队列的长度, 接收到任务的时候, 如果没有达到 corePoolSize 的值, 则核心线程执行任务, 如果达到了, 则入队等候, 如果队列已满, 则新建线程(非核心线程)执行任务, 又如果总线程数到了maximumPoolSize, 并且队列也满了, 则发生错误.

DelayQueue: 队列内元素必须实现 Delayed 接口, 这就意味着你传进去的任务必须先实现Delayed接口. 这个队列接收到任务时, 首先先入队, 只有达到了指定的延时时间, 才会执行任务.

ThreadFactory threadFactory

它是ThreadFactory类型的变量, 用来创建新线程.

默认使用 Executors.defaultThreadFactory() 来创建线程. 使用默认的 ThreadFactory 来创建线程时, 会使新创建的线程具有相同的 NORM_PRIORITY 优先级并且是非守护线程, 同时也设置了线程的名称.

RejectedExecutionHandler handler

表示当拒绝处理任务时的策略, 有以下四种取值:

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常(默认).
ThreadPoolExecutor.DiscardPolicy：直接丢弃任务, 但是不抛出异常.
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务, 然后重新尝试执行任务(重复此过程)
ThreadPoolExecutor.CallerRunsPolicy：用调用者所在的线程来执行任务.

【相关推荐：Java视频教程】

Das obige ist der detaillierte Inhalt vonEinführung in die Verwendung des ThreadPoolExecutor-Thread-Pools in Java. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!