kafka（五）spring-kafka（1）集成方法
一、集成
1、pom依赖
<!--kafka--><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>版本号对应关系可以查看官网
2、配置文件
基础配置：以下是必须的配置
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer#spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer其他配置参考文档
二、使用
1、生产者
使用 KafkaTemplate发送消息。如
package com.example.service.impl;import com.alibaba.fastjson.JSON;
import com.example.dto.UserDTO;
import com.example.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;@Service("userService")
@Slf4j
public class UserServiceImpl implements UserService {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;@Value("${user.topic}")private String userTopic;@Overridepublic void sendUserMsg(UserDTO userDTO) {String msg = JSON.toJSONString(userDTO);ProducerRecord producerRecord = new ProducerRecord(userTopic,msg);kafkaTemplate.send(producerRecord);log.info("user消息发送成功");}
}
2、消费者
使用 @KafkaListener 注解。如
package com.example.listen;import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Optional;
import java.util.Properties;@Component
@Slf4j
public class SchoolConsumer {@KafkaListener(topics = "${school.topic}", groupId = "${school.group.id}")public void consumer(ConsumerRecord<?, ?> record) {try {Object message = record.value();if (message != null) {String msg = String.valueOf(message);log.info("接收到:msg={},topic:{},partition={},offset={}",msg,record.topic(),record.partition(),record.offset());}} catch (Exception e) {log.error("topic:{},is consumed error:{}", record.topic(), e.getMessage());} finally {//ack.acknowledge();}}
}