import requests
import threading
import time
from typing import List

API_URL = "https://jsonplaceholder.typicode.com/posts/{}"
THREAD_NUM = 10
TOTAL_REQUESTS = 20
TIMEOUT = 5

# 1. 初始化：固定長度的結(jié)果列表（與請(qǐng)求順序?qū)?yīng)）+ 互斥鎖（保護(hù)列表寫入）
results: List[tuple] = [None] * TOTAL_REQUESTS  # 初始值為 None，完成后寫入結(jié)果
lock = threading.Lock()  # 線程安全：避免多線程同時(shí)修改同一列表位置

def request_api(index: int):
    """線程執(zhí)行函數(shù)：按索引寫入結(jié)果到固定位置"""
    url = API_URL.format(index % 10 + 1)
    try:
        response = requests.get(url, timeout=TIMEOUT)
        response.raise_for_status()
        result = (index, True, f"響應(yīng)：{response.json()['title'][:20]}...")
    except Exception as e:
        result = (index, False, f"失敗：{str(e)[:30]}")

    # 2. 加鎖寫入結(jié)果（僅鎖定寫入操作，不影響請(qǐng)求并發(fā)）
    with lock:
        results[index] = result  # 關(guān)鍵：按請(qǐng)求索引寫入對(duì)應(yīng)位置

if __name__ == "__main__":
    start_time = time.time()
    threads = []

    # 3. 創(chuàng)建并啟動(dòng)線程（控制線程池大小）
    for i in range(TOTAL_REQUESTS):
        # 限制同時(shí)運(yùn)行的線程數(shù)，避免創(chuàng)建過多線程
        if len(threads) >= THREAD_NUM:
            # 等待任意線程完成后再創(chuàng)建新線程
            threading.Thread.join(threading.Thread.wait(threads))
            threads = [t for t in threads if t.is_alive()]
        
        t = threading.Thread(target=request_api, args=(i,), name=f"Thread-{i}")
        t.start()
        threads.append(t)

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

    # 5. 直接按列表順序輸出（已與提交順序一致）
    print("固定位置存儲(chǔ) - 按請(qǐng)求順序輸出：")
    for idx, is_success, msg in results:
        print(f"任務(wù)[{idx}]：{'?' if is_success else '?'} {msg}")

    print(f"\n總耗時(shí)：{round(time.time() - start_time, 3)}s")

import requests
import threading
from queue import Queue
import time

API_URL = "https://jsonplaceholder.typicode.com/posts/{}"
THREAD_NUM = 10
TOTAL_REQUESTS = 20
TIMEOUT = 5

def worker(task_queue: Queue, result_queue: Queue):
    """工作線程：從任務(wù)隊(duì)列取任務(wù)，執(zhí)行后存入結(jié)果隊(duì)列"""
    while not task_queue.empty():
        try:
            index = task_queue.get(timeout=1)  # 非阻塞取任務(wù)（超時(shí)1秒退出）
            url = API_URL.format(index % 10 + 1)
            try:
                response = requests.get(url, timeout=TIMEOUT)
                response.raise_for_status()
                result = (index, True, f"響應(yīng)：{response.json()['title'][:20]}...")
            except Exception as e:
                result = (index, False, f"失?。簕str(e)[:30]}")
            result_queue.put(result)  # 存入結(jié)果隊(duì)列（無序）
        except Exception:
            break

