Kafka消息队列实战指南

前言

Apache Kafka是一个分布式流处理平台，以其高吞吐量、低延迟、高可用性著称，被广泛应用于日志收集、消息队列、实时数据流处理等场景。本文将系统讲解Kafka的核心概念、使用方法和最佳实践。

一、Kafka核心概念

1.1 基础概念

1
2
3
4
5
6
7
Producer(生产者)    → 发送消息到Kafka
Broker(代理)        → Kafka服务器节点
Topic(主题)         → 消息分类
Partition(分区)     → Topic的物理分组
Consumer(消费者)    → 从Kafka读取消息
Consumer Group      → 消费者组，实现负载均衡
Offset(偏移量)      → 消息在分区中的位置

1.2 架构图

1
2
3
4
5
6
7
8
9
Producer1 ──┐
Producer2 ──┼──→ Topic-A (3 Partitions)
Producer3 ──┘      ├─ Partition-0 (Leader: Broker1, Replicas: Broker2,3)
                   ├─ Partition-1 (Leader: Broker2, Replicas: Broker1,3)
                   └─ Partition-2 (Leader: Broker3, Replicas: Broker1,2)
                         ↓
                   Consumer Group
                   ├─ Consumer1 (读取 P0, P1)
                   └─ Consumer2 (读取 P2)

二、Kafka安装部署

2.1 Docker Compose部署

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# docker-compose.yml
version: '3'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"
    volumes:
      - ./zookeeper-data:/var/lib/zookeeper/data
      - ./zookeeper-logs:/var/lib/zookeeper/log

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
    volumes:
      - ./kafka-data:/var/lib/kafka/data

# 启动
docker-compose up -d

2.2 基础操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 进入Kafka容器
docker exec -it <kafka_container> bash

# 创建Topic
kafka-topics --create \
  --bootstrap-server localhost:9092 \
  --replication-factor 1 \
  --partitions 3 \
  --topic test-topic

# 查看Topic列表
kafka-topics --list \
  --bootstrap-server localhost:9092

# 查看Topic详情
kafka-topics --describe \
  --bootstrap-server localhost:9092 \
  --topic test-topic

# 删除Topic
kafka-topics --delete \
  --bootstrap-server localhost:9092 \
  --topic test-topic

三、生产者(Producer)

3.1 Java生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class SimpleProducer {
    
    public static void main(String[] args) {
        // 配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
                  StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
                  StringSerializer.class.getName());
        
        // 可靠性配置
        props.put(ProducerConfig.ACKS_CONFIG, "all");  // 所有副本确认
        props.put(ProducerConfig.RETRIES_CONFIG, 3);   // 重试次数
        
        // 性能配置
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);  // 批次大小
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);      // 等待时间
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");  // 压缩
        
        // 创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        
        try {
            // 发送消息（异步）
            for (int i = 0; i < 100; i++) {
                String key = "key-" + i;
                String value = "message-" + i;
                
                ProducerRecord<String, String> record = 
                    new ProducerRecord<>("test-topic", key, value);
                
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception == null) {
                            System.out.printf("Sent to partition %d, offset %d\n",
                                metadata.partition(), metadata.offset());
                        } else {
                            exception.printStackTrace();
                        }
                    }
                });
            }
            
            // 同步发送
            ProducerRecord<String, String> record = 
                new ProducerRecord<>("test-topic", "sync-key", "sync-message");
            
            RecordMetadata metadata = producer.send(record).get();
            System.out.println("Sync send successful: " + metadata);
            
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}

3.2 分区策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
// 自定义分区器
public class CustomPartitioner implements Partitioner {
    
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                        Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        
        if (keyBytes == null) {
            // 轮询
            return new Random().nextInt(numPartitions);
        }
        
        // 根据key的hash
        return Math.abs(key.hashCode()) % numPartitions;
    }
    
    @Override
    public void close() {}
    
    @Override
    public void configure(Map<String, ?> configs) {}
}

// 使用自定义分区器
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, 
          CustomPartitioner.class.getName());

3.3 Spring Boot集成

1
2
3
4
5
<!-- pom.xml -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12
# application.yml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      retries: 3
      batch-size: 16384
      linger-ms: 10
      compression-type: snappy
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@Service
public class MessageProducer {
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    public void sendMessage(String topic, String message) {
        // 异步发送
        kafkaTemplate.send(topic, message)
            .addCallback(
                result -> {
                    RecordMetadata metadata = result.getRecordMetadata();
                    System.out.printf("Sent to %s-%d offset %d\n",
                        metadata.topic(), metadata.partition(), metadata.offset());
                },
                ex -> {
                    System.err.println("Send failed: " + ex.getMessage());
                }
            );
    }
    
