from concurrent.futures import Future
# 創(chuàng)建 Future 對(duì)象 future
future = Future()
# 給 future 綁定回調(diào)
def callback(f: Future):
    print("當(dāng)set_result的時(shí)候會(huì)執(zhí)行回調(diào)，result:",
          f.result())
future.add_done_callback(callback)
# 通過(guò) add_done_callback 方法即可給 future 綁定回調(diào)
# 調(diào)用的時(shí)候會(huì)自動(dòng)將 future 作為參數(shù)
# 如果需要多個(gè)參數(shù)，那么就使用偏函數(shù)
# 回調(diào)函數(shù)什么時(shí)候執(zhí)行呢？
# 顯然是當(dāng) future 執(zhí)行 set_result 的時(shí)候
# 如果 future 是向線程池提交函數(shù)時(shí)返回的
# 那么當(dāng)函數(shù)執(zhí)行完畢時(shí)會(huì)自動(dòng)執(zhí)行 future.set_result(xx)
# 并將自身的返回設(shè)置進(jìn)去
# 而這里的 future 是我們手動(dòng)創(chuàng)建的，因此需要手動(dòng)執(zhí)行
future.set_result("嘿嘿")

from concurrent.futures import Future
future = Future()
future.set_result("哼哼")
print(future.result())  # 哼哼
print(future.result())  # 哼哼
print(future.result())  # 哼哼

from concurrent.futures import ThreadPoolExecutor
import time
def task(name, n):
    time.sleep(n)
    return f"{name} 睡了 {n} 秒"
# 創(chuàng)建一個(gè)線程池
# 里面還可以指定 max_workers 參數(shù)，表示最多創(chuàng)建多少個(gè)線程
# Python學(xué)習(xí)交流裙279199867
# 如果不指定，那么每提交一個(gè)函數(shù)，都會(huì)為其創(chuàng)建一個(gè)線程
executor = ThreadPoolExecutor()
# 通過(guò) submit 即可將函數(shù)提交到線程池，一旦提交，就會(huì)立刻運(yùn)行
# 因?yàn)殚_(kāi)啟了一個(gè)新的線程，主線程會(huì)繼續(xù)往下執(zhí)行
# 至于 submit 的參數(shù)，按照函數(shù)名，對(duì)應(yīng)參數(shù)提交即可
# 切記不可寫(xiě)成task("古明地覺(jué)", 3)，這樣就變成調(diào)用了
future = executor.submit(task, "屏幕前的你", 3)
# 由于函數(shù)里面出現(xiàn)了 time.sleep，并且指定的 n 是 3
# 所以函數(shù)內(nèi)部會(huì)休眠 3 秒，顯然此時(shí)處于運(yùn)行狀態(tài)
print(future)
"""
<Future at 0x7fbf701726d0 state=running>
"""
# 我們說(shuō) future 相當(dāng)于一個(gè)容器，包含了內(nèi)部函數(shù)的執(zhí)行狀態(tài)
# 函數(shù)是否正在運(yùn)行中
print(future.running())
"""
True
"""
# 函數(shù)是否執(zhí)行完畢
print(future.done())
"""
False
"""
# 主程序也 sleep 3 秒
time.sleep(3)
# 顯然此時(shí)函數(shù)已經(jīng)執(zhí)行完畢了
# 并且打印結(jié)果還告訴我們返回值類型是 str
print(future)
"""
<Future at 0x7fbf701726d0 state=finished returned str>
"""
print(future.running())
"""
False
"""
print(future.done())
"""
True
"""
# 函數(shù)執(zhí)行完畢時(shí)，會(huì)將返回值設(shè)置在 future 里
# 也就是說(shuō)一旦執(zhí)行了 future.set_result
# 那么就表示函數(shù)執(zhí)行完畢了，然后外界可以調(diào)用 result 拿到返回值
print(future.result())
"""
屏幕前的你 睡了 3 秒
"""

# 提交函數(shù)
future = executor.submit(task, "屏幕前的你", 3)
start = time.perf_counter()
future.result()
end = time.perf_counter()
print(end - start)  # 3.00331525

from concurrent.futures import (
    ThreadPoolExecutor,
    TimeoutError
)
import time
def task(name, n):
    time.sleep(n)
    return f"{name} 睡了 {n} 秒"
executor = ThreadPoolExecutor()
future = executor.submit(task, "屏幕前的你", 3)
try:
    # 1 秒之內(nèi)獲取不到值，拋出 TimeoutError
    res = future.result(1)
except TimeoutError:
    pass
# 再 sleep 2 秒，顯然函數(shù)執(zhí)行完畢了
time.sleep(2)
# 獲取返回值
print(future.result())
"""
屏幕前的你 睡了 3 秒
"""

