+-----------------------------------------------+
|              Python 進(jìn)程 (單個(gè)進(jìn)程)            |
|                                               |
|  +-----------------------------------------+  |
|  |          全局解釋器鎖 (GIL)              |  |
|  |                                         |  |
|  |      ?? 一把鎖，控制 Python 字節(jié)碼執(zhí)行    |  |
|  +-----------------------------------------+  |
|                    ↑                          |
|                    | (獲取/釋放)              |
|                    |                          |
|  +------------+  +------------+  +------------+|
|  |  線程 1    |  |  線程 2    |  |  線程 3    ||
|  |            |  |            |  |            ||
|  | Python代碼 |  | Python代碼 |  | Python代碼 ||
|  |  執(zhí)行中    |  |  等待中    |  |  等待中    ||
|  +------------+  +------------+  +------------+|
|                                               |
+-----------------------------------------------+

import threading
import time

def count_down(n):
    while n > 0:
        n -= 1

# 單線程執(zhí)行
start = time.time()
count_down(100000000)
single_time = time.time() - start

# 多線程執(zhí)行
start = time.time()
t1 = threading.Thread(target=count_down, args=(50000000,))
t2 = threading.Thread(target=count_down, args=(50000000,))
t1.start()
t2.start()
t1.join()
t2.join()
multi_time = time.time() - start

print(f"單線程執(zhí)行時(shí)間: {single_time:.2f}秒")
print(f"雙線程執(zhí)行時(shí)間: {multi_time:.2f}秒")
# 你會(huì)發(fā)現(xiàn)多線程可能比單線程更慢！

import sys

a = []
print(sys.getrefcount(a))  # 查看對(duì)象的引用計(jì)數(shù)

b = a
print(sys.getrefcount(a))  # 引用計(jì)數(shù)增加

# 偽代碼演示競(jìng)爭(zhēng)條件
# 線程1: obj.ref_count += 1
# 線程2: obj.ref_count -= 1
# 如果沒有同步機(jī)制，ref_count 可能出錯(cuò)

import threading
import time

def cpu_intensive_task():
    result = 0
    for i in range(10**7):
        result += i * i
    return result

# 測(cè)試多線程性能
def test_multithreading():
    threads = []
    start_time = time.time()
    
    for _ in range(4):
        t = threading.Thread(target=cpu_intensive_task)
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    print(f"多線程執(zhí)行時(shí)間: {time.time() - start_time:.2f}秒")

# 對(duì)比多進(jìn)程
import multiprocessing

def test_multiprocessing():
    processes = []
    start_time = time.time()
    
    for _ in range(4):
        p = multiprocessing.Process(target=cpu_intensive_task)
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()
    
    print(f"多進(jìn)程執(zhí)行時(shí)間: {time.time() - start_time:.2f}秒")

# 運(yùn)行測(cè)試
if __name__ == "__main__":
    test_multithreading()  # 可能比單線程還慢
    test_multiprocessing()  # 真正的并行，速度更快

import threading
import time
import requests

def download_site(url, session):
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")

def download_all_sites(sites):
    with requests.Session() as session:
        # 單線程
        start_time = time.time()
        for url in sites:
            download_site(url, session)
        print(f"單線程下載時(shí)間: {time.time() - start_time:.2f}秒")
        
        # 多線程
        start_time = time.time()
        threads = []
        for url in sites:
            thread = threading.Thread(target=download_site, args=(url, session))
            thread.start()
            threads.append(thread)
        
        for thread in threads:
            thread.join()
        print(f"多線程下載時(shí)間: {time.time() - start_time:.2f}秒")

# I/O 密集型任務(wù)中，多線程有明顯優(yōu)勢(shì)

from multiprocessing import Pool, cpu_count
import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

def find_primes_parallel(numbers):
    with Pool(processes=cpu_count()) as pool:
        results = pool.map(is_prime, numbers)
    return results

numbers = range(1000000, 1010000)
primes = find_primes_parallel(numbers)

+------------------------+
|   主進(jìn)程 (協(xié)調(diào)者)       |
+------------------------+
          |
    +-----+-----+
    |           |
    v           v
+-------+   +-------+
| 進(jìn)程1 |   | 進(jìn)程2 |
|       |   |       |
|  ??GIL |   |  ??GIL |  ← 每個(gè)進(jìn)程有獨(dú)立的GIL
+-------+   +-------+

// primes.c
#include <Python.h>

static PyObject* find_primes_c(PyObject* self, PyObject* args) {
    Py_BEGIN_ALLOW_THREADS  // 釋放 GIL
    // 執(zhí)行計(jì)算密集型任務(wù)
    Py_END_ALLOW_THREADS    // 重新獲取 GIL
    return Py_BuildValue("i", result);
}

+-----------------------+
|   Python 線程         |
+-----------------------+
          |
          v
+-----------------------+
|   釋放 GIL 的 C 擴(kuò)展   | ← 在C代碼中手動(dòng)釋放GIL
+-----------------------+
          |
          v
+-----------------------+
|   并行計(jì)算 (無(wú)GIL限制)  |
+-----------------------+

+----------------+----------------+----------------+
|   CPython      |   Jython       |   IronPython   |
|   (有GIL)       |   (無(wú)GIL)      |   (無(wú)GIL)      |
+----------------+----------------+----------------+

import asyncio
import aiohttp

async def download_site_async(session, url):
    async with session.get(url) as response:
        content = await response.read()
        print(f"Read {len(content)} from {url}")

async def download_all_sites_async(sites):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in sites:
            task = asyncio.create_task(download_site_async(session, url))
            tasks.append(task)
        await asyncio.gather(*tasks)

# 運(yùn)行異步任務(wù)
asyncio.run(download_all_sites_async(sites))

# 根據(jù)任務(wù)類型選擇方案：

def choose_concurrency_method(task_type, data):
    if task_type == "cpu_intensive":
        # 使用多進(jìn)程
        with multiprocessing.Pool() as pool:
            return pool.map(process_data, data)
    
    elif task_type == "io_intensive":
        # 使用多線程或異步
        with ThreadPoolExecutor() as executor:
            return list(executor.map(process_data, data))
    
    elif task_type == "mixed":
        # 混合方案：進(jìn)程池 + 線程池
        pass