我的堆栈是带有gevents的uwsgi。我试图用一个decorator来 Package 我的api端点，以便将所有请求数据（url、方法、主体和响应）推送到kafka主题，但它不起作用。我的理论是因为我使用的是gevents，我试图在异步模式下运行它们，而实际推送到kafka的异步线程不能用gevents运行。如果我尝试使方法同步，那么它也不工作，它在produce worker中死亡，即在produce之后，调用永远不会返回。尽管这两种方法在pythonshell和在线程上运行uwsgi时效果都很好。
遵循示例代码：1。使用kafka python（异步）

try:
        kafka_producer = KafkaProducer(bootstrap_servers=KAFKAHOST.split(','))
    except NoBrokersAvailable:
        logger.info(u'Kafka Host not available: {}'.format(KAFKAHOST))
        kafka_producer = None

    def send_message_to_kafka(topic, key, message):
        """
        :param topic: topic name
        :param key: key to decide partition
        :param message: json serializable object to send
        :return:
        """
        if not kafka_producer:
            logger.info(u'Kafka Host not available: {}'.format(KAFKAHOST))
            return
        data = json.dumps(message)
        try:
            start = time.time()
            kafka_producer.send(topic, key=str(key), value=data)
            logger.info(u'Time take to push to Kafka: {}'.format(time.time() - start))
        except KafkaTimeoutError as e:
            logger.info(u'Message not sent: {}'.format(KAFKAHOST))
            logger.info(e)
            pass
        except Exception as e:
            logger.info(u'Message not sent: {}'.format(KAFKAHOST))
            logger.exception(e)
            pass

与py kafka（同步）：

try:
    client = KafkaClient(hosts=KAFKAHOST)
except Exception as e:
    logger.info(u'Kafka Host Not Found: {}'.format(KAFKAHOST))
    client = None

def send_message_to_kafka(topic, key, message):
    """
    :param topic: topic name
    :param key: key to decide partition
    :param message: json serializable object to send
    :return:
    """
    if not client:
        logger.info(u'Kafka Host is None')
        return
    data = json.dumps(message)
    try:
        start = time.time()
        topic = client.topics[topic]
        with topic.get_sync_producer() as producer:
            producer.produce(data, partition_key='{}'.format(key))
        logger.info(u'Time take to push to Kafka: {}'.format(time.time() - start))
    except Exception as e:
        logger.exception(e)
        pass