花于陌上开个人站

第三章 生产者与消费者 v1.0 ​

第一部分 生产者 ​

一、发送模式 ​

1.同步发送 ​

同步发送会等待Broker返回发送结果，可靠性高。

java
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void sendSyncMessage() throws Exception {
    SendResult<String, String> result = kafkaTemplate.send("topic-name", "key", "value").get();
    System.out.println("发送成功: " + result.getRecordMetadata().offset());
}

2.异步发送 ​

异步发送通过回调处理结果，性能高。

java
public void sendAsyncMessage() {
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("topic-name", "key", "value");
    
    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override
        public void onSuccess(SendResult<String, String> result) {
            System.out.println("发送成功: " + result.getRecordMetadata().offset());
        }
        
        @Override
        public void onFailure(Throwable ex) {
            System.out.println("发送失败: " + ex.getMessage());
        }
    });
}

3.批量发送 ​

批量发送可以提高吞吐量。

java
public void sendBatchMessage() {
    List<Message<String>> messages = new ArrayList<>();
    messages.add(MessageBuilder.withPayload("message1").setHeader(KafkaHeaders.TOPIC, "topic").build());
    messages.add(MessageBuilder.withPayload("message2").setHeader(KafkaHeaders.TOPIC, "topic").build());
    
    kafkaTemplate.send(messages);
}

二、分区器 ​

1.默认分区策略 ​

  • 有Key：按Key哈希分配到固定分区
  • 无Key：轮询分配到所有分区

2.自定义分区器 ​

java
public class CustomPartitioner implements Partitioner {
    
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, 
                        Object value, byte[] valueBytes, Cluster cluster) {
        // 自定义分区逻辑
        int partitionNum = cluster.partitionCountForTopic(topic);
        return Math.abs(key.hashCode()) % partitionNum;
    }
    
    @Override
    public void close() {}
    
    @Override
    public void configure(Map<String, ?> configs) {}
}

配置使用自定义分区器：

yaml
spring:
  kafka:
    producer:
      partitioner-class-name: com.example.CustomPartitioner

三、消息确认机制 ​

1.acks配置 ​

  • acks=0：不等待确认，性能最高，可能丢消息
  • acks=1：等待Leader确认（默认），可能丢少量消息
  • acks=all：等待所有ISR确认，可靠性最高
yaml
spring:
  kafka:
    producer:
      acks: all

2.重试机制 ​

yaml
spring:
  kafka:
    producer:
      retries: 3
      retry-backoff-ms: 1000

第二部分 消费者 ​

一、消费模式 ​

1.自动提交 ​

yaml
spring:
  kafka:
    consumer:
      enable-auto-commit: true
      auto-commit-interval: 1000

2.手动提交 ​

java
@Component
public class ManualCommitConsumer {
    
    @KafkaListener(topics = "topic-name", groupId = "my-group")
    public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            // 处理消息
            System.out.println("收到消息: " + record.value());
            
            // 手动提交
            ack.acknowledge();
        } catch (Exception e) {
            // 处理失败，不提交
            throw new RuntimeException("消费失败", e);
        }
    }
}

二、消费配置 ​

yaml
spring:
  kafka:
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      enable-auto-commit: false
      max-poll-records: 500
      fetch-min-size: 1
      fetch-max-wait: 500
配置项说明建议值
auto-offset-reset无offset时的重置策略earliest
max-poll-records单次拉取最大记录数500
fetch-min-size最小拉取大小1
fetch-max-wait最大等待时间500

三、消费重试 ​

1.注解方式 ​

java
@Component
public class RetryConsumer {
    
    @RetryableTopic(
        attempts = 3,
        backoff = @Backoff(delay = 1000, multiplier = 2)
    )
    @KafkaListener(topics = "topic-name", groupId = "my-group")
    public void listen(String message) {
        // 业务处理
        processMessage(message);
    }
    
    @DltHandler
    public void handleDlt(String message) {
        // 处理死信队列消息
        System.out.println("进入死信队列: " + message);
    }
}

2.配置方式 ​

yaml
spring:
  kafka:
    consumer:
      enable-auto-commit: false
    listener:
      type: batch
      ack-mode: manual
    retry:
      enabled: true
      max-attempts: 3
      initial-interval: 1000
      max-interval: 10000
      multiplier: 2

四、消息过滤 ​

java
@Component
public class FilterConsumer {
    
    @KafkaListener(topics = "topic-name", groupId = "my-group")
    public void listen(String message) {
        // 只处理符合条件的消息
        if (shouldProcess(message)) {
            processMessage(message);
        }
    }
    
    private boolean shouldProcess(String message) {
        // 自定义过滤逻辑
        return message.contains("important");
    }
}

第三部分 常见问题 ​

一、消息丢失 ​

1.生产者端 ​

  • 配置acks=all
  • 开启重试机制
  • 使用同步发送

2.Broker端 ​

  • 配置多个副本
  • 设置min.insync.replicas
  • 使用同步刷盘

3.消费者端 ​

  • 关闭自动提交
  • 处理完成后手动提交
  • 实现幂等消费

二、消息重复 ​

1.产生原因 ​

  • 生产者重试
  • 消费者重平衡
  • 网络延迟

2.解决方案 ​

  • 幂等性设计
  • 唯一ID去重
  • 数据库唯一约束

三、消息积压 ​

1.监控指标 ​

  • 消费者Lag
  • 消费延迟

2.解决方案 ​

  • 增加消费者数量
  • 增加分区数
  • 优化消费逻辑

四、分区再均衡 ​

1.影响 ​

  • 暂停消费
  • 可能重复消费

2.优化 ​

  • 增加会话超时
  • 使用增量再平衡
  • 避免频繁启停消费者