Пример кода Spring Boot интегрируется с Kafka

新建 Spring Boot 项目

这里使用 Идея Intellij

添加 Кафка 集成 Maven

 <? xml version = "1.0" Encoding = "utf-8"?> <project xmlns = "http://maven.apache.org/pom/4.0.0" xmlns: xsi = "http://www.w3.org/2001/xmlschema-instance" xsi: schemalocation = "http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <dolidyversion> 4.0.0 </modelversion> <groupid> com.example </GroupId> <ArtifactId> demoId> </artifActid> demoId> </artifactid> </artifactid> </artifactid>. <serse> 0.0.1-snapshot </version> <Cackaging> jar </packaging> <mame> demo </name> <Описание> Демо-проект для Spring Boot </description> <parent> <groupd> org.springframework.boot </GroupId> <ratifactid> spring-starter-parent-prentshid </artifactid> <serse> 1. Поиск родителя из репозитория-> </parent> <properties> <project.build.sourceencoding> utf-8 </project.build.sourceencoding> <project.Report.OutputEncoding> utf-8 </project.Reporting.OutputEncoding> <java.version> 1.8 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> <dependency> <groupId> org.springframework.boot </GroupId> <ArtifactId> Spring-boot-starter-test </artifactid> <cracpe> test </scope> </dependency> </depertive> <buld> <blicins> <blicin> <groupid> org.spramework.boot </GroupId> <ratifactid> spring-boot-maven> promaven> plugvin-pramebrame.bout </GroupId> <pracin> promaven> plug. </plugin> </plugins> </build> </project>

项目中 Application.properties 添加

Spring.kafka.bootstrap-servers = VM208: 9092, VM: 9092, VM50: 9092SPRING.KAFKA.CONSUMER.AUTO-OFFSET-RESET = ANDALTSPRING.KAFKA.CONSUMER.GROUP-ID = L. ocal_testspring.kafka.consumer.key-derializer = org.apache.kafka.common.serialization.stringdeserializerspring.kafka.consumer.value-deseri alizer = 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.stringserializerspring.kafka.producer.acks = 1

新建 KAFKACONSUMER 消费类

пакет com.example.demo.consumer; import org.apache.kafka.clients.consumer.consumerRecord; import org.slf4j.logger; импорт org.slf4j.loggerfactory; импорт org.springframework.kafka.annotation.kafkalister; org.springframework.stereotyle.component; @componentpublic class kafkaconsumer {private logger logger = loggerfactory.getlogger (this.getClass ()); @Kafkalistener (themics = {"test"}) public void Listen (ConmerceerRecord <?,?> Record) {System.out.printf ("Offset =%d, Key =%s, значение =%s/n", record.Offset (), record.key (), record.value ()); }}

启动 Spring-Boot 程序, 在 kafka 集群, 模拟发送 Тема, 检验接收

复制代码代码如下: bin/kafka-console-producer.sh-брокер-спирт VM208: 9092, VM210: 9092, VM50: 9092-Тест-Topic

编写 Производитель 代码

пакет com.example.demo.producer; import org.apache.kafka.clients.producer.producerRecord; import org.springframework.beans.factory.antation.autowired; импорт org.springframework.kafka.core.kafkatemplate; org.springframework.stereotyle.component; @componentpublic class kafkaproducer {@autowired private kafkatemplate kafkatemplate; String topic = "test"; public void sendmessage (String Key, String Data) {kafkatemplate.send (new ProducterRecord (тема, ключ, данные)); }}

建立一个 Restful 模拟发送 (//http://localhost:8080/kafka/send.do?key=2&data=allen-test-message)

 Пакет com.example.demo.controller; import com.example.demo.producer.kafkaproducer; import org.springframework.beans.factory.annotation.autowired; импорт org.springframe.web.bindtation.requestmaping; org.springframework.web.bind.annotation.requestmethod; import org.springframework.web.bind.annotation.requestparam; импорт org.springframework.web.bindtation.restcontroller; @restcontrollerpublic Class Class Class. Kafkaproducer; @Requestmapping (value = "/Kafka/send.do", method = requestMethod.get) public String sendmessage (@RequestParam (value = "key") String Key, @RequestParam (value = "data") String Data) {kafkaproducer.sendmessage (key, data); вернуть "успех"; }}

可以发现 Spring-Kafka 大大减少了代码工作量.

官方文档: https://docs.spring.io/spring-kafka/docs/1.2.2.reelease/reference/html/

以上就是本文的全部内容 ， 希望对大家的学习有所帮助 也希望大家多多支持武林网。