我正在尝试在 Python中实现反应器模式。我认为我有一个相当不错的开始使用multiprocessingand select.select。但是，我正在尝试对我的服务器进行压力测试，所以我编写了一个简单的 DoS 客户端来用连接淹没它。但我得到一个有趣的错误：
[WinError 10061] 由于目标机器主动拒绝，无法建立连接
有趣的是，我正在socket.listen(5)为服务器上的积压数量做准备。在我让读者准备好之后，我会select.select显示计数，但我只有 1 或 2 个 - 而不是我期望的 5 个。
对于少量线程（~20）我没有注意到它阻塞，但对于大量线程（50+）它确实倾向于拒绝连接。
我的问题是在服务器端还是客户端（或者只是在操作系统/套接字级别）？这是我可以解决的问题吗？如果是这样，怎么办？
这是我的代码：
客户
import threading
import time
import socket
from contextlib import contextmanager
IP = '127.0.0.1'
PORT = 4200
@contextmanager
def open_socket(ip, port):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sock.connect((ip, port))
yield sock
finally:
sock.close()
class Flood(threading.Thread):
def __init__(self, id):
super(Flood, self).__init__()
self.id = id
self.failed = False
def run(self):
try:
with open_socket(IP, PORT) as sock:
msg = "Hello this is some data from %d" % self.id
sock.send(msg.encode())
except Exception as e:
print(e)
self.failed = True
def make_threads(count):
return [Flood(_) for _ in range(count)]
threads = make_threads(5000)
start = time.time()
for t in threads:
t.start()
for t in threads:
t.join()
print("Failed: ", sum(1 if x.failed else 0 for x in threads))
print("Done in %f seconds" % (time.time() - start))
服务器
import sys
import logging
import socket
import select
import time
import queue
from multiprocessing import Process, Queue, Value
log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG)
log.addHandler(logging.StreamHandler())
IP = '127.0.0.1'
PORT = 4200
keep_running = True
def dispatcher(q, keeprunning):
try:
while keeprunning:
val = None
try:
val = q.get(True, 5)
if val:
log.debug(val[0].recv(1024).decode())
val[0].shutdown(socket.SHUT_RDWR)
val[0].close()
except queue.Empty:
pass
log.debug("Dispatcher quitting")
except KeyboardInterrupt:
log.debug("^C caught, dispatcher quitting")
def mainloop(sock):
readers, writers, errors = [sock], [], []
timeout = 5
while True:
readers, writers, errors = select.select(readers,
writers,
errors,
timeout)
incoming = yield readers, writers, errors
if incoming and len(incoming) == 3:
readers, writers, errors = incoming
if not readers:
readers.append(sock)
def run_server():
keeprunning = Value('b', True)
q = Queue()
p = Process(target=dispatcher, args=(q, keep_running))
try:
p.start()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((IP, PORT))
sock.listen(50)
sock.setblocking(0)
loop = mainloop(sock)
for readers, writers, errors in loop:
if readers:
client, addr = readers[0].accept()
q.put((client, addr))
log.debug('*'*50)
log.debug('%d Readers', len(readers))
log.debug('%d Writers', len(writers))
log.debug('%d Errors', len(errors))
except KeyboardInterrupt:
log.info("^C caught, shutting down...")
finally:
keeprunning.value = False
sock.close()
p.join()
if __name__ == "__main__":
if len(sys.argv) != 2:
print("Usage: test.py (client|server)")
elif sys.argv[1] == 'client':
run_client()
elif sys.argv[1] == 'server':
run_server()