Kafka 3.9.0：部署、监控与消息发送教程

文章目录

• Kafka 3.9.0 部署
• 可视化界面KafkaMap部署
• 测试消息发送

Kafka 3.9.0 部署

docker run -d \\
–name kafka \\
–hostname kafka \\
-p 9092:9092 \\
-e KAFKA_NODE_ID=1 \\
-e KAFKA_PROCESS_ROLES=broker,controller \\
-e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \\
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://<宿主机ip>:9092 \\
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \\
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \\
-e KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 \\
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \\
-e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \\
-e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \\
-e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \\
-e KAFKA_LOG_RETENTION_HOURS=168 \\
apache/kafka:3.9.0

注意：

1、需要把<宿主机ip>换成自己的

2、kafka高版本配置参数中不能出现0.0.0.0这样的ip地址，比如不能将KAFKA_ADVERTISED_LISTENERS配置为PLAINTEXT://0.0.0.0:9092这样的形式，否则会出现类似错误：

Exception in thread "main" java.lang.IllegalArgumentException: requirement failed: advertised.listeners cannot use the nonroutable meta-address 0.0.0.0. Use a routable IP address. at scala.Predef$.require(Predef.scala:337) at kafka.server.KafkaConfig

参数说明：

• KAFKA_PROCESS_ROLES: 启用 KRaft 模式，同时作为代理节点和控制器（替代 Zookeeper）
• KAFKA_LISTENERS: 定义 Kafka 在容器内部监听的端口
  • :9092 用于客户端通信。
  • :9093 用于控制器内部通信。
• KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://<宿主机ip>:9092: 外部客户端访问地址，如果缺少这行配置会导致包括之后的可视化界面等客户端无法连接
• KAFKA_CONTROLLER_QUORUM_VOTERS: 定义投票节点，这里指向自己
• KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1: 因为是单节点部署，必须将副本数设为 1，否则创建 Topic 时会报错或状态异常

可视化界面KafkaMap部署

docker run -d \\
-p 8080:8080 \\
-e DEFAULT_USERNAME=admin \\
-e DEFAULT_PASSWORD=admin \\
–name kafka-map \\
–restart always dushixiang/kafka-map:latest

部署完成后访问8080端口，默认账号密码都是admin

进入后点击右上角的Import Cluster然后填写配置

成功后的界面

测试消息发送

现在可以使用客户端来测试向 kafka 发送消息了，下面使用 Spring Boot 3.0.2 + kafka-client 来进行测试

1、引入依赖，版本号要部署的版本号一致

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.9.0</version>
</dependency>

2、编写测试类

@SpringBootTest
public class KafkaTest {

@Value("${kafka.topic}")
private String TOPIC; // topic名称，如果没有会自动创建
@Value("${kafka.bootstrap-servers}")
private String BOOTSTRAP_SERVERS; // kafka部署地址

@Test
public void testSendMessage() {

// 配置 Producer 属性
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

props.put(ProducerConfig.RETRIES_CONFIG, 0); // 不进行重试
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 2000); // 如果连接不上只等待2秒

// 创建 Producer
try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
String message = "Hello Kafka – " + System.currentTimeMillis();

// 发送消息
ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "test-key", message);
RecordMetadata metadata = producer.send(record).get();

System.out.println("成功发送消息到 topic: " + metadata.topic());
System.out.println("分区: " + metadata.partition() + ", 偏移量: " + metadata.offset());
System.out.println("消息内容: " + message);
} catch (Exception e) {
System.err.println("发送消息失败");
e.printStackTrace();
}
}
}

3、查看结果，可以看到发送成功，kafka也成功接收到了发送的消息