python concurrent.futures

masa depan

Masa hadapan ialah bekas yang boleh menyimpan sama ada hasil pengiraan atau ralat yang berlaku semasa pengiraan itu. Apabila masa depan dicipta, ia bermula dalam keadaan TERTUTUP. Pustaka tidak berhasrat untuk objek ini dibuat secara manual, kecuali mungkin untuk tujuan ujian.

import concurrent.futures as futures

f = futures.Future()
assert(f._result is None)
assert(f._exception is None)
assert(f._state == 'PENDING')

Status PENDING menunjukkan bahawa pengiraan yang diminta oleh pengguna telah didaftarkan dalam kumpulan benang dan diletakkan dalam baris gilir, tetapi masih belum diambil oleh mana-mana urutan untuk dilaksanakan. Sebaik sahaja urutan percuma mengambil tugas (panggilan balik) daripada baris gilir, peralihan masa hadapan kepada keadaan RUNNING. Masa depan hanya boleh dibatalkan semasa berada dalam keadaan TERTUTUP. Oleh itu, terdapat tetingkap masa antara keadaan PENDING dan RUNNING semasa pengiraan yang diminta boleh dibatalkan.

import concurrent.futures as futures

def should_cancel_pending_future():
    f = futures.Future()
    assert(f._state == 'PENDING')
    assert(f.cancel())
    assert(f._state == 'CANCELLED')

def should_not_cancel_running_future():
    f = futures.Future()
    f.set_running_or_notify_cancel()
    assert(f._state == 'RUNNING')
    assert(not f.cancel())

def cancel_is_idempotent():
    f = futures.Future()
    assert(f.cancel())
    assert(f.cancel())


should_cancel_pending_future()
should_not_cancel_running_future()
cancel_is_idempotent()

Operasi yang diminta dalam kumpulan benang boleh sama ada lengkap dengan nilai yang dikira atau mengakibatkan ralat. Tanpa mengira keputusan, peralihan masa depan kepada keadaan SELESAI. Hasil atau ralat kemudiannya disimpan dalam medan yang sepadan.

import concurrent.futures as futures

def future_completed_with_result():
    f = futures.Future()
    f.set_result('foo')
    assert(f._state == 'FINISHED')
    assert(f._result == 'foo')
    assert(f._exception is None)

def future_completed_with_exception():
    f = futures.Future()
    f.set_exception(NameError())
    assert(f._state == 'FINISHED')
    assert(f._result is None)
    assert(isinstance(f._exception, NameError))

future_completed_with_result()
future_completed_with_exception()

Untuk mendapatkan semula hasil pengiraan, kaedah keputusan digunakan. Jika pengiraan masih belum lengkap, kaedah ini akan menyekat urutan semasa (dari mana keputusan dipanggil) sehingga pengiraan selesai atau masa menunggu tamat.

Jika pengiraan berjaya diselesaikan tanpa ralat, kaedah keputusan mengembalikan nilai yang dikira.

import concurrent.futures as futures
import time
import threading

f = futures.Future()
def target():
    time.sleep(1)
    f.set_result('foo')
threading.Thread(target=target).start()
assert(f.result() == 'foo')

Jika pengecualian berlaku semasa pengiraan, keputusan akan menimbulkan pengecualian itu.

import concurrent.futures as futures
import time
import threading

f = futures.Future()
def target():
    time.sleep(1)
    f.set_exception(NameError)
threading.Thread(target=target).start()
try:
    f.result()
    raise Exception()
except NameError:
    assert(True)

Jika kaedah tamat masa sementara menunggu, TimeoutError dinaikkan.

import concurrent.futures as futures

f = futures.Future()
try:
    f.result(1)
    raise Exception()
except TimeoutError:
    assert(f._result is None)
    assert(f._exception is None)

Percubaan untuk mendapatkan hasil pengiraan yang dibatalkan akan menimbulkan Ralat Dibatalkan.

import concurrent.futures as futures

f = futures.Future()
assert(f.cancel())
try:
    f.result()
    raise Exception()
except futures.CancelledError:
    assert(True)

Strategi menunggu

Dalam proses pembangunan, adalah perkara biasa untuk perlu menjalankan pengiraan N pada kumpulan benang dan menunggu selesainya. Untuk mencapai matlamat ini, perpustakaan menyediakan fungsi tunggu. Terdapat beberapa strategi menunggu: FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED.

