华为云分布式消息服务Kafka版对接使用全攻略：从实例创建到生产消费
1. 引言：为什么选择华为云分布式消息服务Kafka版
在微服务架构与大数据处理日益普及的今天，消息队列已成为系统间解耦、流量削峰、异步通信的核心基础设施。Apache Kafka凭借其高吞吐、持久化、水平扩展等特性，在日志收集、流式数据处理、实时监控等场景中占据主导地位。华为云分布式消息服务Kafka版（Distributed Message Service for Kafka）在完全兼容开源Kafka的基础上，提供了全托管、高可用、弹性扩展的企业级消息队列服务。
相较于自建Kafka集群，华为云Kafka版具备以下显著优势：全托管服务让用户无需关心底层服务器的部署与运维，只需聚焦业务逻辑即可；数据支持多副本保存，单节点故障不影响业务使用，服务可用性高达99.95%；Kafka队列按需自动扩展，按实际使用量计费，资源利用率更高；支持亿级消息堆积，在海量堆积下不影响队列性能，单队列并发最高可超过10万TPS；支持SASL_SSL机制对身份认证和数据通道加密传输，确保数据传输过程中不被窃取或篡改。
本文将从零开始，手把手带您完成华为云Kafka实例的创建、客户端对接、生产消费测试、监控告警配置以及常见问题排查，帮助您快速上手并稳定运行业务。
2. 准备工作：创建Kafka实例前的资源规划
在创建Kafka实例之前，需要提前准备好相关的依赖资源。Kafka实例运行于虚拟私有云（VPC）中，因此必须确保已有可用的VPC和子网。Kafka实例可以使用当前账号下已创建的VPC和子网，也可以新创建，但需要注意VPC与Kafka实例必须在相同的区域。如果子网开启了IPv6功能，Kafka实例也将支持IPv6，客户端可以使用IPv6地址连接实例。
安全组是另一个必不可少的资源。不同的Kafka实例可以重复使用相同的安全组，也可以使用不同的安全组。在连接Kafka实例前，需要根据连接方式配置对应的安全组规则。以客户端和Kafka实例的安全组分别为sg-53d4和Default_All为例，客户端所在安全组需要增加允许访问Kafka实例端口的规则。安全组创建后，系统默认添加入方向允许安全组内的弹性云服务器彼此通信的规则。
如果需要通过公网访问Kafka实例，还需要提前准备弹性IP地址。弹性IP地址的数量必须与Kafka实例的代理个数相同，且与Kafka实例在相同的区域。需要注意的是，Kafka实例无法识别开启IPv6转换功能的弹性IP地址。
如果需要对云上的资源进行精细管理，建议使用IAM服务创建IAM用户及用户组并授权，以使得IAM用户获得具体的操作权限。
需要先登录华为云控制台，点击：华为云控制台，还没有账号，点击：注册并关联，已有账号点击：登录后关联
3. 创建Kafka实例：控制台操作全流程
完成资源准备后，即可开始创建Kafka实例。首先登录管理控制台，在左上角选择区域，不同区域的云服务产品之间内网互不相通，应就近选择靠近业务侧的区域以减少网络时延。在管理控制台左上角选择“应用服务 > 分布式消息服务 Kafka”，进入Kafka实例列表页面，然后单击页面右上方的“购买Kafka实例”。
3.1 基础配置
在购买页面中，首先需要选择计费模式，包括按需计费和包年/包月两种。然后设置实例名称和企业项目。在可用区配置方面，根据实际情况选择1个或者3个及以上可用区，实例创建成功后不支持修改可用区。
3.2 规格选型
华为云Kafka实例兼容开源Kafka 1.1.0、2.3.0和2.7版本。实例规格以实例提供的ECS规格和代理个数分类，包括kafka.2u4g.cluster、kafka.4u8g.cluster、kafka.8u16g.cluster、kafka.12u24g.cluster和kafka.16u32g.cluster共5种类型。在选择实例规格时，需要综合考虑业务预期的吞吐量、消息大小和并发量。下表中TPS性能是指以1K大小的消息为例的每秒处理消息条数。
磁盘类型和单个代理磁盘容量根据实际业务需求配置，磁盘容量仅支持设置为100的整数倍，创建后磁盘类型不支持修改。存储空间用于存储消息（包括副本中的消息）、日志和元数据。Kafka实例支持多副本存储，存储空间包含所有副本存储空间总和。例如，业务消息体积预估为100GB，则磁盘容量最少应为100GB乘以副本数再加上预留磁盘大小100GB。每个Kafka代理会预留33GB的磁盘空间用于存储日志和元数据。创建实例时进行磁盘格式化，实际可用磁盘为总磁盘的93%至95%。Kafka实例支持对存储进行扩容，可根据业务增长随时扩容以节约成本。
3.3 网络与认证配置
在VPC和子网配置中，选择已准备好的VPC和子网。安全组选择已创建的安全组。如果客户端使用公网连接，还需要绑定弹性IP地址。
关于SASL认证的开启，需要在创建实例时进行设置，实例创建成功后不支持修改SASL开关。开启SASL后数据加密传输，安全性更高。SASL认证支持PLAIN和SCRAM-SHA-512两种机制。如果同时启用了两种机制，在连接配置中可以使用任意一种。对于早期创建的实例，如果实例详情页面未显示SASL机制，则默认使用PLAIN。
3.4 Topic配置
在创建实例时可以选择是否开启“Kafka自动创建Topic”功能。如果开启，在生产消息时会自动创建一个Topic，默认参数为：分区数为3，副本数为3，老化时间为72小时，不开启同步复制和同步落盘。如果未开启该功能，则需要在实例创建完成后手动创建Topic。为保证性能，建议每个Topic的分区数控制在200以内。
4. 获取连接信息：对接前的关键步骤
Kafka实例创建成功后，在对接客户端之前，需要先获取实例的连接信息。
连接地址与端口：在Kafka控制台的实例详情页“概览”页面可以获取连接地址和端口。对于集群实例（至少3个代理），建议在客户端配置所有连接地址以提高可靠性。公网访问时使用概览页面显示的公网地址。内网访问的端口为9092，公网访问的端口为9094。
Topic名称：在实例的Topic页面获取Topic名称。如果未开启自动创建Topic，需要先手动创建Topic。
认证信息：如果实例开启了密文访问，需要获取实例的用户名、密码、SASL机制和安全协议。用户名可在实例的“用户”页面获取。如果忘记了密码，可以通过重置密码的方式重新获取。安全协议可在概览页面获取。对于早期创建的实例，如果实例详情页面未显示安全协议，则默认使用SASL_SSL。当安全协议为SASL_SSL时，还需要在概览页面下载SSL证书。Java客户端使用JKS格式证书，Python客户端使用CRT格式证书。
5. 客户端对接：Java原生客户端生产消费示例
获取连接信息后，即可开始编写客户端代码进行消息的生产和消费。华为云Kafka版支持在控制台直接查看Java、Go和Python语言的生产消费示例代码。
以下以Java原生客户端为例，展示如何连接华为云Kafka实例并进行消息生产与消费。
5.1 添加Maven依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.7.0</version>
</dependency>5.2 生产者代码
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
public class KafkaProducerExample {
public static void main(String[] args) {
Properties props = new Properties();
// 连接地址，从控制台概览页获取
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"ip1:port1,ip2:port2,ip3:port3");
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, 10);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
// 如果开启了SASL认证，添加以下配置
// 安全协议：SASL_PLAINTEXT或SASL_SSL
props.put("security.protocol", "SASL_PLAINTEXT");
// SASL机制：PLAIN或SCRAM-SHA-512
props.put("sasl.mechanism", "PLAIN");
// SASL用户名和密码
String jaasTemplate =
"org.apache.kafka.common.security.plain.PlainLoginModule required " +
"username=\"your_username\" password=\"your_password\";";
props.put("sasl.jaas.config", jaasTemplate);
// 如果安全协议为SASL_SSL，还需要配置SSL证书
// props.put("ssl.truststore.location", "/path/to/client.jks");
// props.put("ssl.truststore.password", "your_truststore_password");
try (KafkaProducer<String, String> producer =
new KafkaProducer<>(props)) {
for (int i = 0; i < 100; i++) {
ProducerRecord<String, String> record =
new ProducerRecord<>("your_topic_name",
"key-" + i, "value-" + i);
producer.send(record, (metadata, exception) -> {
if (exception != null) {
exception.printStackTrace();
} else {
System.out.printf("Sent message to topic %s partition %d " +
"offset %d%n",
metadata.topic(), metadata.partition(),
metadata.offset());
}
});
}
producer.flush();
}
}
}5.3 消费者代码
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class KafkaConsumerExample {
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"ip1:port1,ip2:port2,ip3:port3");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
// SASL认证配置同生产者
props.put("security.protocol", "SASL_PLAINTEXT");
props.put("sasl.mechanism", "PLAIN");
String jaasTemplate =
"org.apache.kafka.common.security.plain.PlainLoginModule required " +
"username=\"your_username\" password=\"your_password\";";
props.put("sasl.jaas.config", jaasTemplate);
try (KafkaConsumer<String, String> consumer =
new KafkaConsumer<>(props)) {
consumer.subscribe(Collections.singletonList("your_topic_name"));
while (true) {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("Received message: topic=%s, partition=%d, " +
"offset=%d, key=%s, value=%s%n",
record.topic(), record.partition(),
record.offset(), record.key(), record.value());
}
}
}
}
}示例代码中的连接地址、Topic名称和用户信息需要替换为从控制台获取的实际值。
6. Spring Boot集成：企业级开发的最佳实践
在Spring Boot项目中集成Kafka是更为常见的开发方式。华为云官方提供了spring-kafka的集成示例。
6.1 添加依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>需要根据JDK版本选择兼容的spring-kafka版本：JDK 17及以上版本对应spring-kafka 3.x.x，更早版本对应spring-kafka 2.x.x。
6.2 application.properties配置
# Kafka实例的Broker信息，ip:port表示连接地址和端口
spring.kafka.bootstrap-servers=ip1:port1,ip2:port2,ip3:port3
# 生产者配置
spring.kafka.producer.retries=10
spring.kafka.producer.batch-size=16384
spring.kafka.producer.buffer-memory=33554432
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# 消费者配置
spring.kafka.consumer.group-id=test-consumer-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=100
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# SASL认证配置（未开启SASL时删除以下配置）
# SASL机制：PLAIN或SCRAM-SHA-512
spring.kafka.properties.sasl.mechanism=PLAIN
spring.kafka.properties.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="your_username" password="your_password";
# 安全协议：SASL_PLAINTEXT或SASL_SSL
spring.kafka.security.protocol=SASL_PLAINTEXT
# 如果安全协议为SASL_SSL，配置SSL证书路径
# spring.kafka.ssl.trust-store-location=file:D:\\temp\\client.jks
# spring.kafka.ssl.trust-store-password=your_truststore_password上述配置中，SASL的用户名和密码需要从控制台获取。出于安全考虑，建议对用户名和密码进行加密处理。SASL机制为SCRAM-SHA-512时的配置略有不同：
spring.kafka.properties.sasl.mechanism=SCRAM-SHA-512
spring.kafka.properties.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="your_username" password="your_password";6.3 生产者与消费者代码
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class KafkaProducerService {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void sendMessage(String topic, String key, String message) {
kafkaTemplate.send(topic, key, message);
}
}
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@Service
public class KafkaConsumerService {
@KafkaListener(topics = "your_topic_name", groupId = "test-consumer-group")
public void consume(String message) {
System.out.println("Received message: " + message);
}
}相关完整代码可从华为云官方提供的kafka-springboot-demo获取。
7. 访问方式详解：内网、公网与跨VPC访问
华为云Kafka实例支持多种访问方式，以满足不同部署场景的需求。
内网访问：当客户端与Kafka实例处于同一VPC时，可以直接使用内网连接地址进行访问，内网访问的端口为9092。这种方式的网络延迟最低、安全性最高，且不产生公网流量费用，是生产环境的首选方案。
公网访问：如果需要从外部网络访问Kafka实例，需要开启公网访问并为实例配置弹性IP地址。公网访问的端口为9094。公网访问的开启和关闭只有在Kafka实例处于运行状态时才能操作。如果不再需要公网访问，可以随时关闭。
跨VPC访问：如果客户端与Kafka实例不在同一个VPC中，可以通过VPC对等连接、云连接或DNAT等方式实现跨VPC访问。
IPv6访问：在创建Kafka实例时如果开启了IPv6，可以为实例添加IPv6地址以支持IPv6私网和公网访问。
无论采用哪种访问方式，在连接前都需要确保安全组规则已正确配置，否则会导致连接失败。
8. Smart Connect：跨实例数据同步与消息转储
华为云Kafka提供了Smart Connect功能，可以实现Kafka实例与其他云服务之间或两个Kafka实例之间的数据同步。一个实例最多可以创建18个Smart Connect任务。
8.1 启用Smart Connect
在Kafka控制台的实例列表中，在待开启Smart Connect的Kafka实例所在行选择“更多 > 开启Smart Connect”，或者进入实例详情页后在右上角选择“更多 > 开启Smart Connect”。
8.2 主要应用场景
Kafka数据复制：在两个Kafka实例之间实现数据的单向或双向复制，适用于数据备份或迁移场景。Smart Connect任务会将源Topic的配置信息和消息数据同步到目标Topic。
消息转储至OBS：将Kafka实例中的数据实时转储到对象存储服务OBS中，实现消息数据的长期备份。转储文件的目录格式为：bucket-name/file-directory/topic-name/time，例如obs-kafka/smartconnect/topic01/2025/01/01。
Redis数据复制：通过Smart Connect任务将源Redis实例中的数据先复制到Kafka实例的Topic中，再通过Topic将数据复制到目标Redis实例。
9. 监控与告警：保障业务稳定运行
华为云Kafka版提供了完善的监控与告警能力，帮助用户实时掌握实例的运行状态。
9.1 查看监控指标
在Kafka控制台的实例详情页，选择“监控 > 监控详情”即可查看实例的各项监控指标。支持的监控指标包括消息生产速率、消费速率、消息堆积数、磁盘使用率等。在监控详情页面，每次最多可以选择50个资源进行监控数据查看。
9.2 配置告警规则
消息堆积是Kafka使用中最常见的运维问题之一。如果客户端的消费速度跟不上服务端的发送速度，未处理的消息会越来越多，造成消息消费延迟。针对消息堆积数配置告警规则，可以在堆积数超过阈值时通过短信或邮件及时收到通知。
配置告警规则的具体步骤为：在实例详情页的监控详情页面，选择需要创建告警的消费组，在“消息堆积数”图表上单击创建告警规则。在创建告警规则页面设置告警名称、监控范围、告警策略（如连续1次原始值大于等于10000个时触发重要告警，每天发送一次告警通知）。然后创建通知策略，选择“主题订阅”方式，在消息通知服务中创建主题和添加订阅。最后将通知策略关联到告警规则即可。
告警规则创建完成后，可以在云监控服务的“告警 > 告警规则”界面查看新创建的告警规则。
10. 常见问题排查与最佳实践
10.1 Topic不存在错误
在生产或消费消息时，如果遇到“Topic {{topic_name}} not present in metadata after 60000 ms”错误，可以采取以下措施解决：升级Kafka客户端版本到2.7或以上版本，并设置socket.connection.setup.timeout.ms大于1秒且小于request.timeout.ms除以Kafka服务端节点数。
10.2 消息时延过高
如果消息从生产到消费的端到端时延偶发达到6分钟，通常是因为某个消费者无法及时响应rebalance动作，导致整个消费组消费阻塞。建议检查消费者的处理逻辑，确保处理速度能够跟上消息生产速度，同时合理配置max.poll.records等参数。
10.3 扩容期间生产失败
在扩容或缩容代理规格的过程中，代理采用滚动重启的方式进行实例变更。在重启过程中，分区Leader会进行切换，此时生产客户端的元数据缓存中保存的分区Leader为旧分区Leader ID，仍然会向旧分区Leader发送消息，导致生产消息失败。建议在扩容操作期间适当增加生产端的重试次数和超时时间。
10.4 扩容失败资源不足
扩容时如果提示底层ECS或EVS资源不足，但在ECS控制台可以购买对应规格的ECS，通常是因为底层资源配额与控制台界面显示的可用规格配额存在差异。此时需要联系客服增加配额。
10.5 安全组配置建议
安全组是保障Kafka实例安全的第一道防线。建议遵循最小权限原则，只开放必要的端口给必要的IP地址段。对于内网访问，将客户端所在安全组加入Kafka实例安全组的入方向规则中。对于公网访问，建议使用SASL_SSL加密认证，并严格限制访问IP范围。
11. 总结
本文全面介绍了华为云分布式消息服务Kafka版的对接使用方法，从实例创建、资源规划、客户端集成到监控告警和问题排查，覆盖了完整的使用流程。华为云Kafka版在兼容开源Kafka的基础上，提供了全托管、高可用、弹性扩展的企业级能力，显著降低了消息队列的运维复杂度。无论是日志收集、流式数据处理还是应用解耦场景，华为云Kafka版都能提供稳定可靠的消息服务。
在实际生产环境中，建议根据业务量级合理选择实例规格，开启SASL_SSL保障数据安全，配置监控告警及时发现问题，并充分利用Smart Connect实现数据备份与迁移。希望本文能帮助开发者快速上手华为云Kafka，构建高可靠、高性能的消息驱动架构。
问答环节
问1：华为云Kafka实例创建后，SASL认证开关还能修改吗？
答：不能。SASL开关只能在创建实例时设置，实例创建成功后不支持修改。如果创建时未开启SASL，后续无法补开，需要重新创建实例。
问2：客户端连接华为云Kafka时，内网端口和公网端口分别是多少？
答：内网访问的连接端口为9092，公网访问的连接端口为9094。在配置客户端时需根据实际访问方式选择正确的端口。
问3：消息堆积数超过多少时应该配置告警？
答：建议根据业务对消息实时性的要求来设定阈值。一般建议设置为10000个消息作为告警阈值，即连续1次原始值大于等于10000个时触发告警。如果业务对延迟敏感，可以设置更低的阈值。
问4：Kafka实例的分区数上限是多少？
答：分区数上限与实例规格相关。以kafka.2u4g.cluster 3代理规格为例，分区数上限为750。Topic数量等于分区数上限除以每个Topic的分区数。
问5：使用Spring Boot集成华为云Kafka时，如何选择spring-kafka版本？
答：根据JDK版本选择：JDK 17及以上版本对应spring-kafka 3.x.x，更早版本对应spring-kafka 2.x.x。版本不匹配可能导致实例连接失败。
问6：Smart Connect任务最多能创建多少个？
答：一个Kafka实例最多可以创建18个Smart Connect任务。如果需要更多的同步任务，需要评估是否可以通过合并任务或提升实例规格来满足需求。