if __name__ == "__main__":
    start_time = time.time()

    # 1. 初始化隊(duì)列
    task_queue = Queue()  # 按順序存入請(qǐng)求索引（0~19）
    result_queue = Queue()  # 存儲(chǔ)無序的結(jié)果

    # 2. 提交任務(wù)（按順序入隊(duì)）
    for i in range(TOTAL_REQUESTS):
        task_queue.put(i)

    # 3. 啟動(dòng)工作線程
    threads = [threading.Thread(target=worker, args=(task_queue, result_queue)) for _ in range(THREAD_NUM)]
    for t in threads:
        t.start()

    # 4. 主線程按順序提取結(jié)果（流式輸出）
    print("隊(duì)列流式處理 - 按請(qǐng)求順序?qū)崟r(shí)輸出：")
    expected_index = 0  # 期望的下一個(gè)任務(wù)索引（從0開始）
    completed = 0  # 已完成的任務(wù)數(shù)

    while completed < TOTAL_REQUESTS:
        if not result_queue.empty():
            index, is_success, msg = result_queue.get()
            # 匹配期望索引則輸出，否則放回隊(duì)列
            if index == expected_index:
                print(f"任務(wù)[{index}]：{'?' if is_success else '?'} {msg}")
                expected_index += 1
                completed += 1
            else:
                result_queue.put((index, is_success, msg))  # 未匹配則放回
        else:
            time.sleep(0.01)  # 避免空循環(huán)占用CPU

    # 5. 等待所有線程結(jié)束
    for t in threads:
        t.join()

    print(f"\n總耗時(shí)：{round(time.time() - start_time, 3)}s")

import requests
from concurrent.futures import ThreadPoolExecutor
import time

API_URL = "https://jsonplaceholder.typicode.com/posts/{}"
THREAD_NUM = 10
TOTAL_REQUESTS = 20
TIMEOUT = 5

def request_api(index: int) -> tuple:
    """單個(gè)請(qǐng)求函數(shù)：返回（索引，是否成功，結(jié)果信息）"""
    url = API_URL.format(index % 10 + 1)
    try:
        response = requests.get(url, timeout=TIMEOUT)
        response.raise_for_status()
        return (index, True, f"響應(yīng)：{response.json()['title'][:20]}...")
    except Exception as e:
        return (index, False, f"失?。簕str(e)[:30]}")

if __name__ == "__main__":
    start_time = time.time()

    # 1. 提交任務(wù)并獲取 Future 列表（順序與提交一致）
    with ThreadPoolExecutor(max_workers=THREAD_NUM) as executor:
        future_list = [executor.submit(request_api, i) for i in range(TOTAL_REQUESTS)]

        # 2. 按 Future 列表順序獲取結(jié)果（阻塞等待，順序與提交一致）
        print("ThreadPoolExecutor 有序收集 - 按請(qǐng)求順序輸出：")
        for future in future_list:
            index, is_success, msg = future.result()  # 按提交順序阻塞等待
            print(f"任務(wù)[{index}]：{'?' if is_success else '?'} {msg}")

    total_cost = round(time.time() - start_time, 3)
    print(f"\n總耗時(shí)：{total_cost}s")

import aiohttp
import asyncio
import time

API_URL = "https://jsonplaceholder.typicode.com/posts/{}"
TOTAL_REQUESTS = 20
TIMEOUT = 5

async def request_api(session: aiohttp.ClientSession, index: int) -> tuple:
    """異步請(qǐng)求函數(shù)"""
    url = API_URL.format(index % 10 + 1)
    try:
        async with session.get(url, timeout=TIMEOUT) as response:
            response.raise_for_status()
            data = await response.json()
            return (index, True, f"響應(yīng)：{data['title'][:20]}...")
    except Exception as e:
        return (index, False, f"失敗：{str(e)[:30]}")

async def main():
    start_time = time.time()

    # 1. 創(chuàng)建異步會(huì)話（復(fù)用連接，提升效率）
    async with aiohttp.ClientSession() as session:
        # 2. 創(chuàng)建所有異步任務(wù)（順序與提交一致）
        tasks = [request_api(session, i) for i in range(TOTAL_REQUESTS)]
        # 3. 并發(fā)執(zhí)行任務(wù)，按提交順序獲取結(jié)果
        results = await asyncio.gather(*tasks)  # gather 保證結(jié)果順序與任務(wù)順序一致

    # 4. 輸出結(jié)果（已有序）
    print("aiohttp 異步并發(fā) - 按請(qǐng)求順序輸出：")
    for idx, is_success, msg in results:
        print(f"任務(wù)[{idx}]：{'?' if is_success else '?'} {msg}")

    total_cost = round(time.time() - start_time, 3)
    print(f"\n總耗時(shí)：{total_cost}s")

if __name__ == "__main__":
    # 兼容 Windows 系統(tǒng)
    if __name__ == "__main__":
        if sys.platform == "win32":
            asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
        asyncio.run(main())

pip install aiohttp