这在大数据和后端开发领域是一个非常经典的问题。Kafka 是目前最流行的分布式消息队列/流处理平台之一。
针对 Linux 环境下的 Kafka 入门，我为你整理了一份从零到一的实战指南。Kafka 是用 Scala/Java 写的，所以无论你用什么语言开发，服务端部署在 Linux 上都是最标准的做法。
Kafka 依赖 Java 运行环境。
安装 JDK (推荐 JDK 8 或 11) 检查是否已安装：
java -version
如果没有，请安装（以 Ubuntu 为例）：
sudo apt update
sudo apt install openjdk-11-jdk
下载 Kafka 去 Apache Kafka 官网 复制最新版本的链接（选择 Binary 下载）。
# 进入常用安装目录
cd /opt
# 下载 (请替换为最新版本链接)
sudo wget https://downloads.apache.org/kafka/3.6.1/kafka_2.13-3.6.1.tgz
# 解压
sudo tar -xzf kafka_2.13-3.6.1.tgz
# 重命名方便操作
sudo mv kafka_2.13-3.6.1 kafka
cd kafka
Kafka 的运行依赖于 ZooKeeper（旧版本）或 KRaft（新版本）。为了让你快速入门，这里先使用传统的 ZooKeeper 模式（最通用，资料最多）。
启动 ZooKeeper Kafka 自带了一个 ZooKeeper 实例，方便测试。
# 启动 ZooKeeper (后台运行)
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
启动 Kafka Server (Broker)
# 启动 Kafka (后台运行)
bin/kafka-server-start.sh -daemon config/server.properties
验证服务是否启动
# 查看进程
jps
如果看到 QuorumPeerMain (ZooKeeper) 和 Kafka 两个进程，说明启动成功。
这是理解 Kafka 概念（Topic, Producer, Consumer）最快的方式。
在 Kafka 中，消息是按 Topic 分类的。
bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
--partitions 1: 设置 1 个分区（入门简单）。--replication-factor 1: 设置 1 个副本（因为是单机，只能设为1）。bin/kafka-topics.sh --list --bootstrap-server localhost:9092
启动一个控制台生产者，输入消息。
bin/kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092
输入几行文字，按回车发送，然后
Ctrl+C退出。
启动一个控制台消费者，查看刚才发送的消息。
# 从头开始消费
bin/kafka-console-consumer.sh --topic test-topic --from-beginning --bootstrap-server localhost:9092
你会看到刚才发送的消息。保持这个窗口开着，如果在生产者窗口发新消息，这里会实时显示。
光会敲命令不够，必须理解这几个概念：
虽然你在 Linux 上部署，但开发通常是用 Java/Python/Go。这里以 Java 为例（Maven 依赖）：
1. 添加依赖 (pom.xml):
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.6.1</version>
</dependency>
2. 简单的生产者代码:
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class SimpleProducer {
public static void main(String[] args) {
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");
Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("test-topic", "key1", "Hello Kafka from Java!"));
producer.close();
}
}
3. 简单的消费者代码:
import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;
public class SimpleConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group"); // 必须指定组
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset=%d, key=%s, value=%s%n", record.offset(), record.key(), record.value());
}
}
}
}
停止服务 先停 Kafka，再停 ZooKeeper。
bin/kafka-server-stop.sh
bin/zookeeper-server-stop.sh
配置文件位置
config/server.properties: Kafka 主配置（修改端口、日志路径、Broker ID 等）。config/zookeeper.properties: ZooKeeper 配置。常见错误
java.net.ConnectException: Connection refused: 说明 Kafka 没启动，或者 localhost:9092 写错了。Cannot allocate memory: Kafka 默认启动内存较大。编辑 bin/kafka-server-start.sh，找到 KAFKA_HEAP_OPTS，把 -Xmx1G -Xms1G 改成 -Xmx256M -Xms128M。推荐资源：
祝你入门顺利！如果有具体的报错，可以把日志贴出来。
