Locust Python：Kafka消费者在单独的线程中

Question

我正在使用 locust 用于负载测试。我想在单独的线程中注册KAFKA消费者，以测量消息处理的时间。这是我现在得到的：

def register_kafka_consumer(topic, environment):
    def poll():
        request_type = 'KAFKA_MESSAGE_CONSUMED'
        consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id
        })
        consumer.subscribe([topic])
        logging.getLogger().info("Consumer subscribed to topic '%s': %s", topic, consumer)
        try:
            while True:
                msg = consumer.poll()
                if msg is not None and not msg.error():
                    # diff_millis calculation
                    environment.events.request.fire(
                        request_type=request_type,
                        name=topic,
                        response=msg.value(),
                        response_time=diff_millis,
                    )
        except Exception:
            logging.getLogger().error("Error during polling message for consumer: %s",
                                      str(consumer), exc_info=True)
        finally:
            logging.getLogger().info("Kafka consumer closed: %s", str(consumer))

    # here. I'm starting a separate thread
    t = threading.Thread(target=poll)
    t.start()
    logging.getLogger().info("Thread started")


@events.init.add_listener
def on_locust_init(environment, **kwargs):
    register_kafka_consumer(raw_topic, environment)

但是，根据日志，执行被困在中，而true循环。

[2022-05-11 17:20:53,309] INFO/locust.main: Starting web interface at http://localhost:8089
[2022-05-11 17:20:53,340] INFO/root: Consumer subscribed to topic 'some_topic': <cimpl.Consumer object at 0x10c28c510>

因为没有线程启动日志条目。

我在做什么错？有什么方法可以克服这个问题吗？

编辑1

我找到了此页面在蝗虫文档中，带有一个示例绿色的用法。这是我现在得到的。

def register_kafka_consumer(topic, environment):
    def poll():
        request_type = 'KAFKA_MESSAGE_CONSUMED'
        consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id
        })
        consumer.subscribe([topic])
        logging.getLogger().info("Consumer subscribed to topic '%s': %s", topic, consumer)
        try:
            while environment.runner.state not in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:
                time.sleep(1)
                msg = consumer.poll()
                if msg is not None and not msg.error():
                    # diff_millis calculation
                    environment.events.request.fire(
                        request_type=request_type,
                        name=topic,
                        response=msg.value(),
                        response_time=diff_millis,
                    )
        except Exception:
            logging.getLogger().error("Error during polling message for consumer: %s",
                                      str(consumer), exc_info=True)
        finally:
            logging.getLogger().info("Kafka consumer closed: %s", str(consumer))

    if not isinstance(environment.runner, WorkerRunner):
        gevent.spawn(poll)
        logging.getLogger().info("Greenlet started")


@events.init.add_listener
def on_locust_init(environment, **kwargs):
    register_kafka_consumer(raw_topic, environment)

根据日志，Kafka消费者现已注册。

[2022-05-12 12:39:03,310] INFO/locust.main: Starting web interface at http://localhost:8089
[2022-05-12 12:39:03,323] NFO/root: Greenlet started
[2022-05-12 12:39:03,323] INFO/locust.main: Starting Locust 2.8.2
[2022-05-12 12:39:03,379] INFO/root: Consumer subscribed to topic 'kafka_topic': <cimpl.Consumer object at 0x111ef4880>

尽管我现在无法打开Web UI。只有白屏和无限加载。也许这项Grenleet任务以某种方式占领了该过程。有什么想法如何解决这个问题？

Answer 1

好吧，我明白了。问题是由此陈述引起的：

msg = consumer.poll()

没有超时。因此，Python口译员永远等待，并且永远不会给其他机会执行。但是，如果我直接设置超时，一切都像魅力一样工作。这是最终版本。

def register_kafka_consumer(topic, environment):
    def poll():
        request_type = 'KAFKA_MESSAGE_CONSUMED'
        consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id
        })
        consumer.subscribe([topic])
        logging.getLogger().info("Consumer subscribed to topic '%s': %s", topic, consumer)
        try:
            while environment.runner.state not in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:
                time.sleep(1)
                # timeout in seconds
                msg = consumer.poll(timeout=0)
                if msg is not None and not msg.error():
                    # diff_millis calculation
                    environment.events.request.fire(
                        request_type=request_type,
                        name=topic,
                        response=msg.value(),
                        response_time=diff_millis,
                    )
        except Exception:
            logging.getLogger().error("Error during polling message for consumer: %s",
                                      str(consumer), exc_info=True)
        finally:
            logging.getLogger().info("Kafka consumer closed: %s", str(consumer))

    if not isinstance(environment.runner, WorkerRunner):
        gevent.spawn(poll)
        logging.getLogger().info("Greenlet started")


@events.init.add_listener
def on_locust_init(environment, **kwargs):
    register_kafka_consumer(raw_topic, environment)

当consumer.poll（）到达超时时，它只是返回none，循环再次启动。无论如何，我仍然不明白为什么python无法在没有超时的情况下切换poll的上下文。但是，问题现在已经解决。