Tidak mungkin membuat pool thread dalam cache dengan batas ukuran?

127

Tampaknya mustahil untuk membuat kumpulan thread dalam cache dengan batasan jumlah thread yang dapat dibuat.

Berikut adalah bagaimana static Executors.newCachedThreadPool diimplementasikan di perpustakaan Java standar:

 public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

Jadi, gunakan templat itu untuk terus membuat kumpulan utas cache berukuran tetap:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new SynchronusQueue<Runable>());

Sekarang jika Anda menggunakan ini dan mengirimkan 3 tugas, semuanya akan baik-baik saja. Menyerahkan tugas lebih lanjut akan menghasilkan pengecualian eksekusi yang ditolak.

Mencoba ini:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runable>());

Akan menghasilkan semua utas mengeksekusi berurutan. Yaitu, kumpulan utas tidak akan pernah membuat lebih dari satu utas untuk menangani tugas Anda.

Ini adalah bug dalam metode eksekusi ThreadPoolExecutor? Atau mungkin ini disengaja? Atau ada cara lain?

Sunting: Saya ingin sesuatu persis seperti kumpulan thread yang di-cache (membuat thread berdasarkan permintaan dan kemudian membunuhnya setelah beberapa waktu habis) tetapi dengan batas jumlah thread yang dapat dibuat dan kemampuan untuk terus mengantre tugas-tugas tambahan setelah memiliki tekan batas utasnya. Menurut tanggapan Sjlee, ini tidak mungkin. Melihat metode execute () dari ThreadPoolExecutor memang tidak mungkin. Saya perlu subclass ThreadPoolExecutor dan menimpa mengeksekusi () agak seperti SwingWorker tidak, tetapi apa yang dilakukan SwingWorker dalam mengeksekusi () adalah hack lengkap.

Matt Crinklaw-Vogt
sumber
1
Apa pertanyaan Anda? Bukankah kode 2 Anda memberi jawaban untuk judul Anda?
rsp
4
Saya ingin kumpulan utas yang akan menambahkan utas sesuai permintaan saat jumlah tugas bertambah, tetapi tidak akan pernah menambahkan lebih dari beberapa utas maksimal. CachedThreadPool sudah melakukan ini, kecuali itu akan menambah jumlah utas yang tidak terbatas dan tidak berhenti pada ukuran yang sudah ditentukan sebelumnya. Ukuran yang saya definisikan dalam contoh adalah 3. Contoh kedua menambahkan 1 utas, tetapi tidak menambahkan dua lagi saat tugas baru tiba sementara tugas lainnya belum selesai.
Matt Crinklaw-Vogt
Check this, itu memecahkan itu, debuggingisfun.blogspot.com/2012/05/...
Ethan

Jawaban:

235

ThreadPoolExecutor memiliki beberapa perilaku utama berikut, dan masalah Anda dapat dijelaskan oleh perilaku ini.

Ketika tugas diserahkan,

  1. Jika kumpulan thread belum mencapai ukuran inti, itu menciptakan utas baru.
  2. Jika ukuran inti telah tercapai dan tidak ada utas menganggur, itu mengantri tugas.
  3. Jika ukuran inti telah tercapai, tidak ada utas menganggur, dan antrian menjadi penuh, ia menciptakan utas baru (hingga mencapai ukuran maksimum).
  4. Jika ukuran maks telah tercapai, tidak ada utas menganggur, dan antrian menjadi penuh, kebijakan penolakan muncul.

