Springboot使用kafka的两种方式

        <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;
 
/**
 * 启动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;
    }
 
    /**
     * 定义监听
     */
    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());
            // 处理数据
            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();
 
        // 这里可以配置几台broker即可，他会自动从broker去拉取元数据进行缓存
        props.put("bootstrap.servers", bootstrapServers);
        // 这个就是负责把发送的key从字符串序列化为字节数组
        props.put("key.serializer", keySerializer);
        // 这个就是负责把你发送的实际的message从字符串序列化为字节数组
        props.put("value.serializer", valueSerializer);
        // 默认是32兆=33554432
        props.put("buffer.memory", bufferMemory);
        // 一般来说是要自己手动设置的，不是纯粹依靠默认值的，16kb
        props.put("batch.size", batchSize);
        // 发送一条消息出去，100ms内还没有凑成一个batch发送，必须立即发送出去
        props.put("linger.ms", lingerMs);
        // 这个是说你可以发送的最大的请求的大小 默认是1m=1048576
//        props.put("max.request.size", 10485760);
        // follower有没有同步成功你就不管了
        props.put("acks", acks);
        // 这个重试，一般来说，给个3次~5次就足够了，可以cover住一般的异常场景
        props.put("retries", retries);
        // 每次重试间隔100ms
        props.put("retry.backoff.ms", retryBackOffMs);
 
        props.put("max.in.flight.requests.per.connection", maxInFlightRequestsPerConnection);
 
        return new KafkaProducer<>(props);
    }