通义灵码怎么写Kafka消息队列生产者消费者实现-人工智能-PHP中文网

必须先启动kafka服务、创建topic并确保网络连通，否则代码抛timeoutexception或unknowntopicorpartitionexception；再添加kafka-clients 3.7.0依赖，编写同步/异步生产者与带订阅、拉取、关闭逻辑的消费者。

☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜

通义灵码怎么写kafka消息队列生产者消费者实现

要在Java项目中快速实现Kafka消息队列的生产者与消费者功能，必须先确保Kafka服务已启动、topic已创建，且客户端能连通broker；否则代码会抛出TimeoutException或UnknownTopicOrPartitionException。

准备Kafka运行环境

打开终端，进入Kafka安装目录的bin/子目录，执行以下命令启动ZooKeeper（若使用Kafka内置ZK）和Kafka Server：

./zookeeper-server-start.sh config/zookeeper.properties → ./kafka-server-start.sh config/server.properties

新开终端，创建名为test-topic的主题，分区数为1、副本数为1：

./kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test-topic

【必须确认该topic已成功创建，否则后续生产/消费将失败】 验证方式：运行./kafka-topics.sh --list --bootstrap-server localhost:9092，输出中应包含test-topic。

添加Maven依赖

在pom.xml中加入Kafka客户端核心依赖（适配当前主流Kafka 3.x+版本）：

  org.apache.kafka
  kafka-clients
  3.7.0

这一步操作起来很简单，直接把依赖块粘贴进dependencies标签内即可。不需要额外引入ZooKeeper或Spring Kafka——纯客户端场景下，kafka-clients足矣。

编写Kafka生产者

方法一：基础同步发送

新建类SimpleProducer.java，填入以下代码：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer producer = new KafkaProducer(props);
ProducerRecord record = new ProducerRecord("test-topic", "hello-key", "hello-value");
producer.send(record).get();
producer.close();

通义灵码 Linux版

通义灵码 Linux版是阿里云推出的一款AI智能编码助手，专为Linux开发者设计。它支持在Linux操作系统下的JetBrains IDEs、Visual Studio Code等主流集成开发环境中运行。该工具基于通义大模型，提供代码智能生成、实时续写、单元测试生成、代码优化以及研发智能问答等功能，旨在帮助Linux用户在编码过程中提升效率。

下载

注意：调用.get()会阻塞直到Broker返回响应，适合调试；生产环境建议改用回调方式避免线程卡死。

方法二：带回调的异步发送

替换producer.send(record).get()为：

producer.send(record, (metadata, exception) -> {
  if (exception != null) {
    exception.printStackTrace();
  } else {
    System.out.printf("Sent to %s-%d @ offset %d%n", metadata.topic(), metadata.partition(), metadata.offset());
  }
});

编写Kafka消费者

第一步：配置消费者并订阅主题

新建类SimpleConsumer.java，初始化配置：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "demo-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
KafkaConsumer consumer = new KafkaConsumer(props);
consumer.subscribe(Collections.singletonList("test-topic"));

第二步：持续拉取消息并打印

while (true) {
  ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
  for (ConsumerRecord record : records) {
    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
  }
}

第三步：关闭资源

在循环外添加consumer.close();——但实际运行时需配合信号捕获（如Ctrl+C）来触发关闭，否则进程不会退出。