spring:
  application:
    name: spring-boot-01-kafka-base
  kafka:
    bootstrap-servers: 192.168.2.118:9092

package com.zzc.producer;
import jakarta.annotation.Resource;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
@Component
public class EventProducer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendEvent(){
        kafkaTemplate.send("hello-topic", "hello kafka");
    }
}

package com.zzc.producer;
import jakarta.annotation.Resource;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
@Component
public class EventProducer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendEvent(){
        kafkaTemplate.send("hello-topic", "hello kafka");
    }
}

package com.zzc.cosumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class EventConsumer {
    // 采用監(jiān)聽的方式接收事件（消息、數(shù)據(jù)）
    @KafkaListener(topics = {"hello-topic"}, groupId = "hello-group")
    public void onEvent(String event){
        System.out.printf("讀取到的事件：" + event);
    }
}

spring:
 	kafka:
 		consumer:
 			auto-offset-reset: earliest

@Component
public class EventConsumer {
    // 采用監(jiān)聽的方式接收事件（消息、數(shù)據(jù)）
    @KafkaListener(topics = {"hello-topic"}, groupId = "hello-group-02")
    public void onEvent(String event){
        System.out.println("讀取到的事件：" + event);
    }
}

修改為讀取最早的消息
./kafka-consumer-groups.sh --bootstrap-server <your-kafka-bootstrap-servers> --group <your-consumer-group> --topic <your-topic> --reset-offsets --to-earliest --execute
修改為讀取最新的消息
./kafka-consumer-groups.sh --bootstrap-server <your-kafka-bootstrap-servers> --group <your-consumer-group> --topic <your-topic> --reset-offsets --to-latest --execute

./kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group hello-group-02 --topic hello-topic --reset-offsets --to-earliest --execute

    /**
     * 使用message對(duì)象發(fā)送消息
     */
    public void sendEvent02(){
        // 通過構(gòu)建器模式創(chuàng)建Message對(duì)象
        Message<String> message = MessageBuilder.withPayload("hello kafka")
                // 在header中放置topic的名字
                .setHeader(KafkaHeaders.TOPIC, "test-topic-02")
                .build();
        kafkaTemplate.send(message);
    }

@Test
public void test02(){
    eventProducer.sendEvent02();
}

    /**
     * 使用ProducerRecord對(duì)象發(fā)送消息
     */
    public void sendEvent03(){
        // Headers里面是放一些信息（信息是key-value鍵值對(duì)），到時(shí)候消費(fèi)者接收到該消息后，可以拿到這個(gè)Headers里面放的信息
        Headers headers = new RecordHeaders();
        headers.add("phone", "13698001234".getBytes(StandardCharsets.UTF_8));
        headers.add("orderId", "12473289472846178242873".getBytes(StandardCharsets.UTF_8));
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                "test-topic-02",
                0,
                System.currentTimeMillis(),
                "k1",
                "hello kafka",
                headers
        );
        kafkaTemplate.send(producerRecord);
    }

@Test
public void test03(){
    eventProducer.sendEvent03();
}

    public void sendEvent04() {
        // String topic, Integer partition, Long timestamp, K key, @Nullable V data
        kafkaTemplate.send(
                "test-topic-02",
                0,
                System.currentTimeMillis(),
                "k2",
                "hello kafka"
        );
    }

@Test
public void test04(){
    eventProducer.sendEvent04();
}

public void sendEvent05(){
    kafkaTemplate.sendDefault(0, System.currentTimeMillis(), "k3", "hello kafka");
}

@Test
public void test04(){
    eventProducer.sendEvent04();
}

