Kafka作为分布式消息系统，其生产者是数据管道的起点。掌握kafka-console-producer.sh工具的使用对于开发测试和运维都至关重要。本文将系统介绍该工具的各种用法，帮助您高效地向Kafka发送消息。
1 基础消息生产
1.1 最简单的消息发送
/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-producer.sh \
--bootstrap-server 192.168.10.33:9092 \
--topic testtopic
参数解析：
- --bootstrap-server: 指定Kafka集群地址
- --topic: 指定目标主题名称
特点：
- 交互式输入，每行作为一条消息
- 按Ctrl+C退出
- 消息无key，自动分配分区
1.2 从文件批量导入
/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-producer.sh \
--bootstrap-server 192.168.10.33:9092 \
--topic testtopic \
< messages.txt
特点：
- 将文件内容逐行作为消息发送
- 适合数据迁移和批量测试
2 消息结构控制
2.1 带Key的消息发送
/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-producer.sh \
--bootstrap-server 192.168.10.33:9092 \
--topic testtopic \
--property parse.key=true \
--property key.separator="\t"
关键参数：
- parse.key=true: 启用key解析
- key.separator: 指定key/value分隔符（示例为制表符）
# 输入格式：
key1 value1
key2 value2
2.2 自定义分隔符
--property key.separator=":"
特点：
- 支持任意单字符分隔符
- 确保分隔符不在key/value中出现
3 消息传递控制
3.1 指定目标分区
特点：
3.2 同步发送模式
特点：
4 高级生产者配置
4.1 消息压缩配置
--property compression.type=gzip
可选值：
- none(默认)
- gzip
- snappy
- lz4
- zstd
4.2 时间戳配置
--property timestamp.type=CreateTime
选项：
- CreateTime: 生产者创建时间（默认）
- LogAppendTime: broker接收时间
5 可靠性配置
/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-producer.sh \
--bootstrap-server 192.168.10.33:9092 \
--topic testtopic \
--producer-property acks=all \
--producer-property min.insync.replicas=2 \
--producer-property enable.idempotence=true
关键参数：
- acks=all: 等待所有ISR确认
- min.insync.replicas: 最小同步副本数
- enable.idempotence: 启用幂等性
6 性能优化配置
/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-producer.sh \
--bootstrap-server 192.168.10.33:9092 \
--topic testtopic \
--producer-property batch.size=65536 \
--producer-property linger.ms=20 \
--producer-property compression.type=lz4
优化参数：
- batch.size: 增大批次大小（字节）
- linger.ms: 适当增加等待时间
- compression.type: 选择高效压缩算法
