一．創(chuàng)建Spring boot項(xiàng)目，添加如下依賴

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka --> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients --> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.41</version> </dependency>

二．配置文件

server.port=4400

#kafka配置#============== kafka ===================# 指定kafka 代理地址，可以多個(gè)spring.kafka.bootstrap-servers=192.168.102.88:9092# 指定默認(rèn)消費(fèi)者group idspring.kafka.consumer.group-id=jkafka.demo#earliest 當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開(kāi)始消費(fèi)；無(wú)提交的offset時(shí)，從頭開(kāi)始消費(fèi)#latest 當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開(kāi)始消費(fèi)；無(wú)提交的offset時(shí)，消費(fèi)新產(chǎn)生的該分區(qū)下的數(shù)據(jù)#none topic各分區(qū)都存在已提交的offset時(shí)，從offset后開(kāi)始消費(fèi)；只要有一個(gè)分區(qū)不存在已提交的offset，則拋出異常spring.kafka.consumer.auto-offset-reset=latestspring.kafka.consumer.enable-auto-commit=falsespring.kafka.consumer.auto-commit-interval=100# 指定消費(fèi)者消息key和消息體的編解碼方式spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

三．編輯消息實(shí)體

@Datapublic class Message implements Serializable{ /** * */ private static final long serialVersionUID = 2522280475099635810L; //消息ID private String id; //消息內(nèi)容 private String msg; // 消息發(fā)送時(shí)間 private Date sendTime;}

四．消息發(fā)送類

@Componentpublic class KfkaProducer { private static Logger logger = LoggerFactory.getLogger(KfkaProducer.class); @Autowired private KafkaTemplate<String, String> kafkaTemplate; public void send(String topic,Message message) { try { logger.info('正在發(fā)送消息...'); kafkaTemplate.send(topic,JSON.toJSONString(message)); logger.info('發(fā)送消息成功 ----->>>>> message = {}', JSON.toJSONString(message)); } catch (Exception e) { e.getMessage(); } }}

五．發(fā)現(xiàn)監(jiān)聽(tīng)接收類

@Componentpublic class KfkaListener { private static Logger logger = LoggerFactory.getLogger(KfkaListener.class); @KafkaListener(topics = {'hello'}) public void listen(ConsumerRecord<?, ?> record) { Optional<?> kafkaMessage = Optional.ofNullable(record.value()); if (kafkaMessage.isPresent()) { Object message = kafkaMessage.get(); logger.info('接收消息------------ record =' + record); logger.info('接收消息----------- message =' + message); } }}

六．定時(shí)發(fā)送信息測(cè)試類

@EnableScheduling@Componentpublic class PublisherController { private static final Logger log = LoggerFactory.getLogger(PublisherController.class); @Autowired private KfkaProducer kfkaProducer; @Scheduled(fixedRate = 5000) public void pubMsg() { Message msg=new Message(); msg.setId(UUID.randomUUID().toString()); msg.setMsg('發(fā)送這條消息給你，你好?。。。。。。?); msg.setSendTime(new Date()); kfkaProducer.send('hello', msg);; log.info('Publisher sendes Topic... '); }}

七．測(cè)試結(jié)果

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。