1、生产者发送消息

1.1、使用EFAK创建主题my_topic3

1.2、根据kafka官网文档写代码

https://kafka.apache.org/

[root@localhost ~]# /opt/kafka/bin/kafka-configs.sh --version
3.0.0 (Commit:8cb0a5e9d3441962)

1.3、pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.0.5</version><relativePath/> <!-- lookup parent from repository --></parent><!-- Generated by https://start.springboot.io --><!-- 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn --><groupId>com.atguigu.kafka</groupId><artifactId>kafka-producer</artifactId><version>0.0.1-SNAPSHOT</version><name>kafka-producer</name><description>kafka-producer</description><properties><java.version>17</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.0.0</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

1.4、KafkaProducer1.java

package com.atguigu.kafka.producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class KafkaProducer1 {/*** 主函数用于演示如何向Kafka的特定主题发送消息。** @param args 命令行参数（未使用）*/public static void main(String[] args) {// 初始化Kafka生产者配置Properties props = new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.74.148:9092"); // 指定Kafka broker的地址和端口props.put("acks", "all"); // 确认消息写入策略props.put("retries", 0); // 消息发送失败时的重试次数props.put("linger.ms", 1); // 发送缓冲区等待时间props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 指定键的序列化器props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 指定值的序列化器// 创建Kafka生产者实例Producer<String, String> producer = new KafkaProducer<>(props);// 发送消息到主题"my_topic3"producer.send(new ProducerRecord<String, String>("my_topic3",  "hello,1,2,3"));// 注释掉的循环代码块展示了如何批量发送消息//for (int i = 0; i < 100; i++)//    producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));// 关闭生产者实例producer.close();}}

1.5、使用EFAK查看主题

1.6、再次运行KafkaProducer1.java

1.7、再次使用EFAK查看主题