背景：一个小应用，用celery下发任务，任务内容为kafka生产一些数据。

问题：使用confluent_kafka模块时，单独启用kafka可以正常生产消息，但是套上celery后，kafka就无法将新消息生产到topic队列中了。

解决：换了个pykafka模块，结果问题就没有了。

 

我很疑惑啊，是我调用confluent_kafka的方法不对吗，怎么套上celery就不行了呢？

 

可以用的pykafka代码：

tasks.py

from celery import Celeryfrom pykafka import KafkaClientimport jsonapp = Celery('tasks', backend='amqp', broker='amqp://xxx:xxxxxx@localhost/xxxhost')@app.taskdef produce():    client = KafkaClient(hosts="localhost:9092")    print client.topics    topic = client.topics['test_result']    with topic.get_sync_producer() as producer:        for i in range(3):            data = {
   "info": {
   "ip": "1.2.3.4", "port": i}, "type": "test", "task_id": "test_celery_kafka"}            print('Producing message: %s' % data)            producer.produce(json.dumps(data))        print "finish produce"        producer.stop()        print "stop"

run_worker.py

from tasks import producefor i in range(1000):    result = produce.delay()    print result.status

 

 

无法正常生产数据的confluent_kafka代码：

tasks.py

from celery import Celeryfrom kafka_producer import pimport jsonapp = Celery('tasks', backend='amqp', broker='amqp://xxx:xxxxxx@localhost/xxxhost')@app.taskdef produce():    for i in range(3000):        data = {
   "info": {
   "ip": "1.2.3.4"}, "type": "test", "task_id": "test_celery_kafka"}        print('Producing message: %s' % data)        p.produce('test_result3', json.dumps(data))    print "finish produce"    p.flush()    print "finish flush"

run_worker.py

from tasks import produceresult = produce.delay()print result.statusprint result.ready()print result.get()print result.status