blob: 0eac0a5065865f10d8b9199d3a719188c84fad09 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
#!/usr/bin/env python
import threading, logging, time
from kafka.client import KafkaClient
from kafka.consumer import SimpleConsumer
from kafka.producer import SimpleProducer
class Producer(threading.Thread):
daemon = True
def run(self):
client = KafkaClient("localhost", 9092)
producer = SimpleProducer(client)
while True:
producer.send_messages('my-topic', "test")
producer.send_messages('my-topic', "\xc2Hola, mundo!")
time.sleep(1)
class Consumer(threading.Thread):
daemon = True
def run(self):
client = KafkaClient("localhost", 9092)
consumer = SimpleConsumer(client, "test-group", "my-topic")
for message in consumer:
print(message)
def main():
threads = [
Producer(),
Consumer()
]
for t in threads:
t.start()
time.sleep(5)
if __name__ == "__main__":
logging.basicConfig(level=logging.WARN)
main()
|