spring:
  application:
    name: spring-boot-01-kafka-base
  kafka:
    bootstrap-servers: 192.168.2.118:9092
    consumer:
      auto-offset-reset: earliest
    # 配置模板默認(rèn)的主題topic名稱
    template:
      default-topic: default-topic

 /**
     * 通過get方法同步阻塞等待發(fā)送結(jié)果
     */
    public void sendEvent06(){
        CompletableFuture<SendResult<String, String>> completableFuture =
                kafkaTemplate.sendDefault(0, System.currentTimeMillis(), "k3", "hello kafka");
        try {
            // 1.阻塞等待的方式拿結(jié)果
            SendResult<String, String> sendResult = completableFuture.get();
            if (sendResult.getRecordMetadata() != null){
                // kafka服務(wù)器確認(rèn)已經(jīng)接收到了消息
                System.out.println("消息發(fā)送成功：" + sendResult.getRecordMetadata().toString());
            }
            System.out.println("producerRecord: " + sendResult.getProducerRecord());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

@Test
public void test06(){
    eventProducer.sendEvent06();
}

    /**
     * 通過thenAccept方法注冊(cè)回調(diào)函數(shù)
     */
    public void sendEvent07(){
        CompletableFuture<SendResult<String, String>> completableFuture =
                kafkaTemplate.sendDefault(0, System.currentTimeMillis(), "k3", "hello kafka");
        completableFuture.thenAccept(sendResult -> {
            if (sendResult.getRecordMetadata() != null){
                // kafka服務(wù)器確認(rèn)已經(jīng)接收到了消息
                System.out.println("消息發(fā)送成功：" + sendResult.getRecordMetadata().toString());
            }
            System.out.println("producerRecord: " + sendResult.getProducerRecord());
        }).exceptionally( throwable -> {
            // 做失敗的處理
            throwable.printStackTrace();
            return null;
        });
    }

@Test
public void test07(){
    eventProducer.sendEvent07();
}

@Builder
@AllArgsConstructor
@NoArgsConstructor
@Data
public class User {
    private int id;
    private String phone;
    private Date birthDay;
}

/**
* 發(fā)送對(duì)象消息
*/
@Resource
private KafkaTemplate<String, Object> kafkaTemplate2;
private KafkaTemplate<String, Object> kafkaTemplate2;
public void sendEvent08(){
    User user = User.builder().id(1200).phone("13698981234").birthDay(new Date()).build();
    // 分區(qū)編號(hào)為 null ，交給 kafka 自己去分配
    kafkaTemplate2.sendDefault(null, System.currentTimeMillis(), "k4", user);
}

    producer:
      # key和value都默認(rèn)是StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

./kafka-topics.sh --create --topic myTopic --partitions 3 --replication-factor 1 --bootstrap-server 127.0.0.1:9092

@Configuration
public class KafkaConfig {
    // 創(chuàng)建一個(gè)名為helloTopic的Topic并設(shè)置分區(qū)數(shù)為5，分區(qū)副本數(shù)為1
    @Bean
    public NewTopic newTopic(){
        // 副本不能設(shè)置為0 也不能超過節(jié)點(diǎn)數(shù)
        return new NewTopic("helloTopic", 5, (short) 1);
    }
}

    public void sendEvent09(){
        User user = User.builder().id(1200).phone("13698981234").birthDay(new Date()).build();
        kafkaTemplate2.send(
                "helloTopic",
                null,
                System.currentTimeMillis(),
                "k9",
                user
        );    
    }

@Test
public void test09(){
    eventProducer.sendEvent09();
}

@Configuration
public class KafkaConfig {
    // 創(chuàng)建一個(gè)名為helloTopic的Topic并設(shè)置分區(qū)數(shù)為5，分區(qū)副本數(shù)為1
    @Bean
    public NewTopic newTopic(){
        return new NewTopic("helloTopic", 5, (short) 1);
    }
    // 如果要修改分區(qū)數(shù)，只需修改配置值重啟項(xiàng)目即可，修改分區(qū)數(shù)并不會(huì)導(dǎo)致數(shù)據(jù)的丟失，但是分區(qū)數(shù)只能增大不能減少
    @Bean
    public NewTopic updateTopic(){
        return new NewTopic("helloTopic", 10, (short) 1);
    }
}

spring:
  application:
    name: spring-boot-01-kafka-base
  kafka:
    bootstrap-servers: 192.168.2.118:9092
    producer:
      # key和value都默認(rèn)是StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      auto-offset-reset: earliest
    # 配置模板默認(rèn)的主題topic名稱
    template:
      default-topic: default-topic

package com.zzc.config;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.RoundRobinPartitioner;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class KafkaConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    @Value("${spring.kafka.producer.value-serializer}")
    private String valueSerializer;
    @Value("${spring.kafka.producer.key-serializer}")
    private String keySerializer;
    /**
     * 生產(chǎn)者相關(guān)配置
     * @return
     */
    public Map<String, Object> producerConfigs(){
        HashMap<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, RoundRobinPartitioner.class);
        return props;
    }
    public ProducerFactory<String, Object> producerFactory(){
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
    /**
     * KafkaTemplate 覆蓋相關(guān)配置類中的kafkaTemplate
     * @return
     */
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate(){
        return new KafkaTemplate<>(producerFactory());
    }
    // 創(chuàng)建一個(gè)名為helloTopic的Topic并設(shè)置分區(qū)數(shù)為5，分區(qū)副本數(shù)為1
    @Bean
    public NewTopic newTopic(){
        return new NewTopic("helloTopic", 5, (short) 1);
    }
    // 如果要修改分區(qū)數(shù)，只需修改配置值重啟項(xiàng)目即可，修改分區(qū)數(shù)并不會(huì)導(dǎo)致數(shù)據(jù)的丟失，但是分區(qū)數(shù)只能增大不能減少
    @Bean
    public NewTopic updateTopic(){
        return new NewTopic("helloTopic", 10, (short) 1);
    }
}

public void sendEvent09(){
    User user = User.builder().id(1200).phone("13698981234").birthDay(new Date()).build();
    kafkaTemplate2.send(
        "helloTopic",
        user
    );    }

@Test
public void test09(){
    for (int i = 0; i < 5; i++) {
        eventProducer.sendEvent09();
    }
}

public class CustomerPartitioner implements Partitioner {
    private AtomicInteger nextPartition = new AtomicInteger(0);
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] bytes1, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (key == null){
            // 使用輪詢方式選擇分區(qū)
            int next = nextPartition.getAndIncrement();
            // 如果next大于分區(qū)的大小，則重置為0
            if (next >= numPartitions){
                nextPartition.compareAndSet(next, 0);
            }
            System.out.println("分區(qū)值：" + next);
            return next;
        }else {
            // 如果key不為null，則使用默認(rèn)的分區(qū)策略
            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    }
    @Override
    public void close() {
    }
    @Override
    public void configure(Map<String, ?> map) {
    }
}