Lazim kepada semua strategi menunggu ialah jika niaga hadapan yang diluluskan kepada kaedah menunggu sudah selesai, kutipan niaga hadapan yang diluluskan dikembalikan tanpa mengira strategi yang dipilih. Tidak kira bagaimana ia telah diselesaikan sama ada dengan ralat, keputusan atau jika ia telah dibatalkan.

import concurrent.futures as futures

def test(return_when):
    f1, f2, f3 = futures.Future(), futures.Future(), futures.Future()
    f1.cancel()
    f1.set_running_or_notify_cancel() # required
    f2.set_result('foo')
    f3.set_exception(NameError)

    r = futures.wait([f1, f2, f3], return_when=return_when)
    assert(len(r.done) == 3)
    assert(len(r.not_done) == 0)

for return_when in [futures.ALL_COMPLETED, futures.FIRST_EXCEPTION, futures.FIRST_COMPLETED]:
    test(return_when)

ALL_COMPLETED strategi

Strategi ALL_COMPLETED menjamin menunggu penyiapan semua niaga hadapan yang diluluskan, atau keluar selepas tamat masa dengan koleksi niaga hadapan selesai sehingga saat itu, yang mungkin tidak lengkap.

import concurrent.futures as futures

f = futures.Future()
assert(f._result is None)
assert(f._exception is None)
assert(f._state == 'PENDING')

FIRST_COMPLETED

Strategi FIRST_COMPLETED menjamin pemulangan koleksi dengan sekurang-kurangnya satu masa depan yang lengkap atau koleksi kosong sekiranya tamat masa. Strategi ini TIDAK membayangkan bahawa koleksi yang dikembalikan tidak boleh mengandungi berbilang elemen.

import concurrent.futures as futures

def should_cancel_pending_future():
    f = futures.Future()
    assert(f._state == 'PENDING')
    assert(f.cancel())
    assert(f._state == 'CANCELLED')

def should_not_cancel_running_future():
    f = futures.Future()
    f.set_running_or_notify_cancel()
    assert(f._state == 'RUNNING')
    assert(not f.cancel())

def cancel_is_idempotent():
    f = futures.Future()
    assert(f.cancel())
    assert(f.cancel())


should_cancel_pending_future()
should_not_cancel_running_future()
cancel_is_idempotent()

FIRST_EXCEPTION

Strategi FIRST_EXCEPTION mengganggu penantian jika salah satu pengiraan selesai dengan ralat. Jika tiada pengecualian berlaku, tingkah laku adalah sama dengan ALL_COMPLETED masa hadapan.

import concurrent.futures as futures

def future_completed_with_result():
    f = futures.Future()
    f.set_result('foo')
    assert(f._state == 'FINISHED')
    assert(f._result == 'foo')
    assert(f._exception is None)

def future_completed_with_exception():
    f = futures.Future()
    f.set_exception(NameError())
    assert(f._state == 'FINISHED')
    assert(f._result is None)
    assert(isinstance(f._exception, NameError))

future_completed_with_result()
future_completed_with_exception()

ThreadPoolExecutor

Objek bertanggungjawab untuk mencipta kumpulan benang. Kaedah utama untuk berinteraksi dengan objek ini ialah kaedah Hantar. Ia membolehkan untuk mendaftarkan pengiraan dalam kumpulan benang. Sebagai tindak balas, objek Masa Depan dikembalikan, yang digunakan untuk memantau status pengiraan dan mendapatkan hasil akhir.

Hartanah

• Urutan baharu dibuat HANYA mengikut keperluan:
  • Jika terdapat sekurang-kurangnya satu urutan percuma apabila pengiraan diminta, tiada urutan baharu dibuat
  • Jika tiada urutan percuma apabila pengiraan diminta, urutan baharu akan dibuat, dengan syarat had maxWorkers belum dicapai.
  • Jika tiada utas percuma dan had maxWorkers telah dicapai, pengiraan diletakkan dalam baris gilir dan akan diambil oleh urutan seterusnya yang tersedia
• Bilangan maksimum utas yang diperuntukkan untuk keperluan pengiraan secara lalai bersamaan dengan bilangan teras pemproses logik
• Setelah dibuat, benang tidak dimusnahkan, walaupun dalam kes beban rendah

Atas ialah kandungan terperinci python concurrent.futures. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!