Bagaimana untuk menyelesaikan masalah kolam benang Python's ThreadPoolExecutor

Konsep

Python sudah mempunyai modul threading, mengapa anda memerlukan kumpulan benang Apakah kolam benang?

Ambil perangkak sebagai contoh Anda perlu mengawal bilangan utas yang dirangkak pada masa yang sama Dalam contoh, 20 utas dibuat dan hanya 3 utas dibenarkan berjalan pada masa yang sama , kesemua 20 utas perlu dibuat dan dimusnahkan Penciptaan memerlukan sumber sistem.

Malah, hanya tiga utas diperlukan untuk setiap utas diberikan tugasan, dan tugasan yang selebihnya akan dibariskan untuk menunggu Apabila utas menyelesaikan tugasan, tugasan beratur boleh diatur untuk meneruskan pelaksanaan .

Ini ialah idea kumpulan benang (sudah tentu ia tidak semudah itu), tetapi sukar untuk menulis kumpulan benang dengan sempurna sendiri Anda juga perlu mempertimbangkan penyegerakan benang dalam situasi yang rumit. dan kebuntuan boleh berlaku dengan mudah.

Bermula dari Python3.2, perpustakaan standard menyediakan kami dengan modul concurrent.futures, yang menyediakan dua kelas ThreadPoolExecutor dan ProcessPoolExecutor untuk mencapai abstraksi selanjutnya threading dan multiprocessing (Yang utama fokus di sini ialah pada kumpulan benang), yang bukan sahaja dapat membantu kami menjadualkan urutan secara automatik, tetapi juga melakukan:

• Urutan utama boleh mendapatkan status urutan (atau tugasan) tertentu dan nilai pulangan.

• Apabila thread selesai, thread utama boleh tahu serta-merta.

• Jadikan antara muka pengekodan berbilang benang dan berbilang proses konsisten.

Instance

Penggunaan mudah

from concurrent.futures import ThreadPoolExecutor
import time
 
# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
 
executor = ThreadPoolExecutor(max_workers=2)
# 通过submit函数提交执行的函数到线程池中，submit函数立即返回，不阻塞
task1 = executor.submit(get_html, (3))
task2 = executor.submit(get_html, (2))
# done方法用于判定某个任务是否完成
print(task1.done())
# cancel方法用于取消某个任务,该任务没有放入线程池中才能取消成功
print(task2.cancel())
time.sleep(4)
print(task1.done())
# result方法可以获取task的执行结果
print(task1.result())
 
# 执行结果
# False  # 表明task1未执行完成
# False  # 表明task2取消失败，因为已经放入了线程池中
# get page 2s finished
# get page 3s finished
# True  # 由于在get page 3s finished之后才打印，所以此时task1必然完成了
# 3     # 得到task1的任务返回值

ThreadPoolExecutor Apabila membina instance, masukkan parameter max_workers untuk menetapkan bilangan maksimum thread yang boleh dijalankan serentak dalam kolam benang.

Gunakan fungsi serah untuk menyerahkan tugasan (nama fungsi dan parameter) yang perlu dilakukan oleh utas ke kumpulan utas dan kembalikan pemegang tugasan (serupa dengan fail dan lukisan Perhatikan bahawa serahkan(). ) tidak menyekat, tetapi Kembali dengan segera.

Menggunakan pemegang tugas yang dikembalikan oleh fungsi hantar, anda boleh menggunakan kaedah done() untuk menentukan sama ada tugasan telah tamat. Seperti yang dapat dilihat daripada contoh di atas, memandangkan tugasan mempunyai kelewatan 2s, ia dinilai serta-merta selepas tugas1 diserahkan bahawa tugas1 belum selesai, tetapi selepas kelewatan 4s, ia dinilai tugas1 telah selesai.

Gunakan kaedah cancel() untuk membatalkan tugasan yang diserahkan Jika tugasan sudah dijalankan dalam kumpulan benang, ia tidak boleh dibatalkan. Dalam contoh ini, saiz kumpulan benang ditetapkan kepada 2 dan tugasan sedang dijalankan, jadi pembatalan gagal. Jika anda menukar saiz kumpulan benang kepada 1, maka tugas1 diserahkan dahulu, dan tugas2 masih menunggu dalam baris gilir Pada masa ini, ia boleh berjaya dibatalkan.

Gunakan kaedah result() untuk mendapatkan nilai pulangan tugas. Melihat kod dalaman, kami mendapati bahawa kaedah ini menyekat.

sebagai_selesai

Walaupun perkara di atas menyediakan kaedah untuk menentukan sama ada tugasan selesai, ia tidak boleh selalu ditentukan dalam urutan utama.

Kadang-kadang apabila kita tahu bahawa tugasan tertentu telah selesai, kita mendapat hasilnya daripada terus-menerus menilai sama ada setiap tugasan telah selesai.

Ini adalah hasil daripada menggunakan kaedah as_completed untuk mendapatkan semula semua tugasan sekaligus. Kaedah

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
 
# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
 
executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
all_task = [executor.submit(get_html, (url)) for url in urls]
 
for future in as_completed(all_task):
    data = future.result()
    print("in main: get page {}s success".format(data))
 
# 执行结果
# get page 2s finished
# in main: get page 2s success
# get page 3s finished
# in main: get page 3s success
# get page 4s finished
# in main: get page 4s success