package com.zzc.config;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;
public class CustomerProducerInterceptor implements ProducerInterceptor<String, Object> {
    /**
     * 發(fā)送消息時(shí)，會(huì)先調(diào)用該方法，對(duì)信息進(jìn)行攔截，可以在攔截中對(duì)消息做一些處理，記錄日志等操作...
     * @param producerRecord
     * @return
     */
    @Override
    public ProducerRecord<String, Object> onSend(ProducerRecord<String, Object> producerRecord) {
        System.out.println("攔截消息：" + producerRecord.toString());
        return producerRecord;
    }
    /**
     * 服務(wù)器收到消息后的一個(gè)確認(rèn)
     * @param recordMetadata
     * @param e
     */
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if (recordMetadata != null){
            System.out.println("服務(wù)器收到該消息：" + recordMetadata.offset());
        }else {
            System.out.println("消息發(fā)送失敗了，exception = " + e.getMessage());
        }
    }
    @Override
    public void close() {
    }
    @Override
    public void configure(Map<String, ?> map) {
    }
}

@Component
public class EventConsumer {
    // 采用監(jiān)聽的方式接收事件（消息、數(shù)據(jù)）
    @KafkaListener(topics = {"helloTopic"}, groupId = "helloGroup")
    public void onEvent(String event){
        System.out.println("讀取到的事件：" + event);
    }
}

@Component
public class EventProducer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendEvent() {
        kafkaTemplate.send("helloTopic", "hello kafka");
    }
}

spring:
  application:
    name: spring-boot-02-kafka-base
  kafka:
    bootstrap-servers: 192.168.2.118:9092

@SpringBootTest
class KafkaBaseApplicationTests {
    @Resource
    private EventProducer eventProducer;
    @Test
    void test01(){
        System.out.println(111);
        eventProducer.sendEvent();
    }
}

