一、准备工作二、终端命令
生产者命令消费者命令 三、Java实践
搭建项目异步发送-无回调异步发送-有回调同步发送
一、准备工作进入实战之前先熟悉一下topic的相关命令，使用终端命令查询创建一个新topic，用于后期实战；
特别注意：以下命令全部依据kafka文件目录中操作；
如果尚未安装kafka，请移步《centos7系统安装kafka》
查看操作主题命令参数
命令：./bin/kafka-topics.sh
日常工作中常用的命令如下表：
| 参数 | 描述 |
|---|---|
| –bootstrap-server | 连接的 Kafka Broker 主机名称和端口号。 |
| –topic | 操作的 topic 名称。 |
| –create | 创建主题 |
| –delete | 删除主题 |
| –alter | 修改主题 |
| –list | 查看所有主题 |
| –describe | 查看主题详细描述 |
| –partitions | 设置分区数。 |
| –replication-factor | 设置分区副本。 |
| –config | 更新系统默认的配置。 |
查看当前服务器中的所有topic
命令：bin/kafka-topics.sh --list --bootstrap-server localhost:9092
创建topic
命令：bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic kafka-test
选项说明：
- –topic 定义 topic 名–replication-factor 定义副本数–partitions 定义分区数
查看topic详情
命令：bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic kafka-test --describe
修改topic分区
命令：bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic kafka-test --partitions 3
注意：kafka一旦增加分区号，不支持减少分区数；
查看操作者命令参数
命令：bin/kafka-console-producer.sh
| 参数 | 描述 |
|---|---|
| –bootstrap-server | 连接的 Kafka Broker 主机名称和端口号。 |
| –topic | 操作的 topic 名称。 |
生产者发送消息
命令：bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka-test
消费者命令
查看操作消费者命令参数
命令：bin/kafka-console-consumer.sh
| 参数 | 描述 |
|---|---|
| –bootstrap-server | 连接的 Kafka Broker 主机名称和端口号。 |
| –topic | 操作的 topic 名称。 |
| –from-beginning | 从头开始消费。 |
| –group | 指定消费者组名称。 |
消费者消费消息
消费主题中的消息
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-test
主题中所有的数据都读取出来包括历史数据。
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic kafka-test
执行结果如图：
正式进入生产者代码实践之前，首先列举出生产者方大致的参数列表如下：
| 参数 | 解释说明 |
|---|---|
| bootstrap.servers | 生产者连接集群所需的 broker 地 址 清 单 。 例 如hadoop102:9092,hadoop103:9092,hadoop104:9092，可以设置 1 个或者多个，中间用逗号隔开。注意这里并非需要所有的 broker 地址，因为生产者从给定的 broker里查找到其他 broker 信息。 |
| key.serializer 和 value.serializer | 指定发送消息的 key 和 value 的序列化类型。一定要写全类名。 |
| buffer.memory | RecordAccumulator 缓冲区总大小，默认 32m。 |
| batch.size | 缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。 |
| linger.ms | 如果数据迟迟未达到 batch.size，sender 等待 linger.time之后就会发送数据。单位 ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 5-100ms 之间。 |
| acks | 0：生产者发送过来的数据，不需要等数据落盘应答。1：生产者发送过来的数据，Leader 收到数据后应答。-1（all）：生产者发送过来的数据，Leader+和 isr 队列里面的所有节点收齐数据后应答。默认值是-1，-1 和all 是等价的。 |
| max.in.flight.requests.per.connection | 允许最多没有返回 ack 的次数，默认为 5，开启幂等性要保证该值是 1-5 的数字。 |
| retries | 当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是 int 最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_ConNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。 |
| retry.backoff.ms | 两次重试之间的时间间隔，默认是 100ms。 |
| enable.idempotence | 是否开启幂等性，默认 true，开启幂等性。 |
| compression.type | 生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4 和 zstd。 |
1、引入kafka依赖
org.apache.kafka kafka-clients 3.0.0
2、打开消费者监听topic命令
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-test异步发送-无回调
代码如下：
public static void main(String[] args) {
// 1. 创建kafka生产者配置对象
Properties properties = new Properties();
// 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "XXXXXXX:9092");
// key,value 序列化（必须）：key.serializer，value.serializer
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
// 3. 创建 kafka 生产者对象
KafkaProducer kafkaProducer = new KafkaProducer(properties);
// 4. 调用 send 方法,异步发送消息，无回调接口
for (int i = 0; i < 5; i++) {
kafkaProducer.send(new ProducerRecord<>("kafka-test", "producerAsyncMessage-" + i));
}
// 5. 关闭资源
kafkaProducer.close();
}
执行结果如图：
public static void main(String[] args) throws InterruptedException {
// 1. 创建Kafka生产者的配置对象
Properties properties = new Properties();
// 2. 给kafka配置对象添加配置信息
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "xxxxxx:9092");
// 3. key 序列化 key.serializer，value.serializer
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 4. value 序列化 value.serializer
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 5. 创建kafka生产者对象
KafkaProducer kafkaProducer = new KafkaProducer(properties);
for (int i = 0; i < 5; i++) {
kafkaProducer.send(new ProducerRecord<>("kafka-test", "asyncCallbackMessage-" + i), new Callback() {
// 该方法在Producer 收到 ack 时调用，为异步调用
@Override
public void onCompletion(Recordmetadata recordmetadata, Exception e) {
if (e == null) {
// 没有异常，输出信息到控制台
System.out.println("topic：" + recordmetadata.topic() + " -> " + " 分区 " + recordmetadata.partition());
}else {
e.printStackTrace();
}
}
});
// 延迟发送是否会发送到不同分区
Thread.sleep(2);
}
// 5. 关闭资源
kafkaProducer.close();
}
消费者接收消息结果
生产者接收回调结果
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 1. 创建kafka生产者配置对象
Properties properties = new Properties();
// 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "xxxxxx:9092");
// key,value 序列化（必须）：key.serializer，value.serializer
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
// 3. 创建 kafka 生产者对象
KafkaProducer kafkaProducer = new KafkaProducer(properties);
// 4. 调用 send 方法,发送消息
for (int i = 0; i < 5; i++) {
Recordmetadata recordmetadata = kafkaProducer.send(new ProducerRecord<>("kafka-test", "syncMessage-" + i)).get();
System.out.println("send response result,topic：" + recordmetadata.topic() + ", " + " 分区:" + recordmetadata.partition());
}
// 5. 关闭资源
kafkaProducer.close();
}
消费者接收消息结果
[root@VM-4-8-centos kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-test producerAsyncMessage-0 producerAsyncMessage-1 producerAsyncMessage-2 producerAsyncMessage-3 producerAsyncMessage-4 asyncCallbackMessage-0 asyncCallbackMessage-1 asyncCallbackMessage-2 asyncCallbackMessage-3 asyncCallbackMessage-4 syncMessage-0 syncMessage-1 syncMessage-2 syncMessage-3 syncMessage-4
生产者接收发送结果
