Springboot集成kafka,springboot+kafka

引言

卡夫卡近几年更新非常快，也可以看出卡夫卡在企业中是用的频率越来越高，在跳羚中集成卡夫卡还是比较简单的，但是应该注意使用的版本和卡夫卡中基本配置，这个地方需要信心，防止进入坑中。

　　版本对应地址：https://春天。io/projects/spring-卡夫卡

基本环境

跳靴版本2.1.4

　　卡夫卡版本2.2.0

　　jdk 1.8

代码编写

1、基本引用pom

?可扩展标记语言版本=1.0 编码=UTF八号？项目xmlns= http://maven。阿帕奇。org/POM/4。0 .0 xmlns : xsi= http://www。w3。org/2001/XML schema-instance xsi :架构位置= http://maven。阿帕奇。org/POM/4。0 .0 http://maven。阿帕奇。org/xsd/maven-4。0 .0 .“xsd”模型版本4 .-从资源库中查找父级-/parentgroupid com。示例/groupidatifactiddemo/artifact id version 0。0 .1-快照/版本名称kafkademo/名称描述Spring Boot的演示项目/description属性Java。版本1.8/Java。version/properties dependencies dependencygroupidorg。spring框架。boot/groupidatifactidspring-boot-starter-web/artifactId/dependencygroupid MySQL/groupidatifactidmysql-connector-Java/artifactIdscoperuntime/

　　<plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build> </project>

2、基本配置

spring.kafka.bootstrap-servers=2.1.1.1:9092spring.kafka.consumer.group-id=test-consumer-groupspring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializerspring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializerspring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializerspring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer #logging.level.root=debug

3、实体类

package com.example.demo.model; import java.util.Date; public class Messages { private Long id; private String msg; private Date sendTime; public Long getId() { return id; } public void setId(Long id) { this.id = id; } public String getMsg() { return msg; } public void setMsg(String msg) { this.msg = msg; } public Date getSendTime() { return sendTime; } public void setSendTime(Date sendTime) { this.sendTime = sendTime; }}

4、生产者端

package com.example.demo.service; import com.example.demo.model.Messages;import com.google.gson.Gson;import com.google.gson.GsonBuilder;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.kafka.core.KafkaTemplate;import org.springframework.kafka.support.SendResult;import org.springframework.stereotype.Service;import org.springframework.util.concurrent.ListenableFuture; import java.util.Date;import java.util.UUID; @Servicepublic class KafkaSender { @Autowired private KafkaTemplate<String, String> kafkaTemplate; private Gson gson = new GsonBuilder().create(); public void send() { Messages message = new Messages(); message.setId(System.currentTimeMillis()); message.setMsg("123"); message.setSendTime(new Date()); ListenableFuture<SendResult<String, String>> test0 = kafkaTemplate.send("newtopic", gson.toJson(message)); }}

5、消费者

package com.example.demo.service; import org.apache.kafka.clients.consumer.ConsumerRecord;import org.springframework.kafka.annotation.KafkaListener;import org.springframework.stereotype.Service; import java.util.Optional; @Servicepublic class KafkaReceiver { @KafkaListener(topics = {"newtopic"}) public void listen(ConsumerRecord<?, ?> record) { Optional<?> kafkaMessage = Optional.ofNullable(record.value()); if (kafkaMessage.isPresent()) { Object message = kafkaMessage.get(); System.out.println("record =" + record); System.out.println("message =" + message); } } }

6、测试

在启动方法中模拟消息生产者，向kafka中发送消息

package com.example.demo; import com.example.demo.service.KafkaSender;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.ConfigurableApplicationContext; @SpringBootApplicationpublic class KafkademoApplication { public static void main(String[] args) {ConfigurableApplicationContext context = SpringApplication.run(KafkademoApplication.class, args); KafkaSender sender = context.getBean(KafkaSender.class);for (int i = 0; i <1000; i++) {sender.send(); try {Thread.sleep(300);} catch (InterruptedException e) {e.printStackTrace();}} } }

效果展示

　　命令行直接消费消息

遇到的问题

生产端连接kafka超时

　　at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:119)

　　解决方案：

　　修改kafka中的server.properties中的下面配置，将原来的默认配置替换成下面ip+端口的形式，重启kafka

　　到此这篇关于Springboot 2.x集成kafka 2.2.0的示例代码的文章就介绍到这了,更多相关Springboot集成kafka内容请搜索盛行IT以前的文章或继续浏览下面的相关文章希望大家以后多多支持盛行IT！

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。