Python - 如何更简洁地使用生成器？

Question

（Python 3）

我正在使用 Python 生成器从队列中读取消息。

消费者读取队列消息后，如果队列消息处理成功，需要能够告诉生成器删除队列消息。

为了将 .send() 发送到 Python 生成器，看来我必须首先将 .send(None) 发送到生成器。这让我的代码比我想象的要胖。

谁能建议一种方法让 qconsumer.py 用更少的代码行来驱动生成器？我已经确定了我希望消除以下哪些行。

简而言之，我怎样才能使下面的代码更紧凑，关于如何删除行有什么建议吗？

下面的代码是qconsumer.py：

from qserver import Qserver

myqserver = Qserver()

myproducer = myqserver.producer() # trying to eliminate this line
# first send to a generator must be None
myproducer.send(None) # trying to eliminate this line
for msg in myproducer:
    # do something with message
    print(msg)
    if messageprocessok:
        myproducer.send('delete')

下面的代码是qserver.py：

# -*- coding: utf-8 -*-
import boto
from boto.sqs.connection import SQSConnection
from boto.sqs.message import Message

QNAME = 'qinbound'
SQSREGION = 'us-west-1'

class Qserver():
    """A simple Q server."""

    def __init__(self, qname=None, sqsregion=None):
        self.qname = qname or QNAME
        self.sqsregion = sqsregion or SQSREGION
        self.sqsconn = boto.sqs.connect_to_region(self.sqsregion)
        self.q_in = self.sqsconn.get_queue(self.qname)

    def producer(self):
        while True:
            qmessage = self.q_in.read(wait_time_seconds=20)
            if qmessage is None:
                continue
            action = (yield qmessage.get_body())
            if action == 'delete':
                # if processing completed ok, clear message from this queue
                self.q_in.delete_message(qmessage)

Answer 1

您当前的消费者正在丢弃消息，因为每个send 调用都会返回一个。你应该这样做：

myqserver = Qserver()
myproducer = myqserver.producer() 
messageprocessok = False
while True:
    msg = myproducer.send('delete' if messageprocessok else None)
    # do something with message
    print(msg)

或者：

myqserver = Qserver()
myproducer = myqserver.producer() 
msg = next(myproducer)
while True:
    # do something with message
    print(msg)
    msg = myproducer.send('delete' if messageprocessok else None)

您需要分别调用Qserver() 和myqserver.producer() 的事实仅仅是因为您将prouducer 设为类的方法。或者，您可以使用独立函数，或创建一个简单返回 Qserver().producer() 的包装函数。这是单机版：

def producer(qname=None, sqsregion=None):
    qname = qname or QNAME
    sqsregion = sqsregion or SQSREGION
    sqsconn = boto.sqs.connect_to_region(sqsregion)
    q_in = sqsconn.get_queue(qname)
    while True:
        qmessage = q_in.read(wait_time_seconds=20)
        if qmessage is None:
            continue
        action = (yield qmessage.get_body())
        if action == 'delete':
            # if processing completed ok, clear message from this queue
            q_in.delete_message(qmessage)

Answer 2

在了解了您要做什么之后，我想我会避免将 send 与迭代混为一谈。让myqserver 类成为迭代器本身对我来说似乎更有意义：

# -*- coding: utf-8 -*-
import boto
from boto.sqs.connection import SQSConnection
from boto.sqs.message import Message

QNAME = 'qinbound'
SQSREGION = 'us-west-1'

class Qserver():
    """A simple Q server."""
    _current_message = None

    def __init__(self, qname=None, sqsregion=None):
        self.qname = qname or QNAME
        self.sqsregion = sqsregion or SQSREGION
        self.sqsconn = boto.sqs.connect_to_region(self.sqsregion)
        self.q_in = self.sqsconn.get_queue(self.qname)

    def __iter__(self):
        return self

    def __next__(self):
        while True:
            qmessage = self.q_in.read(wait_time_seconds=20)
            if qmessage is not None:
                self._current_message = qmessage
                return qmessage

    next = __next__

    def delete_current(self):
        if self._current_message is not None:
            self.q_in.delete_message(self._current_message)

而用法会是这样的：

from qserver import Qserver

myqserver = Qserver()
for msg in myqserver:
    # do something with message
    print(msg)
    if messageprocessok:
        myqserver.delete_current()