@Component
public class EventConsumer {
    // 采用監(jiān)聽的方式接收事件（消息、數(shù)據(jù)）
    @KafkaListener(topics = {"helloTopic"}, groupId = "helloGroup")
    public void onEvent(@Payload String event,
                        @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition
                        ){
        System.out.println("讀取到的事件：" + event +  ", topic:" + topic + ", partition:" + partition);
    }
}

@Component
public class EventConsumer {
    // 采用監(jiān)聽的方式接收事件（消息、數(shù)據(jù)）
    @KafkaListener(topics = {"helloTopic"}, groupId = "helloGroup")
    public void onEvent(@Payload String event,
                        @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                        ConsumerRecord<String, String> consumerRecord
                        ){
        System.out.println("讀取到的事件：" + event +  ", topic:" + topic + ", partition:" + partition);
        System.out.println("讀取到的consumerRecord：" + consumerRecord.toString());
    }
}

package com.zzc.model;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Date;
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Data
public class User {
    private int id;
    private String phone;
    private Date birthDay;
}

    @KafkaListener(topics = {"helloTopic"}, groupId = "helloGroup")
    public void onEvent2(User user,
                         @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                         @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                         ConsumerRecord<String, String> consumerRecord
                        ){
        System.out.println("讀取到的事件：" + user +  ", topic:" + topic + ", partition:" + partition);
        System.out.println("讀取到的consumerRecord：" + consumerRecord.toString());
    }

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate2;
    public void sendEvent2(){
        User user = User.builder().id(213234).phone("13239407234").birthDay(new Date()).build();
        kafkaTemplate2.send("helloTopic", user);
    }

    @Test
    public void test02(){
        eventProducer.sendEvent2();
    }

spring:
  application:
    name: spring-boot-02-kafka-base
  kafka:
    bootstrap-servers: 192.168.2.118:9092
    producer:
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      value-deserializer: org.springframework.kafka.support.seri

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-json</artifactId>
</dependency>

