【问题标题】:Twisted execute 10 threads in same time and wait for resultTwisted 同时执行 10 个线程并等待结果
【发布时间】:2014-10-06 09:48:01
【问题描述】:

编写一个程序来验证电子邮件语法列表和 MX 记录,因为阻塞编程很耗时,我想异步或通过线程执行此操作,这是我的代码:

with open(file_path) as f:
    # check the status of file, if away then file pointer will be at the last index
    if (importState.status == ImportStateFile.STATUS_AWAY):
        f.seek(importState.fileIndex, 0)

    while True:
        # the number of emails to process is configurable 10 or 20
        emails = list(islice(f, app.config['NUMBER_EMAILS_TO_PROCESS']))
        if len(emails) == 0:
            break;

        importState.fileIndex = importState.fileIndex + len(''.join(emails))

        for email in emails:
            email = email.strip('''<>;,'\r\n ''').lower()
            d = threads.deferToThread(check_email, email)
            d.addCallback(save_email_status, email, importState)

        # set the number of emails processed 
        yield set_nbrs_emails_process(importState)

        # do an insert of all emails
        yield reactor.callFromThread(db.session.commit)

# set file status as success
yield finalize_import_file_state(importState)
reactor.callFromThread(reactor.stop)

查收邮件功能:

def check_email(email):
    pipe = subprocess.Popen(["./check_email", '--email=%s' % email], stdout=subprocess.PIPE)
    status = pipe.stdout.read()
    try:
        status = int(status)
    except ValueError:
        status = -1

    return status

我需要的是同时处理 10 封电子邮件并等待结果。

【问题讨论】:

  • 是否有 10 封电子邮件,或者您希望同时发送不超过 10 封电子邮件?
  • 您的代码中是否有 @inlineCallBacks 装饰器(由所有 yield 语句暗示)?
  • 是的,有@inlineCallBacks,我想批量处理10或20封邮件,然后插入DB。
  • 如果您不关心限制并发处理的电子邮件数量,那么只需按照@Jean-Paul Calderone 的建议使用gatherResults()
  • 您可能应该打开一个单独的问题,关于如何发送一封电子邮件而不会阻止扭曲。

标签: python multithreading python-2.7 twisted


【解决方案1】:

我不确定为什么您的示例代码中涉及线程。您不需要线程与 Twisted 的电子邮件交互,也不需要同时进行。

如果您有一个返回Deferred 的异步函数,您只需调用它十次,十个不同的工作流将并行进行:

for i in range(10):
    async_check_email_returning_deferred()

如果您想知道所有十个结果何时都可用,您可以使用gatherResults:

from twisted.internet.defer import gatherResults
...
email_results = []
for i in range(10):
    email_results.append(async_check_mail_returning_deferred())
all_results = gatherResults(email_results)

all_results 是一个Deferred,当email_results 中的所有Deferreds 都被触发(或者当它们中的第一个被Failure 触发时)将触发。

【讨论】:

  • 你能把这个函数的代码给async_check_email_returning_deferred
  • 这是任何返回延迟的函数。我不确定您所说的“检查电子邮件”到底是什么意思,所以我不知道这个功能实际上是如何实现的。不过,它可能会使用来自twisted.mail 的一些 API。
  • 我添加了检查功能
  • 不幸的是,该函数没有返回 Deferred - 实际上根本不是异步的。它正在阻塞。所以你可以使用的唯一简洁的并发工具是线程。如果您将async_check_mail_returning_deferred 定义为return deferToThread(check_email, email) 之类的东西,此答案仍然适用,因为它使用线程将您的阻塞函数变为异步的延迟返回函数。尽管那时您正在混合线程和进程,这始终是一件冒险的事情。考虑使用 Twisted Mail 而不是 subprocess 模块。
猜你喜欢
  • 1970-01-01
  • 2022-12-05
  • 1970-01-01
  • 2022-12-18
  • 1970-01-01
  • 1970-01-01
  • 2016-11-13
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多