如何在Python中使单例smtp客户端线程安全？

Question

我不确定我是否问的是正确的问题，所以请容忍我。

我有一个Azure函数应用程序，运行python代码。它接收一个JSON有效负载，执行一些转换，并发送电子邮件。

一切正常工作，直到该函数开始接收到大量请求。这导致了异常：smtplib.SMTPDataError: (432, b'4.3.2 Concurrent connections limit exceeded. Visit https://aka.ms/concurrent_sending for more information. [Hostname=xxxxxxxx]')

我决定为连接客户端创建一个单例来解决这个问题：

class EmailSingleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls.__login(*args, **kwargs)
        else:
            try:
                status = cls._instances[cls].noop()[0]
            except smtplib.SMTPServerDisconnected:
                status = -1
            if status != 250:
                log.info("SMTP Client disconnected")
                cls.__login(*args, **kwargs)
        return cls._instances[cls]

    def __login(cls, host, port, user, password, timeout):
        log.info("Refreshing SMTP client")
        cls._instances[cls] = super(EmailSingleton, cls).__call__(host=host, port=port, timeout=timeout)
        cls._instances[cls].starttls()
        cls._instances[cls].login(user=user, password=password)

class EmailConnectionClientSingleton(smtplib.SMTP, metaclass=EmailSingleton):
    pass

并简单地通过调用：

EmailConnectionClientSingleton(
    host=host,
    port=port,
    timeout=timeout,
    user=user,
    password=password
).send_message(msg)

当请求一个接一个地出现时，这很好，但当请求突然到达时，我会得到各种有趣的例外情况：

• smtplib.SMTPDataError: (503, b'5.5.1 Bad sequence of commands')
• smtplib.SMTPDataError: (250, b'2.1.5 Recipient OK')
• smtplib.SMTPDataError: (250, b'2.0.0 OK')
• smtplib.SMTPRecipientsRefused: {'recipient@domain.com': (503, b'5.5.1 Bad sequence of commands')}

看起来，以上所有这些都是由于握手、重叠和反应不正常造成的。只有1/3的电子邮件成功。

我应该如何处理这个问题？我是否应该使用单例客户机(这告诉我应该使用：https://learn.microsoft.com/en-us/azure/azure-functions/manage-connections?tabs=csharp#static-clients)？我是否试图以某种方式使我的单例客户端线程安全？是否为send_message调用创建队列？我是否创建了一个从不同功能发送电子邮件的员工？

我尝试在创建客户机时使用_lock = threading.Lock()，但这并没有帮助，因为问题似乎在于使用客户机而不是创建客户机。

Answer 1

最后我使用了queue.Queue

class Singleton(type):
    _instance = None
    _lock = Lock()

    def __call__(cls, *args, **kwargs):
        if not cls._instance:
            with cls._lock:
                # another thread could have created the instance
                # before we acquired the lock. So check that the
                # instance is still nonexistent.
                if not cls._instance:
                    cls._instance = super().__call__(*args, **kwargs)
        return cls._instance


class QueueSingleton(Queue, metaclass=Singleton):
    pass

我没有立即发送电子邮件，而是把它放在队列中：

queue = QueueSingleton()
queue.put(msg)
log.info("Email put in the queue")

然后我创建了一个新的Azure函数，它使用队列发送电子邮件。

queue = QueueSingleton()

while start_time + FUNCTION_MAX_LIFE > datetime.now():
    try:
        msg = queue.get(timeout=300)
        log.info("Got an email to send!")
    except Empty:
        # If we didn't get anything to send during timeout window
        # end the execution and allow TimeTrigger to spawn a new worker
        log.info("No emails to send. Quitting.")
        break

    try:
        email_connection.send_message(msg)
        queue.task_done()
        log.info(f'Email sent: {msg["To"]}')

这感觉很烦人，但看起来很管用。