python - 如何在不阻塞任何东西的情况下处理 python 类中的 asyncore？

Question

我需要创建一个类来接收和存储 SMTP 消息，即电子邮件。为此，我根据发布的示例使用 asyncore here .但是，asyncore.loop() 处于阻塞状态，因此我无法在代码中执行任何其他操作。

于是我想到了使用线程。这是一个示例代码，显示了我的想法:

class MyServer(smtpd.SMTPServer):
    # derive from the python server class

    def process_message(..):
        # overwrite a smtpd.SMTPServer method to be able to handle the received messages
        ...
        self.list_emails.append(this_email)

    def get_number_received_emails(self):
        """Return the current number of stored emails"""
        return len(self.list_emails)


    def start_receiving(self):
        """Start the actual server to listen on port 25"""

        self.thread =   threading.Thread(target=asyncore.loop)
        self.thread.start()     

    def stop(self):
        """Stop listening now to port 25"""
        # close the SMTPserver from itself
        self.close()
        self.thread.join()

我希望你能明白这张照片。 MyServer 类应该能够以非阻塞方式启动和停止监听端口 25，能够在监听(或不监听)时查询消息。 start 方法启动 asyncore.loop() 监听器，当收到电子邮件时，它会附加到内部列表。类似地，stop 方法应该能够停止此服务器，如建议的那样 here .

尽管这段代码并不像我预期的那样工作(asyncore 似乎永远运行，即使我调用上面的 stop 方法。我提出的 error 被捕获在 stop 中，但不在包含 asyncore.loop() 的 target 函数中)，我不确定我解决问题的方法是否有意义.任何有关修复上述代码或提出更可靠的实现(不 使用第三方软件)的建议，我们都将不胜感激。

Answer 1

提供的解决方案可能不是最复杂的解决方案，但它工作合理并且已经过测试。

首先，asyncore.loop() 的问题在于它会阻塞，直到所有 asyncore channel 关闭，因为用户 Wessie之前在评论中指出。引用smtp example前面提到，结果是 smtpd.SMTPServer 继承自 asyncore.dispatcher(如 smtpd documentation 中所述)，它回答了要关闭哪个 channel 的问题。

因此，可以使用以下更新的示例代码来回答原始问题:

class CustomSMTPServer(smtpd.SMTPServer):
    # store the emails in any form inside the custom SMTP server
    emails = []
    # overwrite the method that is used to process the received 
    # emails, putting them into self.emails for example
    def process_message(self, peer, mailfrom, rcpttos, data):
        # email processing


class MyReceiver(object):
    def start(self):
        """Start the listening service"""
        # here I create an instance of the SMTP server, derived from  asyncore.dispatcher
        self.smtp = CustomSMTPServer(('0.0.0.0', 25), None)
        # and here I also start the asyncore loop, listening for SMTP connection, within a thread
        # timeout parameter is important, otherwise code will block 30 seconds after the smtp channel has been closed
        self.thread =  threading.Thread(target=asyncore.loop,kwargs = {'timeout':1} )
        self.thread.start()     

    def stop(self):
        """Stop listening now to port 25"""
        # close the SMTPserver to ensure no channels connect to asyncore
        self.smtp.close()
        # now it is save to wait for the thread to finish, i.e. for asyncore.loop() to exit
        self.thread.join()

    # now it finally it is possible to use an instance of this class to check for emails or whatever in a non-blocking way
    def count(self):
        """Return the number of emails received"""
        return len(self.smtp.emails)        
    def get(self):
        """Return all emails received so far"""
        return self.smtp.emails
    ....

所以最后，我有一个 start 和一个 stop 方法来在非阻塞环境中开始和停止监听端口 25。