并行：多线程同时操作多个资源类（两个或多个事件在同一时刻发生）

并发：多线程交替操作同一资源类（宏观上的并行，微观上的串行）

concurrent.futures

是Python 3.2引入的新模块，它是在threading和multiprocessing之上的一个通用抽象层，提供了ThreadPoolExecutor和ProcessPoolExecutor两个类，以便使用线程池/进程池并发地执行任务。

map与submit方法

map方法很简单，map的结果和入参顺序是对应的。

future模式更强大，用as_completed顺序是不定的。

import httpx from parsel import Selector from concurrent.futures import ThreadPoolExecutor, as_completed urls = [f"https://news.cnblogs.com/n/page/{page}" for page in range(1, 10 + 1)] def craw(url): r = httpx.get(url) return r.text def parse(html): selector = Selector(html) title = selector.css(".content h2.news_entry a::text")[0].get() return title with ThreadPoolExecutor() as pool: htmls = pool.map(craw, urls) htmls = list(zip(urls, htmls)) for url, html in htmls: print(url, len(html)) print("craw over") with ThreadPoolExecutor() as pool: futures = {} for url, html in htmls: future = pool.submit(parse, html) futures[future] = url for future in as_completed(futures): url = futures[future] print(url, future.result()) print("parse over")

Web服务多线程加速

import flask import json import time from concurrent.futures import ThreadPoolExecutor

app = flask.Flask(name) pool = ThreadPoolExecutor()

def read_file(): time.sleep(0.1) return "file rusult"

def read_db(): time.sleep(0.2) return "db rusult"

def read_api(): time.sleep(0.1) return "api rusult"

@app.route("/") def index(): result_file = pool.submit(read_file) result_db = pool.submit(read_db) result_api = pool.submit(read_api) return json.dumps( { "result_file": result_file.result(), "result_db": result_db.result(), "result_api": result_api.result(), } )

if name == "main": app.run()

Web服务多进程加速

from flask import Flask import json import math from concurrent.futures import ProcessPoolExecutor

app = Flask(name)

def is_prime(n): if n < 2: return False if n == 2: return True if n % 2 == 0: return False

sqrt_n = int(math.floor(math.sqrt(n)))
for i in range(3, sqrt_n, 2):
    if n % i == 0:
        return False

return True

@app.route("/is-prime/") def api_is_prime(numbers: str): number_list = [int(x) for x in numbers.split(",")] results = process_pool.map(is_prime, number_list) return json.dumps(dict(zip(number_list, results)))

if name == "main": process_pool = ProcessPoolExecutor() app.run()