from concurrent.futures import ThreadPoolExecutor
import time
def task(name, n):
    time.sleep(n)
    return f"{name} 睡了 {n} 秒"
def callback(f):
    print(f.result())
executor = ThreadPoolExecutor()
future = executor.submit(task, "屏幕前的你", 3)
# 綁定回調(diào)，3 秒之后自動(dòng)調(diào)用
future.add_done_callback(callback)
"""
屏幕前的你 睡了 3 秒
"""

from concurrent.futures import ThreadPoolExecutor
import time
def task(name, n):
    time.sleep(n)
    return f"{name} 睡了 {n} 秒"
executor = ThreadPoolExecutor()
futures = [executor.submit(task, "屏幕前的你", 3),
           executor.submit(task, "屏幕前的你", 4),
           executor.submit(task, "屏幕前的你", 1)]
# 此時(shí)都處于running
print(futures)
"""
[<Future at 0x1b5ff622550 state=running>,
 <Future at 0x1b5ff63ca60 state=running>, 
 <Future at 0x1b5ff63cdf0 state=running>]
"""
time.sleep(3)
# 主程序 sleep 3s 后
# futures[0]和futures[2]處于 finished
# futures[1]仍處于 running
print(futures)
"""
[<Future at 0x1b5ff622550 state=running>, 
 <Future at 0x1b5ff63ca60 state=running>, 
 <Future at 0x1b5ff63cdf0 state=finished returned str>]
"""

executor = ThreadPoolExecutor()
futures = [executor.submit(task, "屏幕前的你", 5),
           executor.submit(task, "屏幕前的你", 2),
           executor.submit(task, "屏幕前的你", 4),
           executor.submit(task, "屏幕前的你", 3),
           executor.submit(task, "屏幕前的你", 6)]
for future in futures:
    print(future.result())
"""
屏幕前的你 睡了 5 秒
屏幕前的你 睡了 2 秒
屏幕前的你 睡了 4 秒
屏幕前的你 睡了 3 秒
屏幕前的你 睡了 6 秒
"""

executor = ThreadPoolExecutor()
# map 內(nèi)部也是使用了 submit
results = executor.map(task,
                       ["屏幕前的你"] * 3,
                       [3, 1, 2])
# 并且返回的是迭代器
print(results)
"""
<generator object ... at 0x0000022D78EFA970>
"""
# 此時(shí)遍歷得到的是不再是 future
# 而是 future.result()
for result in results:
    print(result)
"""
屏幕前的你 睡了 3 秒
屏幕前的你 睡了 1 秒
屏幕前的你 睡了 2 秒
"""

executor = ThreadPoolExecutor()
results = executor.map(task,
                       ["屏幕前的你"] * 3,
                       [3, 1, 2])
print(list(results))
"""
['屏幕前的你 睡了 3 秒', 
 '屏幕前的你 睡了 1 秒', 
 '屏幕前的你 睡了 2 秒']
"""

from concurrent.futures import (
    ThreadPoolExecutor,
    as_completed
)
import time
def task(name, n):
    time.sleep(n)
    return f"{name} 睡了 {n} 秒"
executor = ThreadPoolExecutor()
futures = [executor.submit(task, "屏幕前的你", 5),
           executor.submit(task, "屏幕前的你", 2),
           executor.submit(task, "屏幕前的你", 1),
           executor.submit(task, "屏幕前的你", 3),
           executor.submit(task, "屏幕前的你", 4)]
for future in as_completed(futures):
    print(future.result())
"""
屏幕前的你 睡了 1 秒
屏幕前的你 睡了 2 秒
屏幕前的你 睡了 3 秒
屏幕前的你 睡了 4 秒
屏幕前的你 睡了 5 秒
"""

executor = ThreadPoolExecutor()
future1 = executor.submit(task, "屏幕前的你", 1)
future2 = executor.submit(task, "屏幕前的你", 2)
future3 = executor.submit(task, "屏幕前的你", 3)
# 取消函數(shù)的執(zhí)行
# 會(huì)將 future 的 _state 屬性設(shè)置為 CANCELLED
future3.cancel()
# 查看是否被取消
print(future3.cancelled())  # False

executor = ThreadPoolExecutor(max_workers=2)
future1 = executor.submit(task, "屏幕前的你", 1)
future2 = executor.submit(task, "屏幕前的你", 2)
future3 = executor.submit(task, "屏幕前的你", 3)
# 如果池子里可以創(chuàng)建空閑線程
# 那么函數(shù)一旦提交就會(huì)運(yùn)行，狀態(tài)為 RUNNING
print(future1._state)  # RUNNING
print(future2._state)  # RUNNING
# 但 future3 內(nèi)部的函數(shù)還沒(méi)有運(yùn)行
# 因?yàn)槌刈永餆o(wú)法創(chuàng)建新的空閑線程了，所以狀態(tài)為 PENDING
print(future3._state)  # PENDING
# 取消函數(shù)的執(zhí)行，前提是函數(shù)沒(méi)有運(yùn)行
# 會(huì)將 future 的 _state 屬性設(shè)置為 CANCELLED
future3.cancel()
# 查看是否被取消
print(future3.cancelled())  # True
print(future3._state)  # CANCELLED

