转载请注明来源：Spring Kafka 2.x 生产/消费消息的基本用法详解 - zifangsky的个人博客
jar包
compile('org.springframework.boot:spring-boot-starter')compile('org.springframework.kafka:spring-kafka')
bootstrap.properties
# 指定kafka 代理地址，可以多个spring.kafka.bootstrap-servers=127.0.0.1:9092#指定默认topic idspring.kafka.template.default-topic=topic-test#指定listener 容器中的线程数，用于提高并发量spring.kafka.listener.concurrency=3#============================provider===============#发送消息重试次数spring.kafka.producer.retries=0#每次批量发送消息的数量spring.kafka.producer.batch-size=16384spring.kafka.producer.buffer-memory=33554432# 指定消息key和消息体的编解码方式spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializerspring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer#=============== consumer =======================# 指定默认消费者group idspring.kafka.consumer.group-id=error-question-consumer-group#若设置为earliest，那么会从头开始读partitionspring.kafka.consumer.auto-offset-reset=earliestspring.kafka.consumer.enable-auto-commit=truespring.kafka.consumer.auto-commit-interval=100# 指定消息key和消息体的编解码方式spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializerspring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer#topicapp.topic.id=error-question
生产者 KafkaProducerServiceImpl.java
@Servicepublic class KafkaProducerServiceImpl implements KafkaProducerService { @Value("${app.topic.id}") private String topic; @Autowired private KafkaTemplate kafkaTemplate; @Override public void send(Object obj) { kafkaTemplate.send(topic, JSON.toJSONString(obj)); }}
消费者 KafkaReceiverServiceImpl.java
@Servicepublic class KafkaReceiverServiceImpl implements KafkaReceiverService { @Autowired private AnswerQuestionLogService answerQuestionLogService; @KafkaListener(topics = {"error-question"}) public void listen(ConsumerRecord<?, ?> record) { Optional<?> value = Optional.ofNullable(record.value()); if(value.isPresent()){ ArrayList<AnswerQuestionLog> answerDatas = JSON.parseObject((String) value.get(), new TypeReference<ArrayList<AnswerQuestionLog>>() {}); answerQuestionLogService.saveQuestion(answerDatas); } }}