    public void sendWithKey(String topic, String key, String message) {
        kafkaTemplate.send(topic, key, message);
    }
}

四、消费者(Consumer)

4.1 Java消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

public class SimpleConsumer {
    
    public static void main(String[] args) {
        // 配置
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
                  StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
                  StringDeserializer.class.getName());
        
        // 自动提交offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        
        // 从最早的消息开始消费
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        
        // 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        
        // 订阅Topic
        consumer.subscribe(Arrays.asList("test-topic"));
        
        try {
            while (true) {
                // 拉取消息
                ConsumerRecords<String, String> records = 
                    consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("topic=%s, partition=%d, offset=%d, key=%s, value=%s\n",
                        record.topic(), record.partition(), record.offset(),
                        record.key(), record.value());
                }
            }
        } finally {
            consumer.close();
        }
    }
}

4.2 手动提交Offset

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// 手动提交配置
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));

try {
    while (true) {
        ConsumerRecords<String, String> records = 
            consumer.poll(Duration.ofMillis(100));
        
        for (ConsumerRecord<String, String> record : records) {
            // 处理消息
            processRecord(record);
        }
        
        // 同步提交
        consumer.commitSync();
        
        // 或异步提交
        consumer.commitAsync((offsets, exception) -> {
            if (exception != null) {
                System.err.println("Commit failed: " + exception.getMessage());
            }
        });
    }
} finally {
    consumer.close();
}

4.3 Spring Boot消费者

1
2
3
4
5
6
7
8
9
10
11
# application.yml
spring:
  kafka:
    consumer:
      group-id: my-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: earliest
      enable-auto-commit: false
    listener:
      ack-mode: manual  # 手动提交
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@Component
public class MessageConsumer {
    
    @KafkaListener(topics = "test-topic", groupId = "my-group")
    public void listen(ConsumerRecord<String, String> record) {
        System.out.printf("Received: key=%s, value=%s, partition=%d, offset=%d\n",
            record.key(), record.value(), record.partition(), record.offset());
    }
    
    // 手动提交
    @KafkaListener(topics = "test-topic", groupId = "my-group")
    public void listenWithAck(ConsumerRecord<String, String> record,
                             Acknowledgment ack) {
        try {
            // 处理消息
            processMessage(record);
            
            // 手动提交
            ack.acknowledge();
        } catch (Exception e) {
            System.err.println("Process failed: " + e.getMessage());
        }
    }
    
    // 批量消费
    @KafkaListener(topics = "test-topic", groupId = "my-group")
    public void listenBatch(List<ConsumerRecord<String, String>> records) {
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("Batch received: " + record.value());
        }
    }
    
    private void processMessage(ConsumerRecord<String, String> record) {
        // 业务逻辑
    }
}

五、高级特性

5.1 事务消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// 生产者配置
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 初始化事务
producer.initTransactions();

try {
    // 开始事务
    producer.beginTransaction();
    
    // 发送消息
    producer.send(new ProducerRecord<>("topic1", "key1", "value1"));
    producer.send(new ProducerRecord<>("topic2", "key2", "value2"));
    
    // 提交事务
    producer.commitTransaction();
    
} catch (Exception e) {
    // 回滚事务
    producer.abortTransaction();
} finally {
    producer.close();
}

5.2 精确一次语义(Exactly Once)

1
2
3
4
5
6
// 生产者配置
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);  // 幂等性
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-tx-id");

// 消费者配置
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

5.3 消息过滤

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// 使用Consumer的filter
@KafkaListener(topics = "test-topic", groupId = "filter-group")
public void listen(ConsumerRecord<String, String> record) {
    // 过滤逻辑
    if (record.key().startsWith("important-")) {
        processImportantMessage(record.value());
    }
}

// 或使用RecordFilterStrategy
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> 
    kafkaListenerContainerFactory() {
    
    ConcurrentKafkaListenerContainerFactory<String, String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    
    factory.setConsumerFactory(consumerFactory());
    
    // 设置过滤策略
    factory.setRecordFilterStrategy(record -> {
        // 返回true则过滤该消息
        return !record.key().startsWith("important-");
    });
    
    return factory;
}

六、性能优化

6.1 生产者优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
Properties props = new Properties();

// 批处理
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);  // 增大批次
props.put(ProducerConfig.LINGER_MS_CONFIG, 20);      // 增加等待时间

// 压缩
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");  // 或snappy, gzip

// 缓冲区
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);  // 64MB