package com.zzc.util;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class JSONUtils {
    private static final ObjectMapper OBJECTMAPPER = new ObjectMapper();
    public static String toJSON(Object object){
        try {
            return OBJECTMAPPER.writeValueAsString(object);
        }catch (JsonProcessingException e){
            throw new RuntimeException(e);
        }
    }
    public static <T> T toBean(String jsonStr, Class<T> clazz){
        try {
            return OBJECTMAPPER.readValue(jsonStr, clazz);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }
}

	public void sendEvent2(){
        User user = User.builder().id(213234).phone("13239407234").birthDay(new Date()).build();
        String userJson = JSONUtils.toJSON(user);
        kafkaTemplate.send("helloTopic", userJson);
    }

    @KafkaListener(topics = {"helloTopic"}, groupId = "helloGroup")
    public void onEvent2(String userStr,
                         @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                         @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                         ConsumerRecord<String, String> consumerRecord
    ){
        User user = (User) JSONUtils.toBean(userStr, User.class);
        System.out.println("讀取到的事件：" + user +  ", topic:" + topic + ", partition:" + partition);
        System.out.println("讀取到的consumerRecord：" + consumerRecord.toString());
    }

spring:
  application:
    name: spring-boot-02-kafka-base
  kafka:
    bootstrap-servers: 192.168.2.118:9092
#    producer:
#      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
#    consumer:
#      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer

spring:
  application:
    name: spring-boot-02-kafka-base
  kafka:
    bootstrap-servers: 192.168.2.118:9092
#    producer:
#      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
#    consumer:
#      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
kafka:
  topic:
    name: helloTopic
  consumer:
    group: helloGroup

    @KafkaListener(topics = {"${kafka.topic.name}"}, groupId = "kafka.consumer.group")
    public void onEvent3(String userStr,
                         @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                         @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                         ConsumerRecord<String, String> consumerRecord
    ){
        User user = (User) JSONUtils.toBean(userStr, User.class);
        System.out.println("讀取到的事件3：" + user +  ", topic:" + topic + ", partition:" + partition);
        System.out.println("讀取到的consumerRecord3：" + consumerRecord.toString());
    }

    @KafkaListener(topics = {"${kafka.topic.name}"}, groupId = "kafka.consumer.group")
    public void onEvent4(String userStr,
                         @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                         @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                         ConsumerRecord<String, String> consumerRecord,
                         Acknowledgment acknowledgment
    ){
        User user = (User) JSONUtils.toBean(userStr, User.class);
        System.out.println("讀取到的事件4：" + user +  ", topic:" + topic + ", partition:" + partition);
        System.out.println("讀取到的consumerRecord4：" + consumerRecord.toString());

    }

  kafka:
    bootstrap-servers: 192.168.2.118:9092
    listener:
      ack-mode: manual

    @KafkaListener(topics = {"${kafka.topic.name}"}, groupId = "kafka.consumer.group")
    public void onEvent4(String userStr,
                         @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                         @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                         ConsumerRecord<String, String> consumerRecord,
                         Acknowledgment acknowledgment
    ){
        try {
            User user = (User) JSONUtils.toBean(userStr, User.class);
            System.out.println("讀取到的事件4：" + user +  ", topic:" + topic + ", partition:" + partition);
            System.out.println("讀取到的consumerRecord4：" + consumerRecord.toString());
            int i = 1 / 0;
            // 可以執(zhí)行完所有業(yè)務(wù)，再進(jìn)行確認(rèn)消息。如果執(zhí)行過程中發(fā)生異常，那么可以再次消費(fèi)此消息
            acknowledgment.acknowledge();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

@Configuration
public class KafkaConfig {
    // 創(chuàng)建一個(gè)名為helloTopic的Topic并設(shè)置分區(qū)數(shù)為5，分區(qū)副本數(shù)為1
    @Bean
    public NewTopic newTopic(){
        return new NewTopic("helloTopic", 5, (short) 1);
    }
}

 @KafkaListener(groupId = "${kafka.consumer.group}",
                	// 配置更加詳細(xì)的監(jiān)聽信息 topics和topicPartitions不能同時(shí)使用
                    topicPartitions = {
                        @TopicPartition(
                                topic = "${kafka.topic.name}",
                            	// 監(jiān)聽topic的0、1、2號(hào)分區(qū)的所有消息
                                partitions = {"0", "1", "2"},
                            	// 監(jiān)聽3、4號(hào)分區(qū)中offset從3開始的消息
                                partitionOffsets = {
                                        @PartitionOffset(partition = "3", initialOffset = "3"),
                                        @PartitionOffset(partition = "4", initialOffset = "3")
                                }
                        )
                    })
    public void onEvent5(String userStr,
                         @Header(value = KafkaHeaders.RECEIVED_TOPIC) String topic,
                         @Header(value = KafkaHeaders.RECEIVED_PARTITION) String partition,
                         ConsumerRecord<String, String> consumerRecord,
                         Acknowledgment acknowledgment
    ){
        try {
            User user = (User) JSONUtils.toBean(userStr, User.class);
            System.out.println("讀取到的事件5：" + user +  ", topic:" + topic + ", partition:" + partition);
            System.out.println("讀取到的consumerRecord5：" + consumerRecord.toString());
            acknowledgment.acknowledge();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public void sendEvent3(){
        for (int i = 0; i < 25; i++) {
            User user = User.builder().id(i).phone("13239407234" + i).birthDay(new Date()).build();
            String userJson = JSONUtils.toJSON(user);
            kafkaTemplate2.send("helloTopic", "k" + i, userJson);
        }

    }

    @Test
    public void test03(){
        eventProducer.sendEvent3();
    }

    consumer:
      # 從最早的消息開始消費(fèi)
      auto-offset-reset: earliest

spring:
  application:
    name: spring-boot-03-kafka-base
  kafka:
    bootstrap-servers: 192.168.2.118:9092
    consumer:
      # 設(shè)置批量最多消費(fèi)多少條消息
      max-poll-records: 20
    listener:
      # 設(shè)置批量消費(fèi)
      type: batch

package com.zzc.springboot03kafkabase.cosumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
public class EventConsumer {
    @KafkaListener(topics = "batchTopic", groupId = "bactchGroup")
    public void onEvent(List<ConsumerRecord<String, String>> records) {
        System.out.println(" 批量消費(fèi)， records.size() = " + records.size() + " ， records = " + records);
    }
}

package com.zzc.springboot03kafkabase.model;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Date;
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Data
public class User {
    private int id;
    private String phone;
    private Date birthDay;
}

package com.zzc.springboot03kafkabase.producer;
import com.zzc.springboot03kafkabase.model.User;
import com.zzc.springboot03kafkabase.util.JSONUtils;
import jakarta.annotation.Resource;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.util.Date;
@Component
public class EventProducer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendEvent(){
        for (int i = 0; i < 125; i++) {
            User user = User.builder().id(i).phone("13239407234" + i).birthDay(new Date()).build();
            String userJson = JSONUtils.toJSON(user);
            kafkaTemplate.send("batchTopic", "k" + i, userJson);
        }
    }
}

package com.zzc.springboot03kafkabase.util;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class JSONUtils {
    private static final ObjectMapper OBJECTMAPPER = new ObjectMapper();
    public static String toJSON(Object object){
        try {
            return OBJECTMAPPER.writeValueAsString(object);
        }catch (JsonProcessingException e){
            throw new RuntimeException(e);
        }
    }
    public static <T> T toBean(String jsonStr, Class<T> clazz){
        try {
            return OBJECTMAPPER.readValue(jsonStr, clazz);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.2.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.zzc</groupId>
    <artifactId>spring-boot-03-kafka-base</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>spring-boot-03-kafka-base</name>
    <description>spring-boot-03-kafka-base</description>
    <properties>
        <java.version>17</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-json</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

    consumer:
      # 設(shè)置批量最多消費(fèi)多少條消息
      max-poll-records: 20
      auto-offset-reset: earliest

package com.zzc;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import java.util.Map;
@SpringBootApplication
public class SpringBoot04KafkaBaseApplication {
    public static void main(String[] args) {
        ApplicationContext context = SpringApplication.run(SpringBoot04KafkaBaseApplication.class, args);
        Map<String, ConsumerFactory> beansOfType = context.getBeansOfType(ConsumerFactory.class);
        beansOfType.forEach((k, v) -> {
            System.out.println(k + " -- " + v);
        });
        Map<String, KafkaListenerContainerFactory> beansOfType2 = context.getBeansOfType(KafkaListenerContainerFactory.class);
        beansOfType2.forEach((k, v) -> {
            System.out.println(k + " -- " + v);
        });
    }
}

package com.zzc.interceptor;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import java.util.Map;
public class CustomConsumerInterceptor implements ConsumerInterceptor<String, String > {
    /**
     * 在消費(fèi)消息之前執(zhí)行
     * @param consumerRecords
     * @return
     */
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> consumerRecords) {
        System.out.println("onConsumer方法執(zhí)行：" + consumerRecords);
        return consumerRecords;
    }
    /**
     * 消息拿到之后，提交offset之前執(zhí)行該方法
     * @param offsets
     */
    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        System.out.println("onCommit方法執(zhí)行：" + offsets);
    }
    @Override
    public void close() {
    }
    @Override
    public void configure(Map<String, ?> map) {
    }
}

package com.zzc.config;
import com.zzc.interceptor.CustomConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class KafkaConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    @Value("${spring.kafka.consumer.value-deserializer}")
    private String valueDeSerializer;
    @Value("${spring.kafka.consumer.key-deserializer}")
    private String keyDeSerializer;
    public Map<String, Object> consumerConfigs(){
        HashMap<String, Object> consumer = new HashMap<>();
        consumer.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServers);
        consumer.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,keyDeSerializer);
        consumer.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeSerializer);
        // 添加一個(gè)消費(fèi)攔截器
        consumer.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, CustomConsumerInterceptor.class.getName());
        return consumer;
    }
    /**
     * 消費(fèi)者創(chuàng)建工廠
     * @return
     */
    @Bean
    public ConsumerFactory<String, String> ourConsumerFactory(){
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }
    /**
     * 監(jiān)聽器容器工廠
     * @param ourConsumerFactory
     * @return
     */
    @Bean
    public KafkaListenerContainerFactory ourKafkaListenerContainerFactory(ConsumerFactory ourConsumerFactory){
        ConcurrentKafkaListenerContainerFactory<String, String> listenerContainerFactory = new ConcurrentKafkaListenerContainerFactory<>();
        listenerContainerFactory.setConsumerFactory(ourConsumerFactory);
        return listenerContainerFactory;
    }
}