from concurrent.futures import ThreadPoolExecutor
def task1():
    1 / 0
def task2():
    pass
executor = ThreadPoolExecutor(max_workers=2)
future1 = executor.submit(task1)
future2 = executor.submit(task2)
print(future1)
print(future2)
"""
<Future at 0x7fe3e00f9e50 state=finished raised ZeroDivisionError>
<Future at 0x7fe3e00f9eb0 state=finished returned NoneType>
"""
# 結(jié)果顯示 task1 函數(shù)執(zhí)行出現(xiàn)異常了
# 那么這個(gè)異常要怎么獲取呢？
print(future1.exception())
print(future1.exception().__class__)
"""
division by zero
<class 'ZeroDivisionError'>
"""
# 如果執(zhí)行沒(méi)有出現(xiàn)異常，那么 exception 方法返回 None
print(future2.exception())  # None
# 注意：如果函數(shù)執(zhí)行出現(xiàn)異常了
# 那么調(diào)用 result 方法會(huì)將異常拋出來(lái)
future1.result()
"""
Traceback (most recent call last):
  File "...", line 4, in task1
    1 / 0
ZeroDivisionError: division by zero
"""

from concurrent.futures import ThreadPoolExecutor
import time
def task(n):
    time.sleep(n)
    return f"sleep {n}"
executor = ThreadPoolExecutor()
future1 = executor.submit(task, 5)
future2 = executor.submit(task, 2)
future3 = executor.submit(task, 4)
# 這里是不會(huì)阻塞的
print("start")
# 遍歷所有的 future，并調(diào)用其 result 方法
# 這樣就會(huì)等到所有的函數(shù)都執(zhí)行完畢之后才會(huì)往下走
for future in [future1, future2, future3]:
    print(future.result())
print("end")
"""
start
sleep 5
sleep 2
sleep 4
end
"""

from concurrent.futures import (
    ThreadPoolExecutor,
    wait
)
import time
def task(n):
    time.sleep(n)
    return f"sleep {n}"
executor = ThreadPoolExecutor()
future1 = executor.submit(task, 5)
future2 = executor.submit(task, 2)
future3 = executor.submit(task, 4)
# return_when 有三個(gè)可選參數(shù)
# FIRST_COMPLETED：當(dāng)任意一個(gè)任務(wù)完成或者取消
# FIRST_EXCEPTION：當(dāng)任意一個(gè)任務(wù)出現(xiàn)異常
#                  如果都沒(méi)出現(xiàn)異常等同于ALL_COMPLETED
# ALL_COMPLETED：所有任務(wù)都完成，默認(rèn)是這個(gè)值
fs = wait([future1, future2, future3],
          return_when="ALL_COMPLETED")
# 此時(shí)返回的fs是DoneAndNotDoneFutures類型的namedtuple
# 里面有兩個(gè)值，一個(gè)是done，一個(gè)是not_done
print(fs.done)
"""
{<Future at 0x1df1400 state=finished returned str>, 
 <Future at 0x2f08e48 state=finished returned str>, 
 <Future at 0x9f7bf60 state=finished returned str>}
"""
print(fs.not_done)
"""
set()
"""
for f in fs.done:
    print(f.result())
"""
start
sleep 5
sleep 2
sleep 4
end
"""

# 使用上下文管理
with ThreadPoolExecutor() as executor:
    future1 = executor.submit(task, 5)
    future2 = executor.submit(task, 2)
    future3 = executor.submit(task, 4)
# 所有函數(shù)執(zhí)行完畢（with語(yǔ)句結(jié)束）后才會(huì)往下執(zhí)行

executor = ThreadPoolExecutor()
future1 = executor.submit(task, 5)
future2 = executor.submit(task, 2)
future3 = executor.submit(task, 4)
# 所有函數(shù)執(zhí)行結(jié)束后，才會(huì)往下執(zhí)行
executor.shutdown()

from concurrent.futures import ProcessPoolExecutor
import time
def task(n):
    time.sleep(n)
    return f"sleep {n}"
executor = ProcessPoolExecutor()
# Windows 上需要加上這一行
if __name__ == '__main__':
    future1 = executor.submit(task, 5)
    future2 = executor.submit(task, 2)
    future3 = executor.submit(task, 4)
    executor.shutdown()
    print(future1.result())
    print(future2.result())
    print(future3.result())
"""
sleep 5
sleep 2
sleep 4
"""