blob: 823d9237318c640700fbf4e553054dc120e33acc (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
from itertools import cycle
import logging
from kafka.common import ProduceRequest
from kafka.protocol import create_message
log = logging.getLogger("kafka")
class SimpleProducer(object):
"""
A simple, round-robbin producer. Each message goes to exactly one partition
"""
def __init__(self, client, topic):
self.client = client
self.topic = topic
self.client.load_metadata_for_topics(topic)
self.next_partition = cycle(self.client.topic_partitions[topic])
def send_message(self, msg):
req = ProduceRequest(self.topic, self.next_partition.next(),
messages=[create_message(msg)])
resp = self.client.send_produce_request([req]).next()
|
