Exemple de code de démarrage de Spring Intégration de Kafka

新建 Boot de printemps 项目

这里使用 Idem Intellij

添加 Kafka 集成 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: ScheMalation = "http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <mackversion> 4.0.0 </ Modelversion> <proupid> com.example </ GroupId> <ArtifActid> Demo </ ArtiFact> Com.example </ GroupId> <ArtifActid> Demio </ Artiford> Com.example </ GroupId> <ArtifActid> Demo </ Artiford> COMM <version> 0.0.1-snapshot </ version> <packaging> jar </packaging> <name> démo </name> <description> Projet de démonstration pour Spring Boot </ Description> <parent> <proupId> org.springframework.boot </prouprid> <Artifactid> printemps-boot-starter-parent </ptetifactid> <version> 1.5.8. <! - Recherche parent du référentiel -> </parent> <properties> <project.build.sourceencoding> utf-8 </project.build.sourceencoding> <project.reportting.outputending> utf-8 </project.report.outputerSice> <java.version> 1.8 </java.version> <dependency> <proupId> org.springframework.boot </proupId> <ArtifActid> printemp-boot-starter-web </ artifactid> </peedency> <dependency> <proupId> org.springframework.kafka </proupency> <pritifactid> printemps-kafka </letifactid> </dedency> <petency> <GroupId> org.springframework.boot </prôdId> <ArtifactId> Spring-Boot-starter-test </ artifactid> <ccope> test </cope> </dependency> </dpendance> <uild> </grouns> <Glugin> <proupId> org.springframework.boot </prounid> <ArtefactId> Spring-Boot-Maven-Plugin </ ArfactId> </Glugin> </Glugins> </Duild> </randing>

项目中 application.properties 添加

Spring.Kafka.bootstrap-Servers = VM208: 9092, VM: 9092, VM50: 9092Spring.Kafka.Consumer.Auto-Offset-Resset = Deromadspring.Kafka.Consumer.Group-ID = L ocal_testspring.kafka.consumer.key-deserializer = 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 消费类

package com.example.demo.consumer; import org.apache.kafka.clients.consumer.consumerCord; import org.slf4j.logger; import org.slf4j.loggerfactory; import org.springframework.kafka.annotation.kafkalistener; Import org.springframework.sterreotype.Component; @componentpublic classe kafkaconsumer {private logger logger = loggerfactory.getLogger (this.getClass ()); @KafKAListener (sujets = {"test"}) public void écouter (ConsumerCord <?,?> Record) {System.out.printf ("offset =% d, key =% s, value =% s / n", enregistre.offset (), record.key (), enregistre.Value ()); }}

启动 Spring-Boot 程序, 在 Kafka 集群, 模拟发送 Topic, 检验接收

°

编写 Producteur 代码

package com.example.demo.producer; import org.apache.kafka.clients.producer.producercord; import org.springframework.beans.factory.annotation.autowired; import org.springframework.kafka.core.kafkatemplate; import; org.springframework.sterreotype.Component; @componentpublic classe kafkaproducer {@autowired private kafkatemplate kafkatemplate; String topic = "test"; public void SendMessage (String Key, String Data) {KafKatemplate.Send (new ProduceRecord (thème, key, data)); }}

建立一个 RESTFUL 模拟发送 (//http://localhost:8080/kafka/send.do?key=2&data=ALLEN-TEST-MESSAGE)

 package com.example.demo.controller; import com.example.demo.producer.kafkaproducer; import org.springframework.beans.factory.annotation.autowired; import org.springframework.web.bind.annotation.requestmapping; org.springframework.web.bind.annotation.requestMethod; import org.springframework.web.bind.annotation.requestParam; import org.springframework.web.bind.annotation.restController; @RestConTrollerPublic Class ProduceController {@Autowired privily privil Kafkaproducer; @RequestMapping (value = "/kafka/send.do", méthode = requestMethod.get) public chaîne SendMessage (@RequestParam (value = "key") string key, @RequestParam (value = "data") String data) {kafkaproducer.sendMessage (key, data); retourner "Success"; }}

可以发现 printemps-kafka 大大减少了代码工作量.

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

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