bash

# 可选
$ export JAVA_HOME=/usr/java/jdk-11.0.10

## 安装 kafka
```bash
$ tar -zxf mv kafka_2.13-3.1.0.tgz
$ mv kafka_2.13-3.1.0.tgz /usr/local/kafka
$ mkdir /tmp/kafka-logs

# 生成随机cluster id
$ ./bin/kafka-storage.sh random-uuid
1QZShiaqQQCN8XE797uesg

# 格式化存储目录，注意我们使用的是kraft中的配置文件

./bin/kafka-storage.sh format -t 1QZShiaqQQCN8XE797uesg -c ./config/kraft/server.properties
Formatting /tmp/kraft-combined-logs

# 启动 kafka
$ ./bin/kafka-server-start.sh ./config/kraft/server.properties

# 创建名为test的topic
$ ./bin/kafka-topics.sh --create --topic test --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092

Created topic test.
# 查看topic
$ ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic test

Topic: test	TopicId: JyDAOV4AQ2mnCyD1Sh4DmA	PartitionCount: 1	ReplicationFactor: 1	Configs: segment.bytes=1073741824
	Topic: test	Partition: 0	Leader: 1	Replicas: 1	Isr: 1

# 生产消息到test主题(使用Ctrl-C停止生产者):
$ ./bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test
> test1
> test2

# 新开一个终端，消费来自test主题的消息:
# --from-beginning 是显示所有消息，而不是从最新的消息开始
$ ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

python

# pip install kafka-python
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')
for i in range(5):
  future = producer.send('test', b'finley %d' % i)
  result = future.get(timeout=10)
  print(result)