Dalam contoh pertama, perhatikan bahwa SynchronousQueue pada dasarnya memiliki ukuran 0. Oleh karena itu, saat Anda mencapai ukuran maksimal (3), kebijakan penolakan akan muncul di (# 4).

Pada contoh kedua, antrian pilihan adalah LinkedBlockingQueue yang memiliki ukuran tidak terbatas. Karena itu, Anda terjebak dengan perilaku # 2.

Anda tidak dapat benar-benar bermain-main dengan tipe cache atau tipe tetap, karena perilaku mereka hampir sepenuhnya ditentukan.

Jika Anda ingin memiliki kumpulan ulir yang dibatasi dan dinamis, Anda perlu menggunakan ukuran inti positif dan ukuran maks dikombinasikan dengan antrian ukuran terbatas. Sebagai contoh,

new ThreadPoolExecutor(10, // core size
    50, // max size
    10*60, // idle timeout
    TimeUnit.SECONDS,
    new ArrayBlockingQueue<Runnable>(20)); // queue with a size

Tambahan : ini adalah jawaban yang cukup lama, dan tampaknya JDK mengubah perilakunya ketika mengenai ukuran inti 0. Sejak JDK 1.6, jika ukuran inti adalah 0 dan kumpulan tidak memiliki utas, ThreadPoolExecutor akan menambahkan utas untuk menjalankan tugas itu. Oleh karena itu, ukuran inti 0 adalah pengecualian dari aturan di atas. Terima kasih Steve untuk membawa yang menjadi perhatian saya.

sjlee
sumber
4
Anda harus menulis beberapa kata tentang metode allowCoreThreadTimeOutuntuk membuat jawaban ini sempurna. Lihat jawaban @ user1046052
hsestupin
1
Jawaban bagus! Hanya satu poin untuk ditambahkan: Kebijakan penolakan lainnya juga layak disebutkan. Lihat jawaban @brianegge
Jeff
1
Bukankah seharusnya perilaku 2 mengatakan 'Jika ukuran maxThread telah tercapai dan tidak ada utas menganggur, itu akan mengantri tugas.' ?
Zoltán
1
Bisakah Anda menguraikan tentang ukuran antrian? Apakah ini berarti bahwa hanya 20 tugas yang dapat diantri sebelum ditolak?
Zoltán
1
@ Zoltán saya menulis ini beberapa waktu yang lalu, jadi ada kemungkinan beberapa perilaku mungkin telah berubah sejak saat itu (saya tidak mengikuti kegiatan terbaru terlalu dekat), tetapi dengan asumsi perilaku ini tidak berubah, # 2 benar seperti yang dinyatakan, dan itu mungkin poin paling penting (dan agak mengejutkan) dari ini. Setelah ukuran inti tercapai, TPE lebih suka mengantri daripada membuat utas baru. Ukuran antrian secara harfiah adalah ukuran antrian yang diteruskan ke TPE. Jika antrian menjadi penuh tetapi belum mencapai ukuran maksimal, itu akan membuat utas baru (tidak menolak tugas). Lihat # 3. Semoga itu bisa membantu.
sjlee
60

Kecuali saya melewatkan sesuatu, solusi untuk pertanyaan awal itu sederhana. Kode berikut mengimplementasikan perilaku yang diinginkan seperti yang dijelaskan oleh poster asli. Ini akan memunculkan hingga 5 utas untuk bekerja pada antrian tanpa batas dan utas menganggur akan berakhir setelah 60 detik.

tp = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<Runnable>());
tp.allowCoreThreadTimeOut(true);

sumber
1
Anda benar. Metode itu ditambahkan dalam jdk 1.6, jadi tidak banyak orang tahu tentang itu. juga, Anda tidak dapat memiliki ukuran kolam inti "min", yang sangat disayangkan.
jtahlborn
4
Satu-satunya kekhawatiran saya tentang ini adalah (dari dokumen JDK 8): "Ketika tugas baru diajukan dalam metode eksekusi (Runnable), dan lebih sedikit dari corePoolSize menjalankan thread, utas baru dibuat untuk menangani permintaan, bahkan jika pekerja lain utas tidak digunakan. "
veegee
Cukup yakin ini tidak benar-benar berfungsi. Terakhir kali saya melihat melakukan hal di atas sebenarnya hanya pernah menjalankan pekerjaan Anda dalam satu utas meskipun Anda menelurkan 5. Sekali lagi, sudah beberapa tahun tetapi ketika saya terjun ke dalam implementasi ThreadPoolExecutor itu hanya dikirim ke utas baru setelah antrian Anda penuh. Menggunakan antrian tanpa batas menyebabkan ini tidak pernah terjadi. Anda dapat menguji dengan mengirimkan pekerjaan dan masuk ke nama utas lalu tidur. Setiap runnable pada akhirnya akan mencetak nama yang sama / tidak dapat dijalankan pada utas lainnya.
Matt Crinklaw-Vogt
2
Ini berhasil, Matt. Anda mengatur ukuran inti ke 0, itu sebabnya Anda hanya punya 1 utas. Kuncinya di sini adalah mengatur ukuran inti ke ukuran maksimal.
T-Gergely
1
@ vegee benar - Ini sebenarnya tidak berfungsi dengan baik - ThreadPoolExecutor hanya akan menggunakan kembali utas saat di atas corePoolSize. Jadi ketika corePoolSize sama dengan maxPoolSize, Anda hanya akan mendapat manfaat dari caching thread ketika pool Anda penuh (Jadi, jika Anda bermaksud untuk menggunakan ini tetapi biasanya tetap di bawah ukuran max pool Anda, Anda mungkin juga mengurangi timeout thread ke rendah nilai, dan perlu diketahui bahwa tidak ada caching - selalu ada utas baru)
Chris Riddell
7

