全部文档
当前文档
最近更新时间：2025-08-18 17:22:14
Kafka API 官方文档链接：Overview（Kafka 2.8.1 API）
该任务以 Java 客户端为例指导您使用 VPC 网络接入消息队列 Kafka 并收发消息。
前提条件
操作步骤
步骤1：准备配置
将下载的 Demo 中的 javakafkademo 上传至 Linux 服务器。
登录 Linux 服务器，进入 javakafkademo 目录，并配置相关参数。
i. 在 pom.xml 中添加以下依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.1</version>
</dependency>ii. 创建Kafka 配置文件 Kafka.properties。
## 配置接入网络，在控制台的集群详情页面主机列表页复制。
bootstrap.servers=xx.xx.xx.xx:xxxx
## 配置Topic，在控制台上kafka eagle 管理页面复制。
topic=XXX
## 配置Consumer Group，您可以自定义设置
group.id=XXX参数 | 说明 |
bootstrap.servers | 接入网络，在控制台的集群详情页面主机列表页复制。 |
topic | Topic 名称，您可以在控制台上 kafka eagle webui页面复制。 |
group.id | 您可以自定义设置，demo 运行成功后可以在 Consumer Group 页面看到该消费者。 |
创建配置文件加载程序 KafkaConfigurer.java。
public class KafkaConfigurer {
private static Properties properties;
public synchronized static Properties getKafkaProperties() {
if (null != properties) {
return properties;
}
//获取配置文件kafka.properties的内容。
Properties kafkaProperties = new Properties();
try { kafkaProperties.load(KafkaProducerDemo.class.getClassLoader().getResourceAsStream("kafka.properties"));
} catch (Exception e) {
System.out.println("getKafkaProperties error");
}
properties = kafkaProperties;
return kafkaProperties;
}
}
步骤2：发送消息
编写生产消息程序KafkaProducerDemo.java。
public class KafkaProducerDemo {
public static void main(String args[]) {
//加载kafka.properties。
Properties kafkaProperties = KafkaConfigurer.getKafkaProperties();
Properties properties = new Properties();
//设置接入点，请通过控制台获取对应Topic的接入点。
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
//消息队列Kafka版消息的序列化方式, 此处demo 使用的是StringSerializer。
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");
//请求的最长等待时间。
properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
//设置客户端内部重试次数。
properties.put(ProducerConfig.RETRIES_CONFIG, 5);
//设置客户端内部重试间隔。
properties.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
//构造Producer对象。
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
//构造一个消息队列Kafka版消息。
String topic = kafkaProperties.getProperty("topic"); //消息所属的Topic，请在控制台申请之后，填写在这里。
String value = "this is kafka msg value"; //消息的内容。
try {
//批量获取Future对象可以加快速度, 但注意, 批量不要太大。
List<Future<RecordMetadata>> futureList = new ArrayList<>(128);
for (int i = 0; i < 10; i++) {
//发送消息，并获得一个Future对象。
ProducerRecord<String, String> kafkaMsg = new ProducerRecord<>(topic,
value + ": " + i);
Future<RecordMetadata> metadataFuture = producer.send(kafkaMsg);
futureList.add(metadataFuture);
}
producer.flush();
for (Future<RecordMetadata> future : futureList) {
//同步获得Future对象的结果。
RecordMetadata recordMetadata = future.get();
System.out.println("produce send ok: " + recordMetadata.toString());
}
} catch (Exception e) {
//客户端内部重试之后，仍然发送失败，业务要应对此类错误。
System.out.println("error occurred");
}
}
}编译并运行 KafkaProducerDemo.java 发送消息。
在Kafka控制台的 kafka eagle webui页面，选择对应的 Topic ，查看刚刚发送的消息。
步骤3：消费消息
创建 Consumer 订阅消息程序 KafkaConsumerDemo.java。
public class KafkaConsumerDemo {
public static void main(String args[]) {
//加载kafka.properties。
Properties kafkaProperties = KafkaConfigurer.getKafkaProperties();
Properties props = new Properties();
//设置接入点，请通过控制台获取对应Topic的接入点。
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
//两次Poll之间的最大允许间隔。
//消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Consumer Group移除并触发Rebalance，默认30s。
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
//每次Poll的最大数量。
//注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
//消息的反序列化方式。
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
//属于同一个组的消费集群，会负载消费消息。
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
//构造消费对象，也即生成一个消费集群。
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
//设置消费组订阅的Topic，可以订阅多个。
//如果GROUP_ID_CONFIG是一样，则订阅的Topic也建议设置成一样。
List<String> subscribedTopics = new ArrayList<>();
//如果需要订阅多个Topic，则在这里添加进去即可。
//每个Topic需要先在控制台进行创建。
String topicStr = kafkaProperties.getProperty("topic");
String[] topics = topicStr.split(",");
for (String topic : topics) {
subscribedTopics.add(topic.trim());
}
consumer.subscribe(subscribedTopics);
//循环消费消息。
while (true) {
try {
ConsumerRecords<String, String> records = consumer.poll(1000);
//必须在下次Poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG。
//建议开一个单独的线程池来消费消息，然后异步返回结果。
for (ConsumerRecord<String, String> record : records) {
System.out.println(
String.format("Consume partition:%d offset:%d", record.partition(), record.offset()));
}
} catch (Exception e) {
System.out.println("consumer error!");
}
}
}
}编译并运行 KafkaConsumerDemo.java 消费消息。
运行结果
Consume partition:0 offset:298
Consume partition:0 offset:299
该任务以 Java 客户端为例指导您在公网网络环境下，使用 SASL_PLAINTEXT 方式接入消息队列 Kafka 并收发消息。
前提条件
操作步骤
步骤1：控制台配置
创建接入点。
i.在 集群列表 页面，单击目标集群 ID，进入集群详情页。
ii.在kafka 服务管理页面开启ACL并滚动重启Broker。
在用户管理页面新建角色，设置密码。
在控制台 kafka eagle webui 页面新建 Topic。
步骤2：添加配置文件
在 pom.xml 中添加以下依赖
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.1</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.5</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>1.6.4</version>
</dependency>
</dependencies>说明: username 是配置的用户名，password 是配置的用户密码。
创建Kafka 配置文件 kafka.properties。
## 配置接入网络，在控制台的集群详情页面主机列表页复制。
bootstrap.servers=kafka-xxxxxxx
## 配置 Topic，在控制台上 kafka eagle页面复制。
topic=XXX
## 配置 consumer group，您可以自定义设置
group.id=XXX
## SASL 配置
java.security.auth.login.config.plain=/xxxx/kafka_client_jaas.conf参数 | 说明 |
bootstrap.servers | 接入网络，在控制台的集群详情页面主机列表页复制。 |
topic | Topic 名称，您可以在控制台上 kafka eagle webui 管理页面复制。 |
group.id | 您可以自定义设置，Demo 运行成功后可以在 Consumer Group 页面看到该消费者。 |
java.security.auth.login.config.plain | 填写 JAAS 配置文件 kafka_client_jaas.conf 的路径。 |
创建配置文件加载程序 KafkaConfigurer.java。
public class KafkaConfigurer {
private static Properties properties;
public static void configureSaslPlain() {
//如果用 -D 或者其它方式设置过，这里不再设置。
if (null == System.getProperty("java.security.auth.login.config")) {
//请注意将 XXX 修改为自己的路径。
System.setProperty("java.security.auth.login.config",
getKafkaProperties().getProperty("java.security.auth.login.config.plain"));
}
}
public synchronized static Properties getKafkaProperties() {
if (null != properties) {
return properties;
}
//获取配置文件 kafka.properties 的内容。
Properties kafkaProperties = new Properties();
try {
kafkaProperties.load(KafkaProducerDemo.class.getClassLoader().getResourceAsStream("kafka.properties"));
} catch (Exception e) {
System.out.println("getKafkaProperties error");
}
properties = kafkaProperties;
return kafkaProperties;
}
}步骤3：发送消息
创建发送消息程序 KafkaSaslProducerDemo.java。
public class KafkaSaslProducerDemo {
public static void main(String[] args) {
//设置 JAAS 配置文件的路径。
KafkaConfigurer.configureSaslPlain();
//加载 kafka.properties。
Properties kafkaProperties = KafkaConfigurer.getKafkaProperties();
Properties props = new Properties();
//设置接入点，请通过控制台获取对应 Topic 的接入点。
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaProperties.getProperty("bootstrap.servers"));
//
// SASL_PLAINTEXT 公网接入
//
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
// SASL 采用 Plain 方式。
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
//消息队列 Kafka 版消息的序列化方式。
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
//请求的最长等待时间。
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
//设置客户端内部重试次数。
props.put(ProducerConfig.RETRIES_CONFIG, 5);
//设置客户端内部重试间隔。
props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
//ack=0 producer 将不会等待来自 broker 的确认，重试配置不会生效。注意如果被限流了后，就会被关闭连接。
//ack=1 broker leader 将不会等待所有 broker follower 的确认，就返回 ack。
//ack=all broker leader 将等待所有 broker follower 的确认，才返回 ack。
props.put(ProducerConfig.ACKS_CONFIG, "all");
//构造 Producer 对象，注意，该对象是线程安全的，一般来说，一个进程内一个Producer对象即可。
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
//构造一个消息队列 Kafka 版消息。
String topic = kafkaProperties.getProperty("topic"); //消息所属的Topic，请在控制台申请之后，填写在这里。
String value = "this is kafka msg value"; //消息的内容。
try {
//批量获取 Future 对象可以加快速度。但注意，批量不要太大。
List<Future<RecordMetadata>> futures = new ArrayList<>(128);
for (int i = 0; i < 100; i++) {
//发送消息，并获得一个Future对象。
ProducerRecord<String, String> kafkaMessage = new ProducerRecord<>(topic,
value + ": " + i);
Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
futures.add(metadataFuture);
}
producer.flush();
for (Future<RecordMetadata> future : futures) {
//同步获得 Future 对象的结果。
RecordMetadata recordMetadata = future.get();
System.out.println("Produce ok:" + recordMetadata.toString());
}
} catch (Exception e) {
//客户端内部重试之后，仍然发送失败，业务要应对此类错误。
System.out.println("error occurred");
}
}
}编译并运行 KafkaSaslProducerDemo.java 发送消息。
在 Kafka 控制台 kafka eagle webui 管理页面，选择对应的 Topic，查看刚刚发送的消息。
步骤4：消费消息
创建 Consumer 订阅消息程序 KafkaSaslConsumerDemo.java。
public class KafkaSaslConsumerDemo {
public static void main(String[] args) {
//设置JAAS配置文件的路径。
KafkaConfigurer.configureSaslPlain();
//加载kafka.properties。
Properties kafkaProperties = KafkaConfigurer.getKafkaProperties();
Properties props = new Properties();
//设置接入点，请通过控制台获取对应Topic的接入点。
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaProperties.getProperty("bootstrap.servers"));
//
// SASL_PLAINTEXT 公网接入
//
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
// SASL 采用 Plain 方式。
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
//消费者超时时长
//消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Consumer Group移除并触发Rebalance，默认30s
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
//两次poll的最长时间间隔
//0.10.1.0 版本前这2个概念是混合的，都用session.timeout.ms表示
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30000);
//每次 Poll 的最大数量。
//注意该值不要改得太大，如果 Poll 太多数据，而不能在下次 Poll 之前消费完，则会触发一次负载均衡，产生卡顿。
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
//消息的反序列化方式。
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
//当前消费集群所属的消费组，请在控制台申请之后填写。
//属于同一个组的消费集群，会负载消费消息。
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
//构造消费对象，也即生成一个消费集群。
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
//设置消费组订阅的 Topic，可以订阅多个。
//如果 GROUP_ID_CONFIG 是一样，则订阅的 Topic 也建议设置成一样。
List<String> subscribedTopics = new ArrayList<String>();
//如果需要订阅多个 Topic，则在这里添加进去即可。
//每个 Topic 需要先在控制台进行创建。
String topicStr = kafkaProperties.getProperty("topic");
String[] topics = topicStr.split(",");
for (String topic : topics) {
subscribedTopics.add(topic.trim());
}
consumer.subscribe(subscribedTopics);
//循环消费消息。
while (true) {
try {
ConsumerRecords<String, String> records = consumer.poll(1000);
//必须在下次 Poll 之前消费完这些数据, 且总耗时不得超过 SESSION_TIMEOUT_MS_CONFIG。
for (ConsumerRecord<String, String> record : records) {
System.out.println(
String.format("Consume partition:%d offset:%d", record.partition(),
record.offset()));
}
} catch (Exception e) {
System.out.println("consumer error!");
}
}
}
}编译并运行 KafkaSaslConsumerDemo.java 消费消息。
运行结果。
Consume partition:0 offset:298
Consume partition:0 offset:299
在 Kafka 控制台 Consumer Group 页面，选择对应的消费组名称，在主题名称输入 Topic 名称，单击查询详情，查看消费详情。
纯净模式