Kafka操作工具。
Project description
使用手册
快速开始
1)生产者
import kafkaplus
producer = kafkaplus.getProducer("localhost:9092")
# 发送一条消息
producer.send("test_topic", "data-a")
# 同步发送一条消息
producer.sendSync("test_topic", "data-b")
# 批量发送消息
producer.sendBatch("test_topic", "data1", "data2", "data3")
2)消费者
import kafkaplus
# 接收消息的回调函数
def callback(data, **kwargs):
print(data)
print(kwargs)
consumer = kafkaplus.getConsumer("localhost:9092")
# 监听topic,使用回调函数处理消息
consumer.onListener("test_topic",group="G-test",callback=callback)
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
kafkaplus-0.0.1.tar.gz
(5.1 kB
view hashes)
Built Distributions
kafkaplus-0.0.1-py3.9.egg
(6.4 kB
view hashes)
Close
Hashes for kafkaplus-0.0.1-py2.py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 8a362238d9d9e35b78da104e2c6bb141cdc2550fdc6117edc9595f853aad5fda |
|
MD5 | 849973357dfcc9e8398a1f79bdc0612d |
|
BLAKE2b-256 | 76b49e69af080331e52fbdd5529c1e76185ff1534e35d7df3430c7be58ffd5a5 |