如何使用 Python 最快实现 1w 个 http 请求

CC 攻击的脚本？

异步请求可以很快，同步要等着上一条返回，肯定是慢的。

@Puteulanus 不是，是一个小网站的秒杀活动

又查了一下，好像 aiohttp 就是异步请求的库？那可能是服务器的问题了。

好歹。。。贴代码吧

asyncio.gather

你需要一个异步 io 的库，或者随便自己写一个比如
https://gist.github.com/deplives/4b2d09421c5d0ee4aba75c69a5b9b070

本地实测请求 https://httpbin.org/status/200 1000 次只需要 2s

除了网络以外，将 py 编译成 cpython 也可以提高效率应该

@tool2d 不是我觉得是我写的代码有问题。我其实不需要 response

@infun 我代码很烂 就是普通请求。确实特别慢

@deplives 感谢我去看看

想问题的方向错了，人家小网站，你这不就是 DOS 攻击么，发得越快，崩得越快

不需要处理响应，直接用一些抓包工具重放不就好了。

@ospider 网站响应倒都是正常的 抗击打能力不错。只是我这并发写得不行

@tomczhen 有推荐的工具吗

不响应， 代码里面做了什么 io 操作吗？ 需求描述的不清楚。

哦，是我搞错了， 你是要发 1W 个 http 请求。

换 nodejs ！ python 就别干这种活了

并发快 崩的更快

@Maboroshii 我觉得是我还不够了解 python

我猜一下，是不是每个 request 的 futures 都分别 await 了？这样就成同步的了

别小看 python ，用 gevent.map ，1w 个请求一次就发了

@deplives 你这里应该是使用了底层的 socket 库来进行请求，效率确实很好 非常满足我的需求。只是有一点，我需要在请求是加上代理。类似于：
proxies = {
'http':f'{ip1}',
'https':f'{ip2}',
}
resp = requests.get(url="127.0.0.1",proxies=proxies}')

非常诚心的请问你这种方式可以实现吗

试试 tornado

https://www.tornadoweb.org/en/stable/httpclient.html

语言只是工具嘛，不如试试 Go

建议直接用 go 10 万都轻轻松松

@alsas 方便给个例子看下吗？

应该是用法不对

用异步方式, 没啥限制的, 顶多就是发出请求没响应, 但肯定能发的出去

用 go 写 很方便，

@gzldc #28 直接遍历 然后 在请求的前面加一个 go 就完事，这就是 go

你把代码发出来 我写一个看看呢

````
import aiohttp
import asyncio

# 读取文件并将每一行的 payload 存储到列表中
with open('payloads.txt', 'r') as file:
payloads = [line.strip() for line in file]

url = 'https://example.com/api' # 替换为您的目标 URL

async def send_request(payload):
async with aiohttp.ClientSession() as session:
try:
async with session.post(url, data=payload) as response:
# 不处理响应，仅打印状态码
print(f"Response status code: {response.status}")
except aiohttp.ClientError as e:
print(f"Error: {e}")

async def main():
tasks = [send_request(payload) for payload in payloads]
await asyncio.gather(*tasks)

# 运行主函数
asyncio.run(main())
```

参考一下，go 语言写的

```go
package main

import (
"bufio"
"fmt"
"net/http"
"net/url"
"os"
"strings"
"sync"
)

func main() {
file, err := os.Open("payloads.txt")
if err != nil {
fmt.Println("Error opening file:", err)
return
}
defer file.Close()

proxyURL, err := url.Parse("http://your-proxy-url:port")
if err != nil {
fmt.Println("Error parsing proxy URL:", err)
return
}

transport := &http.Transport{
Proxy: http.ProxyURL(proxyURL),
}

client := &http.Client{
Transport: transport,
}

var wg sync.WaitGroup

scanner := bufio.NewScanner(file)
for scanner.Scan() {
wg.Add(1)
go func(payload string) {
defer wg.Done()
sendRequest(client, payload)
}(scanner.Text())
}

wg.Wait()
}

