Python 性能 - 最佳并行方法

Question

我正在实现一个 Python 脚本，该脚本需要在每个不到 5 秒的时间内并行发送 1500 多个数据包。

简而言之，我需要的是:

def send_pkts(ip):
    #craft packet
    while True:
        #send packet
        time.sleep(randint(0,3))

for x in list[:1500]:
    send_pkts(x)
    time.sleep(randint(1,5))

我尝试了简单的单线程、多线程、多处理和多处理+多线程形式，但遇到了以下问题:

1. 简单的单线程:“for 延迟”似乎损害了“5 秒”的依赖性。
2. 多线程:由于 Python GIL 的限制，我认为我无法完成我想要的。
3. 多处理:这似乎是最有效的方法。但是，由于进程过多，我运行脚本的虚拟机卡住了(当然，运行了 1500 个进程)。因此变得不切实际。
4. 多处理+多线程:在这种方法中，我创建了更少的进程，每个进程调用一些线程(假设:10 个进程每个调用 150 个线程)。很明显，VM 的卡住速度不如方法 3，但是我能达到的最大“并发数据包发送”是 ~800。 GIL 限制？虚拟机限制？在这次尝试中，我也尝试使用进程池，但结果相似。

我可以使用更好的方法来完成这项任务吗？

[1] 编辑 1:

 def send_pkt(x):
     #craft pkt
     while True:
         #send pkt
         gevent.sleep(0)

 gevent.joinall([gevent.spawn(send_pkt, x) for x in list[:1500]])

[2] 编辑 2(gevent 猴子修补):

from gevent import monkey; monkey.patch_all()

jobs = [gevent.spawn(send_pkt, x) for x in list[:1500]]
gevent.wait(jobs)
#for send_pkt(x) check [1]

但是我收到以下错误:“ValueError:filedescriptor out of range in select()”。所以我检查了我的系统 ulimit(Soft 和 Hard 都是最大值:65536)。之后，我检查了它与 Linux 上的 select() 限制(最大 1024 fds)有关。请查收:http://man7.org/linux/man-pages/man2/select.2.html (BUGS 部分)- 为了克服这个问题，我应该改用 poll() (http://man7.org/linux/man-pages/man2/poll.2.html)。但是使用 poll() 我又回到了相同的限制:因为轮询是一种“阻塞方法”。

问候，

Answer 1

在 Python 中使用并行性时，一个好的方法是使用 ThreadPoolExecutor 或 ProcessPoolExecutor https://docs.python.org/3/library/concurrent.futures.html#module-concurrent.futures根据我的经验，这些效果很好。

可以根据您的使用进行调整的 threadedPoolExecutor 示例。

import concurrent.futures
import urllib.request
import time

IPs= ['168.212. 226.204',
        '168.212. 226.204',
        '168.212. 226.204',
        '168.212. 226.204',
        '168.212. 226.204']

def send_pkt(x):
  status = 'Failed'
  while True:
    #send pkt
    time.sleep(10)
    status = 'Successful'
    break
  return status

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    future_to_ip = {executor.submit(send_pkt, ip): ip for ip in IPs}
    for future in concurrent.futures.as_completed(future_to_ip):
        ip = future_to_ip[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (ip, exc))
        else:
            print('%r send %s' % (url, data))