package com.zzc.cosumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
public class EventConsumer {
    @KafkaListener(topics = {"interTopic"}, groupId = "interGroup", containerFactory = "ourKafkaListenerContainerFactory")
    public void onEvent(ConsumerRecord<String, String> records) {
        System.out.println(" 消費(fèi)消息， records = " + records);
    }
}

package com.zzc.producer;
import com.zzc.model.User;
import com.zzc.util.JSONUtils;
import jakarta.annotation.Resource;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.util.Date;
@Component
public class EventProducer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendEvent() {
        User user = User.builder().id(1023).phone("13239407234").birthDay(new Date()).build();
        String userJson = JSONUtils.toJSON(user);
        kafkaTemplate.send("interTopic", "k", userJson);
    }
}

    @Resource
    private EventProducer eventProducer;
    @Test
    public void test(){
        eventProducer.sendEvent();
    }

@Component
public class EventConsumer {
//    @KafkaListener(topics = {"interTopic"}, groupId = "interGroup", containerFactory = "ourKafkaListenerContainerFactory")
    @KafkaListener(topics = {"interTopic"}, groupId = "interGroup", )
    public void onEvent(ConsumerRecord<String, String> records) {
        System.out.println(" 消費(fèi)消息， records = " + records);
    }
}

package com.zzc.cosumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
public class EventConsumer {
    @KafkaListener(topics = {"topicA"}, groupId = "group1")
    @SendTo("topicB")  // 轉(zhuǎn)發(fā)消息給topicB
    public String onEvent(ConsumerRecord<String, String> record) {
        System.out.println(" 消費(fèi)消息， record = " + record);
        return record.value() + "forward message";
    }
    @KafkaListener(topics = {"topicB"}, groupId = "group2")
    public void onEvent2(List<ConsumerRecord<String, String>> records) {
        System.out.println(" 消費(fèi)消息， record = " + records);
    }
}