Punya masalah yang sama. Karena tidak ada jawaban lain yang menyatukan semua masalah, saya menambahkan milik saya:

Sekarang ditulis dengan jelas dalam dokumen : Jika Anda menggunakan antrian yang tidak memblokir ( LinkedBlockingQueue) pengaturan utas maksimum tidak berpengaruh, hanya utas inti yang digunakan.

begitu:

public class MyExecutor extends ThreadPoolExecutor {

    public MyExecutor() {
        super(4, 4, 5,TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
        allowCoreThreadTimeOut(true);
    }

    public void setThreads(int n){
        setMaximumPoolSize(Math.max(1, n));
        setCorePoolSize(Math.max(1, n));
    }

}

Pelaksana ini memiliki:

  1. Tidak ada konsep utas maksimal karena kami menggunakan antrian tanpa batas. Ini adalah hal yang baik karena antrian tersebut dapat menyebabkan pelaksana membuat sejumlah besar utas tambahan non-inti jika mengikuti kebijakan yang biasa.

  2. Antrian ukuran maks Integer.MAX_VALUE. Submit()akan melempar RejectedExecutionExceptionjika jumlah tugas yang tertunda melebihi Integer.MAX_VALUE. Tidak yakin kami akan kehabisan memori terlebih dahulu atau ini akan terjadi.

  3. Memiliki 4 utas inti yang mungkin. Thread inti idle secara otomatis keluar jika idle selama 5 detik setThreads().

  4. Memastikan jumlah minimum utas inti tidak pernah kurang dari satu, atau yang lainnya submit()akan menolak setiap tugas. Karena utas inti harus> = utas maks, metode ini juga setThreads()menetapkan utas maksimum, meskipun pengaturan utas maksimum tidak berguna untuk antrian tanpa batas.

SD
sumber
Saya pikir Anda juga perlu mengatur 'allowCoreThreadTimeOut' menjadi 'true', jika tidak, setelah utas dibuat, Anda akan menyimpannya selamanya: gist.github.com/ericdcobb/46b817b384f5ca9d5f5d
eric
oops Saya baru saja melewatkan itu, maaf, jawaban Anda sempurna kalau begitu!
eric
6

Dalam contoh pertama Anda, tugas berikutnya ditolak karena AbortPolicyini adalah default RejectedExecutionHandler. ThreadPoolExecutor berisi kebijakan berikut, yang dapat Anda ubah melalui setRejectedExecutionHandlermetode:

CallerRunsPolicy
AbortPolicy
DiscardPolicy
DiscardOldestPolicy

Kedengarannya seperti Anda ingin kolam thread di-cache dengan CallerRunsPolicy.

brianegge
sumber
5

Tidak ada jawaban di sini yang memperbaiki masalah saya, yang berkaitan dengan membuat koneksi HTTP dalam jumlah terbatas menggunakan klien HTTP Apache (versi 3.x). Karena saya butuh beberapa jam untuk mencari tahu pengaturan yang bagus, saya akan membagikan:

private ExecutorService executor = new ThreadPoolExecutor(5, 10, 60L,
  TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
  Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());

Ini menciptakan ThreadPoolExecutoryang dimulai dengan lima dan memegang maksimal sepuluh utas yang berjalan secara simultan CallerRunsPolicyuntuk mengeksekusi.

paul
sumber
Masalah dengan solusi ini adalah jika Anda menambah jumlah atau produsen, Anda akan menambah jumlah utas yang menjalankan utas latar. Dalam banyak kasus, itu bukan yang Anda inginkan.
Gray
3

Per Javadoc untuk ThreadPoolExecutor:

Jika ada lebih dari corePoolSize tetapi kurang dari maksimum threadPoolSize berjalan, utas baru akan dibuat hanya jika antrian penuh . Dengan mengatur corePoolSize dan maksimumPoolSize yang sama, Anda membuat kumpulan utas ukuran tetap.

(Penekanan milikku.)

Jawaban jitter adalah apa yang Anda inginkan, meskipun jawaban saya menjawab pertanyaan Anda yang lain. :)

Jonathan Feinberg
sumber
2

