springboot和kafka快速整合代码示例，三种方式使用默认配置、自定义配置、手动提交配置

目录1、第一种方式，使用默认配置注解开发，简单实用1.1、引入maven1.2、基本配置1.3、生产者，使用KafkaTemplate发送消息1.4、消费者，使用@KafkaListener注解2、不使用默认配置，使用自定义参数，自动提交2.1 基本配置2.2、生产者配置2.3、生产者生产消息2.4、消费者配置2.5、消费者消费3、不使用默认配置，使用自定义参数，手动提交3.1 基本配置和生产配置

一点光辉

1668人浏览 · 2021-05-16 16:57:19

一点光辉 · 2021-05-16 16:57:19 发布

目录

1、第一种方式，使用默认配置注解开发，简单实用

1.1、引入maven

1.2、基本配置

1.3、生产者，使用KafkaTemplate发送消息

1.4、消费者，使用@KafkaListener注解

2、不使用默认配置，使用自定义参数，自动提交

2.1 基本配置

2.2、生产者配置

2.3、生产者生产消息

2.4、消费者配置

2.5、消费者消费

3、不使用默认配置，使用自定义参数，手动提交

3.1 基本配置和生产配置同2.1-2.3

3.2 消费者配置手动提交

3.3、消费者消费手动提交确认

1、第一种方式，使用默认配置注解开发，简单实用

1.1、引入maven

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

1.2、基本配置

#============== kafka ===================
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.auto-commit-interval=100
#消费者群组
spring.kafka.consumer.group-id=test-consumer-group
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer


spring.kafka.producer.bootstrap-servers=localhost:9092
spring.kafka.producer.retries=0
spring.kafka.producer.batch-size=4096
spring.kafka.producer.buffer-memory=40960
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

1.3、生产者，使用KafkaTemplate发送消息

@RequestMapping("/kafka")
public class KafkaController {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @RequestMapping(value = "/send")
    public String sendKafka(@RequestParam(required = false) String key,
                            @RequestParam(required = false) String value) {
        try {
            kafkaTemplate.send("test", key, value);
            return "发送kafka成功";
        } catch (Exception e) {
            return "发送kafka失败";
        }
    }
}

1.4、消费者，使用@KafkaListener注解

@Component
public class MyListener {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());


    //可以定义多个主题
    @KafkaListener(topics = {"test"})
    public void listen(ConsumerRecord<?, ?> record) {
        logger.info("收到消息的value: " + record.value().toString());
    }
}

2、不使用默认配置，使用自定义参数，自动提交

2.1 基本配置

kafka.consumer.zookeeper.connect=localhost:21810
kafka.consumer.servers=localhost:9092
kafka.consumer.enable.auto.commit=true
kafka.consumer.session.timeout=6000
kafka.consumer.auto.commit.interval=100
kafka.consumer.auto.offset.reset=latest
kafka.consumer.topic=test
kafka.consumer.group.id=test
kafka.consumer.concurrency=10

kafka.producer.servers=localhost:9092
kafka.producer.retries=0
kafka.producer.batch.size=4096
kafka.producer.linger=1
kafka.producer.buffer.memory=40960

2.2、生产者配置

@Configuration
@EnableKafka
public class KafkaProducerConfig {
    @Value("${kafka.producer.servers}")
    private String servers;
    @Value("${kafka.producer.retries}")
    private int retries;
    @Value("${kafka.producer.batch.size}")
    private int batchSize;
    @Value("${kafka.producer.linger}")
    private int linger;
    @Value("${kafka.producer.buffer.memory}")
    private int bufferMemory;


    //自定义配置
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    /**
     * 自定义的KafkaTemplate，用来生产消息
     *
     * @return KafkaTemplate<String, String>
     */
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        KafkaTemplate kafkaTemplate
                = new KafkaTemplate<>(producerFactory());
        //kafkaTemplate.setProducerListener();
        return kafkaTemplate;
    }
}

2.3、生产者生产消息

