引入pom依赖

        <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.JsonSerializerkafka.consumer.enable-auto-commit=falsekafka.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;@Autowiredprivate KafkaTemplate kafkaTemplate;/*** 发送kafka消息*/public void send(String jsonMessage) {kafkaTemplate.send(topicName, jsonMessage);}}