新建spring boot项目

这里使用intellij idea

添加kafka集成maven

? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 <?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"> <modelversion>4.0.0</modelversion> <groupid>com.example</groupid> <artifactid>demo</artifactid> <version>0.0.1-snapshot</version> <packaging>jar</packaging> <name>demo</name> <description>demo project for spring boot</description> <parent> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-parent</artifactid> <version>1.5.8.release</version> <relativepath/> <!-- lookup parent from repository --> </parent> <properties> <project.build.sourceencoding>utf-8</project.build.sourceencoding> <project.reporting.outputencoding>utf-8</project.reporting.outputencoding> <java.version>1.8</java.version> </properties> <dependencies> <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> <scope>test</scope> </dependency> </dependencies> <build> <plugins> <plugin> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-maven-plugin</artifactid> </plugin> </plugins> </build> </project>

项目中application.properties 添加

? 1 2 3 4 5 6 7 8 spring.kafka.bootstrap-servers=vm208:9092,vm:9092,vm50:9092 spring.kafka.consumer.auto-offset-reset=latest spring.kafka.consumer.group-id=local_test spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.stringdeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.stringdeserializer spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.stringserializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.stringserializer spring.kafka.producer.acks=1

新建kafkaconsumer消费类

? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 package com.example.demo.consumer; import org.apache.kafka.clients.consumer.consumerrecord; import org.slf4j.logger; import org.slf4j.loggerfactory; import org.springframework.kafka.annotation.kafkalistener; import org.springframework.stereotype.component; @component public class kafkaconsumer { private logger logger = loggerfactory.getlogger(this.getclass()); @kafkalistener(topics = {"test"}) public void listen(consumerrecord<?, ?> record) { system.out.printf("offset = %d,key =%s,value=%s\n", record.offset(), record.key(), record.value()); } }

启动spring-boot程序,在kafka集群,模拟发送topic,检验接收

复制代码 代码如下: bin/kafka-console-producer.sh --broker-list vm208:9092,vm210:9092,vm50:9092 --topic test

编写producer代码

? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 package com.example.demo.producer; import org.apache.kafka.clients.producer.producerrecord; import org.springframework.beans.factory.annotation.autowired; import org.springframework.kafka.core.kafkatemplate; import org.springframework.stereotype.component; @component public class kafkaproducer { @autowired private kafkatemplate kafkatemplate; string topic="test"; public void sendmessage(string key,string data){ kafkatemplate.send(new producerrecord(topic,key,data)); } }

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

? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 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; import org.springframework.web.bind.annotation.requestmethod; import org.springframework.web.bind.annotation.requestparam; import org.springframework.web.bind.annotation.restcontroller; @restcontroller public class producercontroller { @autowired private kafkaproducer 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); return "sucess"; } }

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

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

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

原文链接：https://blog.csdn.net/baifanwudi/article/details/78282620?fps=1&locationNum=10