1

我在 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 不起作用。

4

1 回答 1

0

为了实现这一点，你可以使用 TCP 的 Keepalive（就像你已经做过的那样）并设置它的延迟，pings...但是这个 apporach 应该只用于长期连接，并且只在 Unix 上可用。有一些在这里阅读。

您还可以对套接字进行一些调度，在一段时间过去时关闭它们或在它们处于活动状态时延迟它们。我做了一个使用您的代码的示例：

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是以秒为单位的超时。在这段时间过去后，没有任何动作延迟它，套接字将被关闭。handle_close() 中的行确保它不会因为调度程序中的任务而被调用两次。

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

服务器类（获取 SocketSched 实例并将其传递给新通道）：

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。

于 2013-08-03T15:47:52.853 回答