springboot對kafka的client很好的實現了集成，使用非常方便，本文也實現了一個在springboot中實現操作kafka的demo。
1.pom配置
只需要在dependencies中增加 spring-kafka的配置即可。完整效果如下：
|
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
|
<parent> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-parent</artifactid> <version>1.5.4.release</version> </parent> <properties> <java.version>1.8</java.version> <spring-kafka.version>1.2.2.release</spring-kafka.version> <project.build.sourceencoding>utf-8</project.build.sourceencoding> </properties> <dependencies> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-web</artifactid> </dependency> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter</artifactid> </dependency> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-test</artifactid> <scope>test</scope> </dependency> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-aop</artifactid> </dependency> <!-- spring-kafka --> <dependency> <groupid>org.springframework.kafka</groupid> <artifactid>spring-kafka</artifactid> <version>${spring-kafka.version}</version> </dependency> <dependency> <groupid>org.springframework.kafka</groupid> <artifactid>spring-kafka-test</artifactid> <version>${spring-kafka.version}</version> <scope>test</scope> </dependency> </dependencies> |
2.生產者
參數配置類，其參數卸載yml文件中，通過@value注入
|
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
|
package com.dhb.kafka.producer;import org.apache.kafka.clients.producer.producerconfig;import org.apache.kafka.common.serialization.stringserializer;import org.springframework.beans.factory.annotation.value;import org.springframework.context.annotation.bean;import org.springframework.context.annotation.configuration;import org.springframework.kafka.core.defaultkafkaproducerfactory;import org.springframework.kafka.core.kafkatemplate;import org.springframework.kafka.core.producerfactory;import java.util.hashmap;import java.util.map;@configurationpublic class senderconfig { @value("${kafka.bootstrap-servers}") private string bootstrapservers; @bean public map<string,object> producerconfigs() { map<string,object> props = new hashmap<>(); props.put(producerconfig.bootstrap_servers_config,this.bootstrapservers); props.put(producerconfig.key_serializer_class_config, stringserializer.class); props.put(producerconfig.value_serializer_class_config,stringserializer.class); props.put(producerconfig.acks_config,"0"); return props; } @bean public producerfactory<string,string> producerfactory() { return new defaultkafkaproducerfactory<>(producerconfigs()); } @bean public kafkatemplate<string,string> kafkatemplate() { return new kafkatemplate<string, string>(producerfactory()); } @bean public sender sender() { return new sender(); }} |
消息發送類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
package com.dhb.kafka.producer;import lombok.extern.slf4j.slf4j;import org.springframework.beans.factory.annotation.autowired;import org.springframework.kafka.core.kafkatemplate;@slf4jpublic class sender { @autowired private kafkatemplate<string,string> kafkatemplate; public void send(string topic,string payload) { log.info("sending payload='{}' to topic='{}'",payload,topic); this.kafkatemplate.send(topic,payload); }} |
3.消費者
參數配置類
|
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
|
package com.dhb.kafka.consumer;import org.apache.kafka.clients.consumer.consumerconfig;import org.apache.kafka.common.serialization.stringdeserializer;import org.springframework.beans.factory.annotation.value;import org.springframework.context.annotation.bean;import org.springframework.context.annotation.configuration;import org.springframework.kafka.annotation.enablekafka;import org.springframework.kafka.config.concurrentkafkalistenercontainerfactory;import org.springframework.kafka.core.consumerfactory;import org.springframework.kafka.core.defaultkafkaconsumerfactory;import java.util.hashmap;import java.util.map;@configuration@enablekafkapublic class receiverconfig { @value("${kafka.bootstrap-servers}") private string bootstrapservers; public map<string,object> consumerconfigs() { map<string,object> props = new hashmap<>(); props.put(consumerconfig.bootstrap_servers_config,bootstrapservers); props.put(consumerconfig.key_deserializer_class_config, stringdeserializer.class); props.put(consumerconfig.value_deserializer_class_config,stringdeserializer.class); props.put(consumerconfig.group_id_config,"helloword"); return props; } @bean public consumerfactory<string,string> consumerfactory() { return new defaultkafkaconsumerfactory<>(consumerconfigs()); } @bean public concurrentkafkalistenercontainerfactory<string,string> kafkalistenercontainerfactory() { concurrentkafkalistenercontainerfactory<string,string> factory = new concurrentkafkalistenercontainerfactory<>(); factory.setconsumerfactory(consumerfactory()); return factory; } @bean public receiver receiver() { return new receiver(); }} |
消息接受類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
package com.dhb.kafka.consumer;import lombok.extern.slf4j.slf4j;import org.springframework.kafka.annotation.kafkalistener;import java.util.concurrent.countdownlatch;@slf4jpublic class receiver { private countdownlatch latch = new countdownlatch(1); public countdownlatch getlatch() { return latch; } @kafkalistener(topics = "${kafka.topic.helloworld}") public void receive(string payload) { log.info("received payload='{}'",payload); latch.countdown(); }} |
3.web測試類
定義了一個基于http的web測試接口
|
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
|
package com.dhb.kafka.web;import com.dhb.kafka.producer.sender;import lombok.extern.slf4j.slf4j;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.restcontroller;import javax.servlet.http.httpservletrequest;import javax.servlet.http.httpservletresponse;import java.io.ioexception;@restcontroller@slf4jpublic class kafkaproducer { @autowired sender sender; @requestmapping(value = "/sender.action", method = requestmethod.post) public void exec(httpservletrequest request, httpservletresponse response,string data) throws ioexception{ this.sender.send("testtopic",data); response.setcharacterencoding("utf-8"); response.setcontenttype("text/json"); response.getwriter().write("success"); response.getwriter().flush(); response.getwriter().close(); }} |
4.啟動類及配置
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
package com.dhb.kafka;import org.springframework.boot.springapplication;import org.springframework.boot.autoconfigure.springbootapplication;@springbootapplicationpublic class kafkaapplication { public static void main(string[] args) { springapplication.run(kafkaapplication.class,args); }} |
application.yml
|
1
2
3
4
|
kafka: bootstrap-servers: 192.168.162.239:9092 topic: helloworld: testtopic |
程序結構：
包結構
5.讀寫測試
通過執行kafkaapplication的main方法啟動程序。然后打開postman進行測試：
運行后返回success
生產者日志：
消費者日志：
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
原文鏈接：http://www.jianshu.com/p/3dcb64e49ac5?utm_source=tuicool&utm_medium=referral