package com.zzc.producer;
import com.zzc.model.User;
import com.zzc.util.JSONUtils;
import jakarta.annotation.Resource;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.util.Date;
@Component
public class EventProducer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendEvent() {
        User user = User.builder().id(1023).phone("13239407234").birthDay(new Date()).build();
        String userJson = JSONUtils.toJSON(user);
        kafkaTemplate.send("topicA", "k", userJson);
    }
}

package com.zzc.config;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class KafkaConfig {
    // 創(chuàng)建一個(gè)名為helloTopic的Topic并設(shè)置分區(qū)數(shù)為5，分區(qū)副本數(shù)為1
    @Bean
    public NewTopic newTopic(){
        return new NewTopic("myTopic", 10, (short) 1);
    }
}

package com.zzc.cosumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
public class EventConsumer {
    // concurrency 消費(fèi)者數(shù)量
    @KafkaListener(topics = {"myTopic"}, groupId = "myGroup", concurrency = "3")
    public void onEvent(ConsumerRecord<String, String> records) {
        System.out.println(" 消費(fèi)消息， records = " + records);
    }
}

package com.zzc.producer;
import com.zzc.model.User;
import com.zzc.util.JSONUtils;
import jakarta.annotation.Resource;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.util.Date;
@Component
public class EventProducer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendEvent() {
        for (int i = 0; i < 100; i++) {
            User user = User.builder().id(i).phone("13239407234" + i).birthDay(new Date()).build();
            String userJson = JSONUtils.toJSON(user);
            kafkaTemplate.send("myTopic", "k" + i, userJson);
        }
    }
}

package com.zzc;
import com.zzc.producer.EventProducer;
import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class SpringBoot06KafkaBaseApplicationTests {
    @Resource
    private EventProducer eventProducer;
    @Test
    public void test(){
        eventProducer.sendEvent();
    }
}

spring:
  application:
    name: spring-boot-06-kafka-base
  kafka:
    bootstrap-servers: 192.168.2.118:9092
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: earliest