// 重试
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

// 超时
props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);

6.2 消费者优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
Properties props = new Properties();

// 批量拉取
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024 * 1024);  // 1MB
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);

// 会话超时
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);

// 分区分配策略
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
          "org.apache.kafka.clients.consumer.RangeAssignor");

6.3 Broker优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# server.properties

# 网络线程
num.network.threads=8
# IO线程
num.io.threads=16

# Socket缓冲区
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

# 日志保留
log.retention.hours=168
log.retention.bytes=1073741824
log.segment.bytes=1073741824

# 日志清理
log.cleanup.policy=delete

# 副本配置
default.replication.factor=3
min.insync.replicas=2

# 压缩
compression.type=producer

七、监控与运维

7.1 Kafka Manager

1
2
3
4
5
6
# Docker部署
docker run -d \
  --name kafka-manager \
  -p 9000:9000 \
  -e ZK_HOSTS="localhost:2181" \
  hlebalbau/kafka-manager:latest

7.2 监控指标

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# JMX监控
kafka-run-class kafka.tools.JmxTool \
  --object-name kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec \
  --reporting-interval 1000

# 查看消费者组offset
kafka-consumer-groups --bootstrap-server localhost:9092 \
  --group my-group \
  --describe

# 输出：
# TOPIC     PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# test-topic    0       100             150             50
# test-topic    1       200             200             0

7.3 常见问题排查

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# 查看Topic配置
kafka-configs --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name test-topic \
  --describe

# 修改Topic配置
kafka-configs --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name test-topic \
  --alter \
  --add-config retention.ms=86400000

# 查看分区分配
kafka-consumer-groups --bootstrap-server localhost:9092 \
  --group my-group \
  --members \
  --describe

# 重置offset
kafka-consumer-groups --bootstrap-server localhost:9092 \
  --group my-group \
  --topic test-topic \
  --reset-offsets \
  --to-earliest \
  --execute

八、实战案例

8.1 日志收集系统

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
// 日志生产者
@Service
public class LogProducer {
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    public void sendLog(String level, String message) {
        String logMessage = String.format("%s [%s] %s",
            LocalDateTime.now(), level, message);
        
        kafkaTemplate.send("app-logs", level, logMessage);
    }
}

// 日志消费者
@Component
public class LogConsumer {
    
    @KafkaListener(topics = "app-logs", groupId = "log-processor")
    public void processLog(ConsumerRecord<String, String> record) {
        String level = record.key();
        String message = record.value();
        
        if ("ERROR".equals(level)) {
            // 发送告警
            sendAlert(message);
        }
        
        // 存储到ES
        saveToElasticsearch(message);
    }
}

8.2 订单处理系统

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
// 订单服务
@Service
public class OrderService {
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    @Transactional
    public void createOrder(Order order) {
        // 保存订单到数据库
        orderRepository.save(order);
        
        // 发送消息
        String message = JSON.toJSONString(order);
        kafkaTemplate.send("order-created", order.getId(), message);
    }
}

// 库存服务
@Component
public class InventoryConsumer {
    
    @KafkaListener(topics = "order-created", groupId = "inventory-service")
    public void handleOrderCreated(String message, Acknowledgment ack) {
        try {
            Order order = JSON.parseObject(message, Order.class);
            
            // 扣减库存
            inventoryService.deduct(order.getItems());
            
            // 发送库存扣减成功消息
            kafkaTemplate.send("inventory-deducted", order.getId(), message);
            
            // 手动提交
            ack.acknowledge();
            
        } catch (Exception e) {
            // 库存不足，发送失败消息
            kafkaTemplate.send("order-failed", orderId, "库存不足");
        }
    }
}

九、最佳实践

核心要点

  1. 分区设计：合理规划分区数量
  2. 副本配置：至少3个副本保证高可用
  3. 消息幂等：处理重复消息
  4. 顺序保证：相同key发送到同一分区
  5. 监控告警：关注lag、吞吐量等指标

注意事项

  • ✅ 避免单条消息过大（建议<1MB）
  • ✅ 合理设置retention时间
  • ✅ 消费者处理失败要有重试机制
  • ✅ 定期备份重要数据
  • ✅ 监控磁盘空间

十、总结

Kafka作为分布式消息队列的佼佼者：

核心优势

  • 高吞吐：百万级TPS
  • 低延迟：毫秒级
  • 高可用：副本机制
  • 持久化：消息持久化到磁盘

适用场景

  • 日志收集
  • 消息队列
  • 流式处理
  • 事件溯源

掌握Kafka，构建高性能的消息系统！

参考资料