ada satu opsi lagi. Alih-alih menggunakan SynchronousQueue baru Anda juga dapat menggunakan antrian lain, tetapi Anda harus memastikan ukurannya adalah 1, sehingga akan memaksa layanan executorser untuk membuat utas baru.

Ashkrit
sumber
Saya pikir maksud Anda ukuran 0 (secara default), sehingga tidak akan ada tugas yang antri dan benar-benar memaksa layanan executors untuk membuat utas baru setiap saat.
Leonmax
2

Tidak terlihat seolah-olah salah satu jawaban benar-benar menjawab pertanyaan - pada kenyataannya saya tidak bisa melihat cara melakukan ini - bahkan jika Anda subkelas dari PooledExecutorService karena banyak metode / properti bersifat pribadi misalnya membuat addIfUnderMaximumPoolSize dilindungi Anda bisa lakukan hal berikut:

class MyThreadPoolService extends ThreadPoolService {
    public void execute(Runnable run) {
        if (poolSize() == 0) {
            if (addIfUnderMaximumPoolSize(run) != null)
                return;
        }
        super.execute(run);
    }
}

Yang paling dekat yang saya dapatkan adalah ini - tetapi bahkan itu bukan solusi yang sangat baik

new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()) {
    public void execute(Runnable command) {
        if (getPoolSize() == 0 && getActiveCount() < getMaximumPoolSize()) {        
            super.setCorePoolSize(super.getCorePoolSize() + 1);
        }
        super.execute(command);
    }

    protected void afterExecute(Runnable r, Throwable t) {
         // nothing in the queue
         if (getQueue().isEmpty() && getPoolSize() > min) {
             setCorePoolSize(getCorePoolSize() - 1);
         }
    };
 };

ps tidak menguji di atas

Stuart
sumber
2

Ini solusi lain. Saya pikir solusi ini berperilaku seperti yang Anda inginkan (meskipun tidak bangga dengan solusi ini):

final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {
    public boolean offer(Runnable o) {
        if (size() > 1)
            return false;
        return super.offer(o);
    };

    public boolean add(Runnable o) {
        if (super.offer(o))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }
};

RejectedExecutionHandler handler = new RejectedExecutionHandler() {         
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        queue.add(r);
    }
};

dbThreadExecutor =
        new ThreadPoolExecutor(min, max, 60L, TimeUnit.SECONDS, queue, handler);
Stuart
sumber
2

Ini yang Anda inginkan (setidaknya saya kira begitu). Untuk penjelasan, periksa jawaban Jonathan Feinberg

Executors.newFixedThreadPool(int n)

Membuat kumpulan utas yang menggunakan kembali sejumlah utas yang beroperasi dari antrian yang tidak dibatasi bersama. Pada titik mana pun, paling banyak nThreads threads akan menjadi tugas pemrosesan aktif. Jika tugas tambahan diajukan ketika semua utas aktif, mereka akan menunggu dalam antrian sampai utas tersedia. Jika ada utas yang berakhir karena kegagalan selama eksekusi sebelum shutdown, yang baru akan menggantikannya jika diperlukan untuk menjalankan tugas selanjutnya. Utas di kumpulan akan ada sampai ditutup secara eksplisit.

naik opelet
sumber
4
Tentu, saya bisa menggunakan kolam utas tetap tetapi itu akan meninggalkan n utas selamanya, atau sampai saya menelepon shutdown. Saya menginginkan sesuatu yang persis seperti kumpulan thread yang di-cache (membuat thread berdasarkan permintaan dan kemudian membunuhnya setelah beberapa waktu habis) tetapi dengan batas jumlah thread yang dapat dibuat.
Matt Crinklaw-Vogt
0
  1. Anda dapat menggunakan ThreadPoolExecutorseperti yang disarankan oleh @sjlee

    Anda dapat mengontrol ukuran kolam secara dinamis. Lihat pertanyaan ini untuk lebih jelasnya:

    Pool Thread Dinamis

    ATAU

  2. Anda dapat menggunakan newWorkStealingPool API, yang telah diperkenalkan dengan java 8.

    public static ExecutorService newWorkStealingPool()

    Membuat kumpulan thread yang bekerja-mencuri menggunakan semua prosesor yang tersedia sebagai level paralelisme targetnya.

Secara default, level paralelisme diatur ke jumlah inti CPU di server Anda. Jika Anda memiliki 4 server CPU inti, ukuran kumpulan thread akan menjadi 4. API ini mengembalikan ForkJoinPooltipe ExecutorService dan memungkinkan pencurian pekerjaan dari thread yang menganggur dengan mencuri tugas dari utas yang sibuk di ForkJoinPool.

