# 啟動(dòng) Zookeeper（Kafka 3.0+ 已內(nèi)置）
bin/zookeeper-server-start.sh config/zookeeper.properties
# 啟動(dòng) Kafka 服務(wù)
bin/kafka-server-start.sh config/server.properties

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>3.1.0</version>
</dependency>

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: com.example.config.AttendanceSerializer # 自定義序列化器
      batch-size: 16384 # 批量發(fā)送優(yōu)化
      linger-ms: 100 # 批量發(fā)送延遲
      retries: 3
      acks: all
    consumer:
      group-id: attendance-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: com.example.config.AttendanceDeserializer # 自定義反序列化器
      auto-offset-reset: earliest
      enable-auto-commit: true
      max-poll-records: 500
      isolation.level: read_committed
      # 事務(wù)配置（可選）
      transaction-id-prefix: attendance-trans

package jnpf.model.attendance.event;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
import java.io.Serializable;
import java.util.Date;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class AttendanceStatisticsSingleDto implements Serializable {
    private static final long serialVersionUID = 1L;
    @NotBlank(message = "租戶Id不能為空")
    private String tenantId;
    @NotBlank(message = "考勤組Id不能為空")
    private String groupId;
    @NotBlank(message = "用戶Id不能為空")
    private String userId;
    @NotNull(message = "日期不能為空")
    private Date day;
}

package com.example.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import jnpf.model.attendance.event.AttendanceStatisticsSingleDto;
import org.apache.kafka.common.serialization.Serializer;
import java.util.Map;
public class AttendanceSerializer implements Serializer<AttendanceStatisticsSingleDto> {
    private final ObjectMapper objectMapper = new ObjectMapper();
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 配置方法，無需額外操作
    }
    @Override
    public byte[] serialize(String topic, AttendanceStatisticsSingleDto data) {
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new RuntimeException("序列化失敗: " + data, e);
        }
    }
    @Override
    public void close() {
        // 無需關(guān)閉資源
    }
}

package com.example.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import jnpf.model.attendance.event.AttendanceStatisticsSingleDto;
import org.apache.kafka.common.serialization.Deserializer;
import java.util.Map;
public class AttendanceDeserializer implements Deserializer<AttendanceStatisticsSingleDto> {
    private final ObjectMapper objectMapper = new ObjectMapper();
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 配置方法，無需額外操作
    }
    @Override
    public AttendanceStatisticsSingleDto deserialize(String topic, byte[] data) {
        try {
            return objectMapper.readValue(data, AttendanceStatisticsSingleDto.class);
        } catch (Exception e) {
            throw new RuntimeException("反序列化失敗: " + topic, e);
        }
    }
    @Override
    public void close() {
        // 無需關(guān)閉資源
    }
}

package com.example.service;
import com.example.config.AttendanceSerializer;
import jnpf.model.attendance.event.AttendanceStatisticsSingleDto;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class AttendanceProducer {
    private final KafkaTemplate<String, AttendanceStatisticsSingleDto> kafkaTemplate;
    public AttendanceProducer(KafkaTemplate<String, AttendanceStatisticsSingleDto> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    public void sendAttendanceData(AttendanceStatisticsSingleDto attendance) {
        // 發(fā)送消息到主題 attendance-topic
        kafkaTemplate.send("attendance-topic", attendance.getUserId(), attendance);
        System.out.println("? 消息發(fā)送成功: " + attendance.getUserId() + " | " + attendance.getDay());
    }
}

package com.example.consumer;
import com.example.config.AttendanceDeserializer;
import jnpf.model.attendance.event.AttendanceStatisticsSingleDto;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
@Service
public class AttendanceConsumer {
    @KafkaListener(
        topics = "attendance-topic",
        groupId = "attendance-group",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void listen(AttendanceStatisticsSingleDto attendance, 
                      Acknowledgment ack,
                      Headers headers) {
        // 獲取 Kafka 元數(shù)據(jù)
        int partition = Integer.parseInt(headers.lastHeader("partition").value().toString());
        long offset = headers.lastHeader("offset").value().toString().getBytes()[0];
        System.out.println("? 消息消費(fèi)成功 | 分區(qū): " + partition + 
                          " | 偏移量: " + offset + 
                          " | 用戶: " + attendance.getUserId());
        // 手動(dòng)提交偏移量（可選）
        ack.acknowledge();
    }
}

package com.example;
import com.example.service.AttendanceProducer;
import jnpf.model.attendance.event.AttendanceStatisticsSingleDto;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import java.util.Date;
@SpringBootTest
public class KafkaIntegrationTest {
    @Autowired
    private AttendanceProducer producer;
    @Test
    public void testSendAttendanceData() {
        AttendanceStatisticsSingleDto attendance = AttendanceStatisticsSingleDto.builder()
            .tenantId("tenant_001")
            .groupId("group_001")
            .userId("user_001")
            .day(new Date())
            .build();
        producer.sendAttendanceData(attendance);
        // 實(shí)際測(cè)試需等待消費(fèi)者處理（此處簡(jiǎn)化）
        System.out.println("? 測(cè)試消息已發(fā)送");
    }
}