Python threading全景指南分享

import threading, time

# 方式一：把函数塞给 Thread
def worker(n):
    print(f'Worker {n} start')
    time.sleep(1)
    print(f'Worker {n} done')

for i in range(3):
    t = threading.Thread(target=worker, args=(i,))
    t.start()

# 方式二：继承 Thread 并重写 run
class MyThread(threading.Thread):
    def __init__(self, n):
        super().__init__()
        self.n = n
    def run(self):
        print(f'MyThread {self.n} start')
        time.sleep(1)
        print(f'MyThread {self.n} done')

MyThread(10).start()

threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
[t.start() for t in threads]
[t.join() for t in threads]  # 等全部结束
print('all done')

counter = 0
lock = threading.Lock()

def add():
    global counter
    for _ in range(100000):
        with lock:             # 等价于 lock.acquire(); try: ... finally: lock.release()
            counter += 1

threads = [threading.Thread(target=add) for _ in range(2)]
[t.start() for t in threads]
[t.join() for t in threads]
print(counter)   # 200000

rlock = threading.RLock()
def foo():
    with rlock:
        bar()

def bar():
    with rlock:   # 同一线程，再次获取成功
        pass

import random, time
q, MAX = [], 5
cond = threading.Condition()

def producer():
    while True:
        with cond:
            while len(q) == MAX:
                cond.wait()          # 等待队列有空位
            item = random.randint(1, 100)
            q.append(item)
            print('+', item, q)
            cond.notify()            # 通知消费者
        time.sleep(0.5)

def consumer():
    while True:
        with cond:
            while not q:
                cond.wait()
            item = q.pop(0)
            print('-', item, q)
            cond.notify()
        time.sleep(0.6)

threading.Thread(target=producer, daemon=True).start()
threading.Thread(target=consumer, daemon=True).start()
time.sleep(5)

sem = threading.Semaphore(3)
def downloader(url):
    with sem:
        print('downloading', url)
        time.sleep(2)

event = threading.Event()

def waiter():
    print('wait...')
    event.wait()          # 阻塞
    print('go!')

threading.Thread(target=waiter).start()
time.sleep(3)
event.set()               # 发令

barrier = threading.Barrier(3)

def phase(name):
    print(name, 'ready')
    barrier.wait()
    print(name, 'go')

for i in range(3):
    threading.Thread(target=phase, args=(i,)).start()

local = threading.local()

def show():
    print(f'{threading.current_thread().name} -> {local.x}')

def task(n):
    local.x = n
    show()

for i in range(3):
    threading.Thread(target=task, args=(i,)).start()

def hello():
    print('hello, timer')
threading.Timer(3.0, hello).start()

from concurrent.futures import ThreadPoolExecutor
import requests, time

URLS = ['https://baidu.com'] * 20

def fetch(url):
    return requests.get(url).status_code

with ThreadPoolExecutor(max_workers=10) as pool:
    for code in pool.map(fetch, URLS):
        print(code)

logging.basicConfig(
    format='%(asctime)s [%(threadName)s] %(message)s',
    level=logging.INFO)