概述
kafka 是一个具有分布式特点的发布订阅的消息队列。可以系统解耦，流量削峰，异步通信，分布式存储。
具有相同消息队列功能的其他产品：MQ, REDIS 等。
名词解释（重点概念）
- producer: 生产者，由生产者负责把消息发送到 kafka 服务器。
- consumer: 消费者，由消费者从服务器上读取消息。
- consumer: 消费者组。
- Topic ： 订阅发布消息的频道。
- Broker: kafka 集群中的服务器。负责转发消息。
简单使用
kafka 默认监听的端口为9092
1 启动 zookeeper
./bin/zookeeper-server-start.sh conifg/zookeeper.properties
2 启动 kafka 服务(必须先启动zookeeper)
./bin/kafka-server-start.sh config/server.properties
kafka 命令行
1 创建一个topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
2 查看已经创建的topic
bin/kafka-topics.sh --list --zookeeper localhost:2181
3 创建生产者，并发送消息。
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
> hello word
4 创建消费者，现在消息队列中已经有了数据。
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
> hello world
python-kafka
kafka 提供了python 的工具包。
1 安装kafka 工具包。
pip install kafka
2 创建Producer 的demo
import json
from kafka import KafkaProducer
data = {"a": 1, "b": 2, "c":3}
producer = KafkaProducer(bootstrap_servers=["localhost:9092"],
api_version=(0, 10, 0),
value_serializer=lambda v: json.dumps(v).encode('utf-8'))
for i in range(1000):
producer.send("test", data, partition=0)
3 创建消费者脚本。
```python
from kafka import KafkaConsumer
consumer = KafkaConsumer('test',
group_id="group2",
bootstrap_servers=["localhost:9092"])
for msg in consumer:
print(msg.value)
命令行其他工具
- 查看topic 列表 ，包含当前服务器上所有的topic.
bin/kafka-topics.sh --zookeeper host ip:port --list
- 删除指定topic 需要在 server.properties 中设置 delete.topic.enable=true.
./bin/kafka-topics.sh --zookeeper host ip:port --delete --topic xxx
- 查看指定topic 的详细信息。
bin/kafka-topics.sh --zookeeper hsotip:port --describe --topic xx