@RestController
@RequestMapping("/kafka")
public class KafkaController {
    //这里注入的是配置里面自定义的KafkaTemplate 
    @Autowired
    private KafkaTemplate kafkaTemplate;

    @RequestMapping(value = "/send")
    public String sendKafka(@RequestParam(required = false) String key,
                            @RequestParam(required = false) String value) {
        try {
            kafkaTemplate.send("test", key, value);
            return "发送kafka成功";
        } catch (Exception e) {
            return "发送kafka失败";
        }
    }
}

2.4、消费者配置

@Configuration
@EnableKafka
public class KafkaConsumerConfig {
    @Value("${kafka.consumer.servers}")
    private String servers;
    @Value("${kafka.consumer.enable.auto.commit}")
    private boolean enableAutoCommit;
    @Value("${kafka.consumer.session.timeout}")
    private String sessionTimeout;
    @Value("${kafka.consumer.auto.commit.interval}")
    private String autoCommitInterval;
    @Value("${kafka.consumer.group.id}")
    private String groupId;
    @Value("${kafka.consumer.auto.offset.reset}")
    private String autoOffsetReset;
    @Value("${kafka.consumer.concurrency}")
    private int concurrency;

    /**
     * 基本配置
     * @return Map<String, Object>
     */
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        return propsMap;
    }

    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }
    //@KafkaListener(topics = {"testAck"},containerFactory = "")
    //containerFactory默认是当前bean
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory
                = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(concurrency);
        factory.getContainerProperties().setPollTimeout(1500);
        return factory;
    }
}

2.5、消费者消费

@Component
public class MyListener {

    @KafkaListener(topics = {"test"})
    public void listen(ConsumerRecord<?, ?> record) {
        System.out.println("收到消息的key: " + record.key());
        System.out.println("收到消息的value: " + record.value().toString());
    }
}

3、不使用默认配置，使用自定义参数，手动提交

3.1 基本配置和生产配置同2.1-2.3

3.2 消费者配置手动提交

@Configuration
@EnableKafka
public class KafkaConsumerConfig {
    @Value("${kafka.consumer.servers}")
    private String servers;
    @Value("${kafka.consumer.enable.auto.commit}")
    private boolean enableAutoCommit;
    @Value("${kafka.consumer.session.timeout}")
    private String sessionTimeout;
    @Value("${kafka.consumer.auto.commit.interval}")
    private String autoCommitInterval;
    @Value("${kafka.consumer.group.id}")
    private String groupId;
    @Value("${kafka.consumer.auto.offset.reset}")
    private String autoOffsetReset;
    @Value("${kafka.consumer.concurrency}")
    private int concurrency;

    /**
     * 基本配置
     * @return Map<String, Object>
     */
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        return propsMap;
    }

    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    public Map<String, Object> consumerConfigsAck() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        return propsMap;
    }

    public ConsumerFactory<String, String> consumerFactoryAck() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigsAck());
    }

    //@KafkaListener(topics = {"testAck"},containerFactory = "factoryAck")
    //containerFactory是当前bean
    @Bean("factoryAck")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>>
    kafkaListenerContainerFactoryAck() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory
                = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactoryAck());
        factory.setConcurrency(concurrency);
        factory.getContainerProperties().setPollTimeout(1500);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}

3.3、消费者消费手动提交确认

@Component
public class MyListenerAck {

    //containerFactory 使用自定义的factoryAck
    @KafkaListener(topics = {"testAck"},containerFactory = "factoryAck")
    public void listen(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        try {
            System.out.println("自行确认方式收到消息的key: " + record.key());
            System.out.println("自行确认方式收到消息的value: " + record.value().toString());
        } finally {
            System.out.println("消息确认！");
            ack.acknowledge();
        }
    }
}

Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

cover

Kafka服务端日志详解

Kafka开源项目指南

cover

zookeeper+kafka群集

Kafka开源项目指南

cover

Kafka服务端的各种机制实现原理

Kafka开源项目指南

所有评论(0)

查看更多评论

一点光辉

已为社区贡献1条内容