如何实现Node.js网站的并发压力测试？替代串行BeautifulSoup测试方案

阿华AIGC实验室

2026-5-26

嘿，我来帮你优化这个压力测试脚本！串行跑1000次请求确实效率太低了，而且也没法真实模拟用户同时访问的场景。下面给你几个Python里常用的并发优化方案，都是针对这类IO密集型（网络请求）任务的绝佳选择：

方案1：使用多线程（ThreadPoolExecutor + requests）

多线程非常适合处理网络请求这种IO密集型任务——大部分时间都在等服务器响应，线程可以在等待的时候切换去处理其他请求。Python的concurrent.futures.ThreadPoolExecutor能帮你轻松实现并发：

import requests
from bs4 import BeautifulSoup
from random import randint
from concurrent.futures import ThreadPoolExecutor

def send_request():
    l = randint(2008, 2018)
    first = "year=%d" % l
    k = randint(1600, 2400)
    second = "cc=%d" % k
    url = f"http://xx.xxx.xxx.xxx:xxxx/outputData?{first}&{second}&submit=Submit"
    res = requests.get(url)
    soup = BeautifulSoup(res.text, 'lxml')
    print(soup)

if __name__ == "__main__":
    # 控制并发数，建议根据服务器承受能力调整（比如10-50）
    with ThreadPoolExecutor(max_workers=20) as executor:
        # 提交1000个请求任务
        executor.map(send_request, range(1000))

注意：max_workers别设得太大，不然可能一下子把服务器打垮，或者触发对方的限流机制，先从小数值测试调整。

方案2：使用异步IO（aiohttp + asyncio）

异步IO是处理大量网络请求更高效的方式——它不需要创建多个线程，而是通过单线程内的事件循环调度任务，资源占用更低。需要用aiohttp代替requests（因为requests不支持异步）：

import asyncio
import aiohttp
from bs4 import BeautifulSoup
from random import randint

async def send_request(session):
    l = randint(2008, 2018)
    first = f"year={l}"
    k = randint(1600, 2400)
    second = f"cc={k}"
    url = f"http://xx.xxx.xxx.xxx:xxxx/outputData?{first}&{second}&submit=Submit"
    
    async with session.get(url) as res:
        text = await res.text()
        soup = BeautifulSoup(text, 'lxml')
        print(soup)

async def main():
    # 用信号量控制并发量，避免请求过载
    semaphore = asyncio.Semaphore(20)
    
    async def bounded_send_request(session):
        async with semaphore:
            await send_request(session)
    
    async with aiohttp.ClientSession() as session:
        # 创建1000个异步任务
        tasks = [bounded_send_request(session) for _ in range(1000)]
        await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

这个方案的性能比多线程更好，尤其是请求数量极大时，速度提升会很明显。

方案3：使用多进程（multiprocessing）

多进程更适合CPU密集型任务，但如果你的脚本后续需要对响应做大量CPU处理（比如复杂解析），也可以考虑用它。不过纯网络请求场景下，多线程或异步已经足够高效，这里给个简单示例：

import requests
from bs4 import BeautifulSoup
from random import randint
from multiprocessing import Pool

def send_request(_):
    l = randint(2008, 2018)
    first = "year=%d" % l
    k = randint(1600, 2400)
    second = "cc=%d" % k
    url = f"http://xx.xxx.xxx.xxx:xxxx/outputData?{first}&{second}&submit=Submit"
    res = requests.get(url)
    soup = BeautifulSoup(res.text, 'lxml')
    print(soup)

if __name__ == "__main__":
    # 进程数一般设为CPU核心数的1-2倍
    with Pool(processes=4) as pool:
        pool.map(send_request, range(1000))

额外小建议