如何让kafka-python或pykafka作为uwsgi和gevent的异步生产者工作？

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了如何让kafka-python或pykafka作为uwsgi和gevent的异步生产者工作？，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含2988字，纯文字阅读大概需要5分钟。

内容图文

我的堆栈是uwsgi与gevents.我试图用装饰器包装我的api端点,将所有请求数据(url,方法,正文和响应)推送到kafka主题,但它不起作用.我的理论是因为我正在使用gevents,并且我试图在异步模式下运行它们,实际推送到kafka的异步线程无法与gevents一起运行.并且如果我尝试使方法同步,那么它也不起作用,它在产品工作者中死亡,即在产生调用后永远不会返回.虽然这两种方法在python shell上运行良好,但是我在线程上运行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

最佳答案:

我对pykafka有更多经验,所以我可以回答那一节. pykafka使用可插入的线程处理程序,并且内置了gevent支持.您需要使用use_greenlets = True来实例化KafkaClient. Docs here

关于你的方法的其他想法.为每条消息创建一个新的主题对象和生产者是非常昂贵的.最好创建一次并重用.

# setup once
client = KafkaClient(hosts=KAFKAHOST, use_greenlets=True)
topic = client.topics[topic]
producer = topic.get_sync_producer() 

def send_message_to_kafka(producer, key, message):
    """
    :param producer: pykafka producer
    :param key: key to decide partition
    :param message: json serializable object to send
    :return:
    """

    data = json.dumps(message)
    try:
        start = time.time()
        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 # for at least once delivery you will need to catch network errors and retry.

最后,kafka从批处理和压缩中获得了所有速度.使用同步生成器可以防止客户端利用这些功能.它会工作,但速度较慢,占用空间更多.某些应用程序需要同步,但如果您遇到性能瓶颈,则可能需要将应用程序重新考虑为批处理消息.

内容总结

以上是互联网集市为您收集整理的如何让kafka-python或pykafka作为uwsgi和gevent的异步生产者工作？全部内容，希望文章能够帮你解决如何让kafka-python或pykafka作为uwsgi和gevent的异步生产者工作？所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端