Python 并行编程（多线程、多进程）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import threading
import time

def crawl(link, delay=3):
    print(f"crawl started for {link}")
    time.sleep(delay)  # Blocking I/O (simulating a network request)
    print(f"crawl ended for {link}")

links = [
    "https://python.org",
    "https://docs.python.org",
    "https://peps.python.org",
]

# Start threads for each link
threads = []
for link in links:
    # Using `args` to pass positional arguments and `kwargs` for keyword arguments
    t = threading.Thread(target=crawl, args=(link,), kwargs={"delay": 2})
    threads.append(t)

# Start each thread
for t in threads:
    t.start()

# Wait for all threads to finish
for t in threads:
    t.join()

1

class threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# threading.py
from itertools import count as _count

# Helper to generate new thread names
_counter = _count(1).__next__

def _newname(name_template):
    return name_template % _counter()

class Thread:
    _initialized = False

    def __init__(
        self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None
    ):
        assert group is None, "group argument must be None for now"
        if kwargs is None:
            kwargs = {}
        if name:
            name = str(name)
        else:
            name = _newname("Thread-%d")
            if target is not None:
                try:
                    target_name = target.__name__
                    name += f" ({target_name})"
                except AttributeError:
                    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

import threading
import time

def worker(num):
    """线程执行的任务函数"""
    print(f"Worker {num} started")
    time.sleep(2)  # 模拟耗时操作
    print(f"Worker {num} finished")

# 创建线程
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()  # 启动线程

# 等待所有线程完成
for t in threads:
    t.join()

print("All threads completed")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import threading
import time

class MyThread(threading.Thread):
    def __init__(self, num):
        threading.Thread.__init__(self)
        self.num = num
    
    def run(self):
        """重写run方法"""
        print(f"Custom thread {self.num} started")
        time.sleep(1)  # 模拟耗时操作
        print(f"Custom thread {self.num} finished")

# 创建并启动线程
threads = []
for i in range(3):
    t = MyThread(i)
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print("All custom threads completed")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from concurrent.futures import ThreadPoolExecutor
import time

def task(name):
    print(f"Task {name} started")
    time.sleep(2)
    print(f"Task {name} finished")
    return f"Result of {name}"

# 使用线程池
with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交任务到线程池
    futures = [executor.submit(task, i) for i in range(5)]
    
    # 获取结果
    for future in futures:
        print(future.result())

print("All tasks completed")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import threading

lock = threading.Lock()
counter = 0

def increment():
    global counter
    with lock:  # 自动获取和释放锁
        counter += 1

threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)  # 输出 10

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import threading


rlock = threading.RLock()

def recursive_func(n):
    if n > 0:
        with rlock:  # 可重入
            recursive_func(n - 1)

recursive_func(5)  # 不会死锁

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import time
import threading

semaphore = threading.Semaphore(3)  # 最多3个线程同时运行

def task():
    with semaphore:
        print(threading.current_thread().name, "working")
        time.sleep(1)

for i in range(10):
    threading.Thread(target=task).start()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import threading


condition = threading.Condition()
items = []

def consumer():
    with condition:
        while not items:
            condition.wait()  # 等待生产者通知
        print("Consumed:", items.pop())

def producer():
    with condition:
        items.append(1)
        condition.notify()  # 唤醒一个消费者

threading.Thread(target=consumer).start()
threading.Thread(target=producer).start()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

import time
import threading


event = threading.Event()

def waiter1():
    print("waiter1 Waiting for event...")
    event.wait()
    print("waiter1: Event triggered!")

def waiter2():
    print("waiter2 Waiting for event...")
    event.wait()
    print("waiter2: Event triggered!")

threading.Thread(target=waiter1).start()
threading.Thread(target=waiter2).start()

time.sleep(2)
event.set()  # 唤醒所有等待线程

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import time
import threading


barrier = threading.Barrier(3)  # 需要3个线程到达

def worker():
    print("Reached barrier")
    barrier.wait()
    print("Passed barrier")

for _ in range(3):
    threading.Thread(target=worker).start()
    time.sleep(1)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import time
import threading


from queue import Queue

q = Queue()

def producer():
    q.put("data")

def consumer():
    print("Got:", q.get())

# 先启动两个消费线程
threading.Thread(target=consumer).start()
threading.Thread(target=consumer).start()
# 3秒后启动一个生产线程
time.sleep(3)
threading.Thread(target=producer).start()  # 数据生产后，其中一个消费线程会拿到数据并退出，另一个线程继续等待
# 2秒后再启动一个生产线程
time.sleep(2)
threading.Thread(target=producer).start()  # 数据再生产后，另一个消费线程拿到数据退出，主线程结束

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import multiprocessing

def worker():
    print("Process working")

processes = []
for _ in range(5):
    p = multiprocessing.Process(target=worker)
    processes.append(p)
    p.start()

for p in processes:
    p.join()

1
2
3
4
5
6
7
8

import asyncio

async def main():
    print('Hello ...')
    await asyncio.sleep(1)
    print('... World!')

asyncio.run(main())

1

asyncio.run(coro, *, debug=None, loop_factory=None)

1
2
3
4
5
6

async def main():
    await asyncio.sleep(1)
    print('hello')

with asyncio.Runner() as runner:
    runner.run(main())

1

class asyncio.Runner(*, debug=None, loop_factory=None)