Python Kafka 多进程与线程

Question

我可以使用KafkaConsumer 在不同的线程中消费消息。

但是，当我使用 multiprocessing.Process 而不是 threading.Thread 时，出现错误：

OSError: [Errno 9] Bad file descriptor

question 和 documentation 建议使用多处理并行处理消息是可能的。有人可以分享一个工作示例吗？

编辑

这里有一些示例代码。抱歉，原始代码太复杂了，所以我在这里创建了一个示例，希望能传达正在发生的事情。如果我使用 threading.Thread 而不是 multiprocessing.Process，此代码可以正常工作。

from multiprocessing import Process

class KafkaWrapper():
    def __init__(self):
        self.consumer = KafkaConsumer(bootstrap_servers='my.server.com')

    def consume(self, topic):
        self.consumer.subscribe(topic)
        for message in self.consumer:
            print(message.value)

class ServiceInterface():
    def __init__(self):
        self.kafka_wrapper = KafkaWrapper()

    def start(self, topic):
        self.kafka_wrapper.consume(topic)

class ServiceA(ServiceInterface):
    pass

class ServiceB(ServiceInterface):
    pass


def main():

    serviceA = ServiceA()
    serviceB = ServiceB()

    jobs=[]
    # The code works fine if I used threading.Thread here instead of Process
    jobs.append(Process(target=serviceA.start, args=("my-topic",)))
    jobs.append(Process(target=serviceB.start, args=("my-topic",)))

    for job in jobs:
        job.start()

    for job in jobs:
        job.join()

if __name__ == "__main__":
    main()

这是我看到的错误（同样，我的实际代码与上面的示例不同，如果我使用 threading.Thread 可以正常工作，但如果我使用 multiprocessing.Process 则不行）：

File "/usr/local/Cellar/python3/3.6.2/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/usr/local/Cellar/python3/3.6.2/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "service_interface.py", line 58, in start
    self._kafka_wrapper.start_consuming(self.service_object_id)
  File "kafka_wrapper.py", line 141, in start_consuming
    for message in self._consumer:
  File "venv/lib/python3.6/site-packages/kafka/consumer/group.py", line 1082, in __next__
    return next(self._iterator)
  File "venv/lib/python3.6/site-packages/kafka/consumer/group.py", line 1022, in _message_generator
    self._client.poll(timeout_ms=poll_ms, sleep=True)
  File "venv/lib/python3.6/site-packages/kafka/client_async.py", line 556, in poll
    responses.extend(self._poll(timeout, sleep=sleep))
  File "venv/lib/python3.6/site-packages/kafka/client_async.py", line 573, in _poll
    ready = self._selector.select(timeout)
  File "/usr/local/Cellar/python3/3.6.2/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/process.py", line 249, in _bootstrap
    self.run()
  File "/usr/local/Cellar/python3/3.6.2/Frameworks/Python.framework/Versions/3.6/lib/python3.6/selectors.py", line 577, in select
    kev_list = self._kqueue.control(None, max_ev, timeout)
  File "/usr/local/Cellar/python3/3.6.2/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "service_interface.py", line 58, in start
    self._kafka_wrapper.start_consuming(self.service_object_id)
  File "kafka_wrapper.py", line 141, in start_consuming
    for message in self._consumer:
  File "venv/lib/python3.6/site-packages/kafka/consumer/group.py", line 1082, in __next__
    return next(self._iterator)
  File "venv/lib/python3.6/site-packages/kafka/consumer/group.py", line 1022, in _message_generator
    self._client.poll(timeout_ms=poll_ms, sleep=True)
  File "venv/lib/python3.6/site-packages/kafka/client_async.py", line 556, in poll
    responses.extend(self._poll(timeout, sleep=sleep))
OSError: [Errno 9] Bad file descriptor
  File "venv/lib/python3.6/site-packages/kafka/client_async.py", line 573, in _poll
    ready = self._selector.select(timeout)
  File "/usr/local/Cellar/python3/3.6.2/Frameworks/Python.framework/Versions/3.6/lib/python3.6/selectors.py", line 577, in select
    kev_list = self._kqueue.control(None, max_ev, timeout)
OSError: [Errno 9] Bad file descriptor

Answer 1

Kafka 消费者可以是多进程或多线程（确保使用的客户端库正确支持 Kafka Consumer Group，这是 Kafka 早期版本所必需的），由您选择。

但是，如果我们想使用进程，Kafka 客户端库需要做一些事情，以保证自己 fork 安全，使用的底层 TCP 连接（连接到 Kafka 服务器）不应该被共享不止一个进程。这就是您遇到连接错误的原因。

作为一种解决方法，您不应在生成进程之前创建 KafkaConsumer。而是将操作移到每个进程中。

另一种方法是使用单线程/进程获取消息，并使用额外的进程池来执行实际操作。