Django配置kafka消息队列的实现

pip install confluent-kafka

KAFKA_SETTINGS = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest',
}

from confluent_kafka import Consumer, KafkaError
from django.conf import settings
def kafka_handler():
    c = Consumer(settings.KAFKA_SETTINGS)
    c.subscribe(['my-topic'])
    while True:
        msg = c.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                print('End of partition reached')
            else:
                print('Error: {}'.format(msg.error()))
        else:
            print('Received message: {}'.format(msg.value()))

if __name__ == '__main__':
    from myapp.kafka_handler import kafka_handler
    kafka_handler()

from confluent_kafka import Producer
from django.conf import settings
def send_message(message):
    p = Producer(settings.KAFKA_SETTINGS)
    topic = 'my-topic'
    p.produce(topic, message.encode('utf-8'))
    p.flush()