新建spring boot項目
這里使用intellij idea
|
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:9092spring.kafka.consumer.auto-offset-reset=latestspring.kafka.consumer.group-id=local_testspring.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.stringserializerspring.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;@componentpublic 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;@componentpublic 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;@restcontrollerpublic 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
