Kafka调研笔记

安装部署

安装服务端

安装环境：笔记本电脑+Ubuntu系统

1. 需要先安装JAVA

# 安装 OpenJDK
sudo apt update
sudo apt install -y openjdk-11-jdk# 验证安装
java -version

2. 安装 Kafka

# 添加 Apache 仓库
wget -qO - https://packages.confluent.io/deb/7.6/archive.key | sudo apt-key add -
sudo add-apt-repository "deb [arch=amd64] https://packages.confluent.io/deb/7.6 stable main"
sudo add-apt-repository "deb https://packages.confluent.io/clients/deb $(lsb_release -cs) main"# 安装 Confluent Platform（包含 Kafka）
sudo apt update
sudo apt install -y confluent-community-2.13# 或只安装 Kafka
sudo apt install -y confluent-kafka-2.13

启动服务

# 启动Zookeeper（Kafka的依赖）
sudo systemctl start confluent-zookeeper# 启动Kafka
sudo systemctl start confluent-kafka# 设置开机自启
sudo systemctl enable confluent-zookeeper
sudo systemctl enable confluent-kafka

测试

测试1：创建主题

# 使用 kafka-topics 命令
kafka-topics --create \--topic test-topic \--bootstrap-server localhost:9092 \--partitions 1 \--replication-factor 1# 或如果命令不存在，用完整路径
/usr/bin/kafka-topics --create \--topic test-topic \--bootstrap-server localhost:9092 \--partitions 1 \--replication-factor 1

测试2：列出主题

kafka-topics --list --bootstrap-server localhost:9092

测试3：生产和消费消息

# 终端1：启动生产者
kafka-console-producer --topic test-topic --bootstrap-server localhost:9092
# 输入一些消息，按Ctrl+C退出# 终端2：启动消费者
kafka-console-consumer --topic test-topic \--from-beginning \--bootstrap-server localhost:9092

示例代码

安装依赖包

Install-Package Confluent.Kafka

生产者

// 1. 配置 Producer
using Confluent.Kafka;var config = new ProducerConfig
{// Kafka 服务器地址，多个用逗号分隔BootstrapServers = "192.168.126.194:9092",// 设置acks为all，确保数据可靠性Acks = Acks.All,// 如果没有指定Key，这里可以为空，或者设置分区器Partitioner = Confluent.Kafka.Partitioner.Random
};// 2. 创建 Producer 实例 (Key为Null, Value为string)
// 如果需要Key，可将 Null 改为 string (如 Producer<string, string>)
using (var producer = new ProducerBuilder<Null, string>(config).Build())
{string topic = "my-topic";string message = "Hello Confluent Kafka from C#!";try{while (true){// 3. 异步发送消息var deliveryResult = producer.ProduceAsync(topic, new Message<Null, string>{Value = message}).GetAwaiter().GetResult();Console.WriteLine($"成功发送消息到: {deliveryResult.Topic} (分区: {deliveryResult.Partition}, 偏移量: {deliveryResult.Offset})");Thread.Sleep(1000);}}catch (ProduceException<Null, string> e){Console.WriteLine($"发送失败: {e.Error.Reason}");}
}

消费者

// 1. 配置 Consumer
using Confluent.Kafka;var config = new ConsumerConfig
{BootstrapServers = "192.168.126.194:9092",// 消费者组ID，非常重要！同组消费者会负载均衡消费消息GroupId = "test-consumer-group-1",// 自动提交偏移量的设置EnableAutoCommit = false, // 建议手动提交以防止数据丢失// 如果没有初始偏移量，从最早开始读AutoOffsetReset = AutoOffsetReset.Earliest
};// 2. 创建 Consumer 实例
using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
{string topic = "my-topic";// 3. 订阅主题 (支持正则表达式，如 "^my-.*")consumer.Subscribe(topic);CancellationTokenSource cts = new CancellationTokenSource();Console.CancelKeyPress += (_, e) => {e.Cancel = true; // 防止进程终止cts.Cancel();};try{Console.WriteLine("消费者启动，正在监听消息...");while (true){// 4. 拉取消息 (Consume 方法是阻塞的，直到有消息或超时)try{var consumeResult = consumer.Consume(cts.Token);Console.WriteLine($"收到消息: {consumeResult.Message.Value} [Topic: {consumeResult.Topic}, Partition: {consumeResult.Partition}, Offset: {consumeResult.Offset}]");// 5. 手动提交偏移量 (表示消息已处理完毕)// 这里可以基于业务逻辑处理，只有确认处理成功才Commitconsumer.Commit(consumeResult);}catch (ConsumeException e){Console.WriteLine($"消费出错: {e.Error.Reason}");}}}catch (OperationCanceledException){// Ctrl+C 被触发，优雅关闭consumer.Close();}
}