func sendRequest(client *http.Client, payload string) {
url := "https://example.com/target"
req, err := http.NewRequest("GET", url, strings.NewReader(payload))
if err != nil {
fmt.Println("Error creating request:", err)
return
}

req.Header.Set("Content-Type", "application/json")

_, err = client.Do(req)
if err != nil {
fmt.Println("Error sending request:", err)
return
}
}
```

python3.10 aiohttp 5000 次 本地 大概需要 10s

import asyncio
import time

import aiohttp


def timed(func):
async def wrapper():
start = time.time()
await func()
print(f"costs: {time.time() - start:.3f} S")

return wrapper


async def req(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return resp.status


@timed
async def main():
tasks = [req('http://httpbin.org/status/200') for _ in range(5000)]
result = await asyncio.gather(*tasks)
print(len(result), ' status: ', result[-1])

有点灰产的意思，op 可交流？

请求太快把网站干死了还能秒不？

locust + FastHttpUser

go 实现

package main

import (
"bufio"
"fmt"
"net/http"
"os"
"sync"
)

// worker 是用来发起 HTTP 请求的协程
func worker(payloads <-chan string, wg *sync.WaitGroup) {
defer wg.Done()
for payload := range payloads {
// 这里是你的 HTTP 请求逻辑
_, err := http.Post("你的 URL", "application/json", strings.NewReader(payload))
if err != nil {
// 处理错误，根据需要记录或忽略
fmt.Println("Error sending request:", err)
continue
}
// 这里不处理响应体
}
}

func main() {
var wg sync.WaitGroup

// 创建一个 channel 来传递 payload
payloads := make(chan string, 100) // 缓冲区大小可以根据需要调整

// 设置并发数，可以根据你的机器性能和网络条件调整
for i := 0; i < 10; i++ {
wg.Add(1)
go worker(payloads, &wg)
}

// 打开文件
file, err := os.Open("yourfile.txt")
if err != nil {
fmt.Println("Error opening file:", err)
return
}
defer file.Close()

// 读取文件中的每一行作为 payload
scanner := bufio.NewScanner(file)
for scanner.Scan() {
payloads <- scanner.Text()
}
close(payloads) // 关闭 channel

// 等待所有的 worker 完成
wg.Wait()

if err := scanner.Err(); err != nil {
fmt.Println("Error reading from file:", err)
}
}

Python 实现

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def send_request(payload):
"""发送 HTTP POST 请求到指定的 URL"""
try:
response = requests.post('你的 URL', json=payload)
# 注意：这里我们不处理响应内容
except Exception as e:
print(f"请求发送异常: {e}")

def load_payloads(file_path):
"""从文件中读取每行作为单独的 payload"""
with open(file_path, 'r') as file:
return [line.strip() for line in file]

def main(file_path, max_workers=20):
"""主函数：读取 payloads 并并发发送请求"""
payloads = load_payloads(file_path)
start_time = time.time()
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# 将每个 payload 提交到线程池
future_to_payload = {executor.submit(send_request, payload): payload for payload in payloads}

# 等待所有的请求完成（这一步是可选的，只是为了能够处理可能的异常）
for future in as_completed(future_to_payload):
payload = future_to_payload[future]
try:
future.result() # 如果请求产生异常，这里会抛出
except Exception as exc:
print(f'{payload} 生成了一个异常: {exc}')

print(f"所有请求发送完毕，耗时：{time.time() - start_time:.2f}秒")

if __name__ == "__main__":
main("yourfile.txt")

@xiyou007
@NBNPC
@NotZhang

感谢 这对我帮助很大

加代理会慢吧。

单 ip 1w 个请求显然会被封 ip 小网站 200QPS 都不一定扛得住。

不如把代码贴出来给大家看看，正常情况下使用 aio ，几万个请求都轻轻松松

简单暴力法，nohup 尽量多启进程，十个八个的不嫌多，核心够多的话，二十三十的也不是不行

@gzldc #23 那你得自己改造一下

wrk 写个脚本应该挺快的?

wrk

有交流群么

一秒 10 个，应该是同步执行了，没有并发。并发参考 35 楼的写法。

@gzldc #23 有代理需求，执行脚本的时候 用 tsock 不就行了吗

@deplives 哥们的 git 库是隐藏了吗 方便再放出来一下吗

@gzldc #52 https://gist.github.com/bob-zebedy/4b2d09421c5d0ee4aba75c69a5b9b070