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

spring:
  kafka:
    bootstrap-servers: kafka.tyjt.com:9092
    consumer:
      auto-offset-reset: earliest
      group-id: sharingan-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer

@Configuration
@EnableKafka
public class ConsumerConfigDemo {
    @KafkaListener(topics = {"test"},groupId = "group1")
    public void kafkaListener(String topic,String message){
        System.out.println("消息："+message);
    }
}

@Slf4j
@Service
public class KafkaProducerService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendMessage(String topic, String key, String message) throws Exception {
        kafkaTemplate.send(topic,key,message);
    }
}

package com.tyjt.sharingan.kafka;
 
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.listener.AcknowledgingConsumerAwareMessageListener;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
 
import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
 
/**
 * 啟動(dòng)kafka consumer
 *
 * @author 種鑫
 * @date 2023/10/18 17:26
 */
@EnableKafka
@Component
@Slf4j
public class KafkaConsumerMgr {
    @Resource
    ConcurrentKafkaListenerContainerFactory<String, byte[]> containerFactory;
 
    Map<String, ConcurrentMessageListenerContainer<?, ?>> containerMap = new ConcurrentHashMap<>();
    public void startListener(KafkaProtoConsumer kafkaConsumer) {
        //  停止相同的
        if (containerMap.containsKey(kafkaConsumer.getTopic())) {
            containerMap.get(kafkaConsumer.getTopic()).stop();
        }
        ConcurrentMessageListenerContainer<String, byte[]> container = createListenerContainer(kafkaConsumer);
        container.start();
        containerMap.put(kafkaConsumer.getTopic(), container);
    }
 
    private ConcurrentMessageListenerContainer<String, byte[]> createListenerContainer(KafkaProtoConsumer consumer) {
        ConcurrentMessageListenerContainer<String, byte[]> container = containerFactory.createContainer(consumer.topic());
        container.setBeanName(consumer.group() + "-" + consumer.topic());
        container.setConcurrency(consumer.getPartitionCount());
        consumer.deployContainer(container);
        // 防止被修改的配置
        ContainerProperties containerProperties = container.getContainerProperties();
        containerProperties.setMessageListener(new Listener<>(consumer));
        containerProperties.setAsyncAcks(false);
        containerProperties.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        containerProperties.setGroupId(consumer.group());
        return container;
    }
 
    /**
     * 定義監(jiān)聽
     */
    private static class Listener<T> implements AcknowledgingConsumerAwareMessageListener<String, T> {
        private final KafkaConsumer<T> kafkaConsumer;
 
        public Listener(KafkaConsumer<T> consumer) {
            this.kafkaConsumer = consumer;
        }
 
        @Override
        public void onMessage(ConsumerRecord<String, T> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
            log.info("group【{}】接收到來自topic【{}】的消息", kafkaConsumer.group(), data.topic());
            // 處理數(shù)據(jù)
            kafkaConsumer.process(data.value());
            // 提交offset
            log.info("group【{}】提交topic【{}】的offset", kafkaConsumer.group(), data.topic());
            consumer.commitSync();
        }
    }
}

@Bean
    public KafkaProducer<String, byte[]> kafkaProducer() {
 
        Properties props = new Properties();
 
        // 這里可以配置幾臺(tái)broker即可，他會(huì)自動(dòng)從broker去拉取元數(shù)據(jù)進(jìn)行緩存
        props.put("bootstrap.servers", bootstrapServers);
        // 這個(gè)就是負(fù)責(zé)把發(fā)送的key從字符串序列化為字節(jié)數(shù)組
        props.put("key.serializer", keySerializer);
        // 這個(gè)就是負(fù)責(zé)把你發(fā)送的實(shí)際的message從字符串序列化為字節(jié)數(shù)組
        props.put("value.serializer", valueSerializer);
        // 默認(rèn)是32兆=33554432
        props.put("buffer.memory", bufferMemory);
        // 一般來說是要自己手動(dòng)設(shè)置的，不是純粹依靠默認(rèn)值的，16kb
        props.put("batch.size", batchSize);
        // 發(fā)送一條消息出去，100ms內(nèi)還沒有湊成一個(gè)batch發(fā)送，必須立即發(fā)送出去
        props.put("linger.ms", lingerMs);
        // 這個(gè)是說你可以發(fā)送的最大的請(qǐng)求的大小 默認(rèn)是1m=1048576
//        props.put("max.request.size", 10485760);
        // follower有沒有同步成功你就不管了
        props.put("acks", acks);
        // 這個(gè)重試，一般來說，給個(gè)3次~5次就足夠了，可以cover住一般的異常場(chǎng)景
        props.put("retries", retries);
        // 每次重試間隔100ms
        props.put("retry.backoff.ms", retryBackOffMs);
 
        props.put("max.in.flight.requests.per.connection", maxInFlightRequestsPerConnection);
 
        return new KafkaProducer<>(props);
    }