Ravindra babu
sumber
0

Masalahnya diringkas sebagai berikut:

Saya menginginkan sesuatu yang persis seperti kumpulan thread yang di-cache (membuat thread berdasarkan permintaan dan kemudian membunuhnya setelah beberapa waktu habis) tetapi dengan batas jumlah thread yang dapat dibuat dan kemampuan untuk terus mengantre tugas-tugas tambahan setelah mencapai thread-nya batas utas.

Sebelum menunjuk ke solusi saya akan menjelaskan mengapa solusi berikut tidak bekerja:

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());

Ini tidak akan mengantri tugas apa pun ketika batas 3 tercapai karena SynchronousQueue, menurut definisi, tidak dapat menampung elemen apa pun.

new ThreadPoolExecutor(0, 3, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());

Ini tidak akan membuat lebih dari satu utas karena ThreadPoolExecutor hanya membuat utas yang melebihi corePoolSize jika antrian penuh. Tapi LinkedBlockingQueue tidak pernah penuh.

ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 3, 60, TimeUnit.SECONDS,
    new LinkedBlockingQueue<Runnable>());
executor.allowCoreThreadTimeOut(true);

Ini tidak akan menggunakan kembali utas sampai corePoolSize telah tercapai karena ThreadPoolExecutor menambah jumlah utas sampai corePoolSize tercapai bahkan jika utas yang ada menganggur. Jika Anda dapat hidup dengan kekurangan ini maka ini adalah solusi termudah untuk masalah ini. Ini juga merupakan solusi yang dijelaskan dalam "Java Concurrency in Practice" (catatan kaki pada hal. 175).

Satu-satunya solusi lengkap untuk masalah yang dideskripsikan tampaknya adalah yang melibatkan menimpa metode antrian offerdan menulis RejectedExecutionHandlerseperti yang dijelaskan dalam jawaban atas pertanyaan ini: Bagaimana cara mendapatkan ThreadPoolExecutor untuk meningkatkan utas secara maksimal sebelum mengantri?

Stefan Feuerhahn
sumber
0

Ini berfungsi untuk Java8 + (dan lainnya, untuk saat ini ..)

     Executor executor = new ThreadPoolExecutor(3, 3, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>()){{allowCoreThreadTimeOut(true);}};

di mana 3 adalah batas jumlah utas, dan 5 adalah batas waktu untuk utas menganggur.

Jika Anda ingin memeriksa apakah itu berfungsi sendiri , berikut adalah kode untuk melakukan pekerjaan:

public static void main(String[] args) throws InterruptedException {
    final int DESIRED_NUMBER_OF_THREADS=3; // limit of number of Threads for the task at a time
    final int DESIRED_THREAD_IDLE_DEATH_TIMEOUT=5; //any idle Thread ends if it remains idle for X seconds

    System.out.println( java.lang.Thread.activeCount() + " threads");
    Executor executor = new ThreadPoolExecutor(DESIRED_NUMBER_OF_THREADS, DESIRED_NUMBER_OF_THREADS, DESIRED_THREAD_IDLE_DEATH_TIMEOUT, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>()) {{allowCoreThreadTimeOut(true);}};

    System.out.println(java.lang.Thread.activeCount() + " threads");

    for (int i = 0; i < 5; i++) {
        final int fi = i;
        executor.execute(() -> waitsout("starting hard thread computation " + fi, "hard thread computation done " + fi,2000));
    }
    System.out.println("If this is UP, it works");

    while (true) {
        System.out.println(
                java.lang.Thread.activeCount() + " threads");
        Thread.sleep(700);
    }

}

static void waitsout(String pre, String post, int timeout) {
    try {
        System.out.println(pre);
        Thread.sleep(timeout);
        System.out.println(post);
    } catch (Exception e) {
    }
}

Output dari kode di atas untuk saya adalah

1 threads
1 threads
If this is UP, it works
starting hard thread computation 0
4 threads
starting hard thread computation 2
starting hard thread computation 1
4 threads
4 threads
hard thread computation done 2
hard thread computation done 0
hard thread computation done 1
starting hard thread computation 3
starting hard thread computation 4
4 threads
4 threads
4 threads
hard thread computation done 3
hard thread computation done 4
4 threads
4 threads
4 threads
4 threads
3 threads
3 threads
3 threads
1 threads
1 threads
Ondřej
sumber