python - asyncore 关闭旧套接字

Question

我在 python 上使用 asyncore 获得了 tcp 服务器:

class AsyncClientHandler(asyncore.dispatcher_with_send):
    def __init__(self,sock):
        asyncore.dispatcher_with_send.__init__(self,sock)

        self.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        self.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

        self.message=""
        self.protocol=Protocol(DBSession, logger)

    def handle_read(self):
        data = self.recv(8192)
        if data:
            self.message+=data
            while TERMINATOR in self.message:   
                index=self.message.index(TERMINATOR)
                msg=self.message[:index]
                self.message=self.message[index+len(TERMINATOR):]

                answer=self.protocol.process_msg(msg, DBSession, tarif_dict)
                if answer:
                    msg = HEADER+answer+TERMINATOR
                    self.send(msg)

    def handle_close(self):
        self.close()

class AsyncServer(asyncore.dispatcher):
    def __init__(self, host, port): 
        asyncore.dispatcher.__init__(self)

        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)

        self.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        self.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

        self.set_reuse_addr()
        self.bind((host, port))
        self.listen(5)

    def handle_accept(self):
        pair = self.accept()
        if pair is None:
            pass
        else:
            sock, addr = pair
            logging.info("Incoming connection from %s",repr(addr))
            AsyncClientHandler(sock)

有些客户端并没有关闭连接，所以在某些时候服务器会因为大量的套接字而崩溃。

如何在一段时间后关闭不活动的套接字？ settimeout 不起作用。

Answer 1

要实现这一点，您可以使用 TCP 的 Keepalive(就像您已经做过的那样)并设置其延迟、ping...但是这个方法应该只用于持久连接并且只在 Unix 上可用。读一读 here .

您还可以对套接字进行一些调度，在一段时间后关闭它们，或者在它们处于事件状态时延迟它们。我用你的代码做了一个例子:

import sched, time

class SocketSched(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.daemon = True
        self.to_run = []
        self.scheds = {}
        self.start()

    def add(self, what):
        self.to_run.append(what.values()[0])
        self.scheds.update(what)

    def run(self):
        while True:
            if self.to_run:
                run = self.to_run.pop()
                if not run.empty(): run.run()
                else: self.to_run.append(run)

这里我们在不同的线程中定义了新的调度器类——这很重要，sched 模块会不断地阻塞，比如 asyncore.loop()。这需要稍微修改您的代码:

class AsyncClientHandler(asyncore.dispatcher_with_send):
    def __init__(self,sock, sch_class):
        ...
        self.delay = 10
        self.sch_class = sch_class
        self.sch = sched.scheduler(time.time, time.sleep)
        self.sch_class.add({self.fileno(): self.sch})
        self.event = self.sch_class.scheds[self.fileno()].enter(self.delay, 1, self.handle_close, ())

    ...

    def delay_close(self):
        self.sch_class.scheds[self.fileno()].cancel(self.event)
        self.event = self.sch_class.scheds[self.fileno()].enter(self.delay, 1, self.handle_close, ())

    ...

    def handle_close(self):
        try:
            self.sch_class.scheds[self.fileno()].cancel(self.event)
        except:
            pass
        ...

self.delay 是以秒为单位的超时。这段时间过去后，没有任何 Action 延迟它，套接字将关闭。 handle_close() 中的一行确保它不会由于调度程序中的任务而被调用两次。

现在您必须将 self.delay_close() 添加到确保套接字处于事件状态的每个方法的开头，例如。 handle_read().

服务器类(获取 SocketSched 实例并将其传递给新 channel ):

class AsyncServer(asyncore.dispatcher):
    def __init__(self, host, port, sch_class): 
        ...
        self.sch_class = sch_class

    ...

    def handle_accept(self):
    ...
            AsyncClientHandler(sock, self.sch_class)

准备好了。使用这个:

server = AsyncServer('', 1337, SocketSched())
asyncore.loop()

此解决方案有效，但在某些关闭事件上可能容易出错。无论如何，套接字将在给定超时发生时读取、延迟和关闭。不幸的是，运行这样的调度循环会占用一些 CPU。