springboot集成kafka收发消息

引入pom依赖，建议使用gson解析kafka数据，使用fastjson解析有时会出现问题。

        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.8.2</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

application.properties配置

#消费端配置
kafka.consumer.bootstrap-servers=23.99.189.34:9092,23.99.189.87:9092,23.99.189.98:9092,23.99.189.105:9092,23.99.189.126:9092
kafka.consumer.group-id=group_kafka
kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.JsonSerializer
kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonSerializer


kafka.consumer.enable-auto-commit=false

kafka.consumer.session.timeout=30000
kafka.consumer.concurrency=3

#kafka.consumer.auto-offset-reset=latest
kafka.consumer.auto-offset-reset=earliest
kafka.consumer.auto.commit.interval.ms=120000
kafka.consumer.properties.request.timeout.ms=180000


#生产端配置
spring.kafka.bootstrap-servers=23.99.189.34:9092,23.99.189.87:9092,23.99.189.98:9092,23.99.189.105:9092,23.99.189.126:9092
spring.kafka.producer.retries=0
spring.kafka.producer.acks=all
kafka.producer.topic=abnormal_people
spring.kafka.producer.batch-size=16384
spring.kafka.producer.properties.linger.ms=0
spring.kafka.producer.buffer-memory = 33554432
spring.kafka.producer.key-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

消费者配置

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
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 org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class kafkaConsumerConfig {
    @Value("${kafka.consumer.bootstrap-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.ms}")
    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;


    @Bean("kafkaListenerContainnerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(concurrency);
       factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.getContainerProperties().setPollTimeout(1500);
        return factory;
    }

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

    }


    private 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.HEARTBEAT_INTERVAL_MS_CONFIG,100);
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        
//kafka有jaas验证配置
propsMap.put("sasl.jaas.config","org.apache.kafka.common.security.plain.PlainLoginModule required username=\"cigc_dsjj\" password=\"1qaz2wsx\";");
//kafka有jaas验证配置       
propsMap.put("sasl.mechanism","PLAIN");
//kafka有jaas验证配置
        propsMap.put("security.protocol", "SASL_PLAINTEXT");
        return propsMap;
    }

}

消费者

package com.abc.renchewu.kafka;
import com.abc.renchewu.kafka.entity.CarEntity;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import java.util.Optional;
/**
 * kafka监听消费
 * @author ws
 *
 */
@Component
@Slf4j
public class KafkaConsumerListener {

    @KafkaListener(topics = "CIGC_DSJJ_COV19_AREA", containerFactory = "kafkaListenerContainnerContainerFactory")
    public void listenOnMessageManual(ConsumerRecord<String, String> record,
                                      Acknowledgment ack) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Gson gson = new Gson();
            log.info("接收车辆消息:{}", kafkaMessage.get());
            CarEntity carEntity = gson.fromJson(kafkaMessage.get().toString(), CarEntity.class);
           //手动签收
            ack.acknowledge();
        }
    }
}

生产者

package com.dcqc.kafkatranscode.utils;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;


@Component
@Slf4j
public class kafkaUtils {
    @Value("${kafka.producer.topic}")
    private String topicName;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 发送kafka消息
     */
    public void send(String jsonMessage) {
        kafkaTemplate.send(topicName, jsonMessage);
    }

}