blob: eef113e9af85142f29a0fa9556c69072e0e0fd9c (
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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
|
#!/usr/bin/env python
from __future__ import print_function
import threading, logging, time
from kafka import KafkaConsumer, KafkaProducer
msg_size = 524288
producer_stop = threading.Event()
consumer_stop = threading.Event()
class Producer(threading.Thread):
big_msg = b'1' * msg_size
def run(self):
producer = KafkaProducer(bootstrap_servers='localhost:9092')
self.sent = 0
while not producer_stop.is_set():
producer.send('my-topic', self.big_msg)
self.sent += 1
producer.flush()
class Consumer(threading.Thread):
def run(self):
consumer = KafkaConsumer(bootstrap_servers='localhost:9092',
auto_offset_reset='earliest')
consumer.subscribe(['my-topic'])
self.valid = 0
self.invalid = 0
for message in consumer:
if len(message.value) == msg_size:
self.valid += 1
else:
self.invalid += 1
if consumer_stop.is_set():
break
consumer.close()
def main():
threads = [
Producer(),
Consumer()
]
for t in threads:
t.start()
time.sleep(10)
producer_stop.set()
consumer_stop.set()
print('Messages sent: %d' % threads[0].sent)
print('Messages recvd: %d' % threads[1].valid)
print('Messages invalid: %d' % threads[1].invalid)
if __name__ == "__main__":
logging.basicConfig(
format='%(asctime)s.%(msecs)s:%(name)s:%(thread)d:%(levelname)s:%(process)d:%(message)s',
level=logging.INFO
)
main()
|