package com.zzc.config;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.RoundRobinAssignor;
import org.apache.kafka.clients.producer.RoundRobinPartitioner;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class KafkaConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    @Value("${spring.kafka.consumer.value-deserializer}")
    private String valueDeSerializer;
    @Value("${spring.kafka.consumer.key-deserializer}")
    private String keyDeSerializer;
    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;
    public Map<String, Object> consumerConfigs(){
        HashMap<String, Object> consumer = new HashMap<>();
        consumer.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServers);
        consumer.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,keyDeSerializer);
        consumer.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeSerializer);
        consumer.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        // 設(shè)置消費(fèi)者策略為輪詢模式
        consumer.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName());
        return consumer;
    }
    // 創(chuàng)建一個(gè)名為helloTopic的Topic并設(shè)置分區(qū)數(shù)為5，分區(qū)副本數(shù)為1
    @Bean
    public NewTopic newTopic(){
        return new NewTopic("myTopic", 10, (short) 1);
    }
    /**
     * 消費(fèi)者創(chuàng)建工廠
     * @return
     */
    @Bean
    public ConsumerFactory<String, String> ourConsumerFactory(){
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }
    /**
     * 監(jiān)聽器容器工廠
     * @param ourConsumerFactory
     * @return
     */
    @Bean
    public KafkaListenerContainerFactory ourKafkaListenerContainerFactory(ConsumerFactory ourConsumerFactory){
        ConcurrentKafkaListenerContainerFactory<String, String> listenerContainerFactory = new ConcurrentKafkaListenerContainerFactory<>();
        listenerContainerFactory.setConsumerFactory(ourConsumerFactory);
        return listenerContainerFactory;
    }
}

package com.zzc.cosumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
public class EventConsumer {
    // concurrency 設(shè)置消費(fèi)者數(shù)量    containerFactory 設(shè)置監(jiān)聽器容器工廠
    @KafkaListener(topics = {"myTopic"}, groupId = "myGroup4", concurrency = "3", containerFactory = "ourKafkaListenerContainerFactory")
    public void onEvent(ConsumerRecord<String, String> records) {
        System.out.println(Thread.currentThread().getId() + " --> 消費(fèi)消息， records = " + records);
    }
}

 每次消費(fèi)一個(gè)消息并且提交以后，會(huì)保存當(dāng)前消費(fèi)到的最近的一個(gè) offset ；
 在 kafka 中，有一個(gè) __consumer_offsets 的 topic ， 消費(fèi)者消費(fèi)提交的 offset 信息會(huì)寫入到
該 topic 中， __consumer_offsets 保存了每個(gè) consumer group 某一時(shí)刻提交的 offset 信息
， __consumer_offsets 默認(rèn)有 50 個(gè)分區(qū)；
 consumer_group 保存在哪個(gè)分區(qū)中的計(jì)算公式：
 Math.abs(“groupid”.hashCode())%groupMetadataTopicPartitionCount ;

package com.zzc.cosumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class EventConsumer {
    @KafkaListener(topics = {"offsetTopic"}, groupId = "offsetGroup")
    public void onEvent(ConsumerRecord<String, String> records) {
        System.out.println(Thread.currentThread().getId() + " --> 消費(fèi)消息， records = " + records);
    }
}

package com.zzc.producer;
import com.zzc.model.User;
import com.zzc.util.JSONUtils;
import jakarta.annotation.Resource;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.util.Date;
@Component
public class EventProducer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendEvent() {
        for (int i = 0; i < 2; i++) {
            User user = User.builder().id(i).phone("13239407234" + i).birthDay(new Date()).build();
            String userJson = JSONUtils.toJSON(user);
            kafkaTemplate.send("offsetTopic", "k" + i, userJson);
        }
    }
}

spring:
  application:
    name: spring-boot-07-kafka-base
  kafka:
    bootstrap-servers: 192.168.2.118:9092
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

package com.zzc;
import com.zzc.producer.EventProducer;
import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class SpringBoot07KafkaBaseApplicationTests {
    @Resource
    private EventProducer eventProducer;
    @Test
    public void test(){
        eventProducer.sendEvent();
    }
}