as_completed() ialah penjana Apabila tiada tugasan selesai, ia akan menyekat Apabila tugasan tertentu selesai, ia akan yield tugasan ini dan pernyataan di bawah gelung for boleh dilaksanakan. . , dan kemudian teruskan sekat dan gelung sehingga semua tugasan selesai.

Juga dapat dilihat daripada keputusan bahawa tugasan yang selesai dahulu akan memberitahu utas utama terlebih dahulu.

peta

Selain kaedah as_completed di atas, anda juga boleh menggunakan kaedah executor.map, tetapi terdapat sedikit perbezaan.

from concurrent.futures import ThreadPoolExecutor
import time
 
# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
 
executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
 
for data in executor.map(get_html, urls):
    print("in main: get page {}s success".format(data))
# 执行结果
# get page 2s finished
# get page 3s finished
# in main: get page 3s success
# in main: get page 2s success
# get page 4s finished
# in main: get page 4s success

menggunakan kaedah map tanpa menggunakan kaedah submit terlebih dahulu Kaedah map mempunyai makna yang sama dengan python dalam pustaka standard <🎜, yang kedua-duanya melaksanakan setiap satu. elemen dalam urutan yang sama. map

Kod di atas adalah untuk melaksanakan fungsi

bagi setiap elemen urls dan memperuntukkan setiap kumpulan benang. Dapat dilihat bahawa hasil pelaksanaan adalah berbeza daripada hasil kaedah get_html di atas Urutan output adalah sama dengan susunan senarai as_completed Genap jika tugasan 2s dilaksanakan terlebih dahulu, ia akan dilaksanakan terlebih dahulu Tugas yang mencetak selama 3 saat selesai dahulu, dan kemudian tugasan yang mencetak selama 2 saat selesai. Kaedah urlstunggu

membolehkan utas utama disekat sehingga keperluan yang ditetapkan dipenuhi. Kaedah

from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED
import time
 
# 参数times用来模拟网络请求的时间
def get_html(times):
    time.sleep(times)
    print("get page {}s finished".format(times))
    return times
 
executor = ThreadPoolExecutor(max_workers=2)
urls = [3, 2, 4] # 并不是真的url
all_task = [executor.submit(get_html, (url)) for url in urls]
wait(all_task, return_when=ALL_COMPLETED)
print("main")
# 执行结果 
# get page 2s finished
# get page 3s finished
# get page 4s finished
# main

wait menerima 3 parameter, urutan tugas menunggu, masa tamat masa dan keadaan menunggu.

Syarat menunggu wait lalai kepada

, menunjukkan bahawa anda mahu menunggu sehingga semua tugasan tamat.

return_whenAnda boleh melihat dalam keputusan yang sedang dijalankan bahawa semua tugasan memang telah selesai, dan urutan utama mencetak ALL_COMPLETED.

Syarat menunggu juga boleh ditetapkan kepada main, yang bermaksud bahawa penantian akan berhenti apabila tugasan pertama selesai.

Analisis kod sumberFIRST_COMPLETED

dalam modul bermaksud objek masa hadapan, yang boleh difahami sebagai

operasi yang diselesaikan pada masa hadapancocurrent.future, yang tidak segerak Asas pengaturcaraan. future

Selepas kumpulan benang submit(), objek future ini dikembalikan. Tugasan tidak selesai apabila dikembalikan, tetapi akan diselesaikan pada masa hadapan.

juga boleh dipanggil bekas pemulangan tugas, yang akan menyimpan hasil dan status tugas.

Maka bagaimanakah ThreadPoolExecutor mengendalikan objek ini secara dalaman?

Berikut ialah pengenalan ringkas kepada sebahagian daripada kod ThreadPoolExecutor:

kaedah 1.init Perkara penting utama dalam kaedah

init ialah baris gilir tugas dan pengumpulan benang Dalam kaedah lain Perlu digunakan dalam.

2 Terdapat dua objek penting dalam kaedah penyerahan

submit, objek _base.Future() dan _WorkItem() dan objek bertanggungjawab Jalankan tugas dan tetapkan objek _WorkItem(), dan akhirnya objek future akan dikembalikan. Anda dapat melihat bahawa keseluruhan proses kembali serta-merta tanpa menyekat. future

3. kaedah adjust_thread_count

Maksud kaedah ini mudah difahami, terutamanya untuk mencipta bilangan utas yang ditentukan. Walau bagaimanapun, pelaksanaannya agak sukar untuk difahami Sebagai contoh, weakref.ref dalam fungsi pelaksanaan thread melibatkan konsep seperti rujukan yang lemah, yang akan difahami kemudian.

4._WorkItem object

Tanggungjawab objek adalah untuk melaksanakan tugas dan menetapkan keputusan. Kerumitan utama di sini ialah _WorkItemself.future.set_result(result).

5. Fungsi pelaksanaan benang--_worker

Ini ialah entri fungsi yang dinyatakan apabila kumpulan benang mencipta benang, terutamanya daripada baris gilir dalam urutan Keluarkan

tugas dan laksanakannya, tetapi parameter pertama fungsi masih belum begitu jelas. Biarkan itu untuk kemudian.

Atas ialah kandungan terperinci Bagaimana untuk menyelesaikan masalah kolam benang Python's ThreadPoolExecutor. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!