這篇文章給大家介紹Kafka怎么在Spring Boot中使用，內容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。
讓客戶滿意是我們工作的目標，不斷超越客戶的期望值來自于我們對這個行業(yè)的熱愛。我們立志把好的技術通過有效、簡單的方式提供給客戶，將通過不懈努力成為客戶在信息化領域值得信任、有價值的長期合作伙伴，公司提供的服務項目有：域名注冊、虛擬主機、營銷軟件、網站建設、張掖網站維護、網站推廣。
系統(tǒng)環(huán)境
使用遠程服務器上搭建的kafka服務
Ubuntu 16.04 LTS
kafka_2.12-0.11.0.0.tgz
zookeeper-3.5.2-alpha.tar.gz
集成過程
1.創(chuàng)建spring boot工程，添加相關依賴：
4.0.0
com.laravelshao.springboot
spring-boot-integration-kafka
0.0.1-SNAPSHOT
jar
spring-boot-integration-kafka
Demo project for Spring Boot
org.springframework.boot
spring-boot-starter-parent
2.0.0.RELEASE
UTF-8
UTF-8
1.8
org.springframework.boot
spring-boot-starter
org.springframework.kafka
spring-kafka
org.springframework.boot
spring-boot-starter-json
org.springframework.boot
spring-boot-starter-test
test
org.springframework.boot
spring-boot-maven-plugin
2.添加配置信息，這里使用yml文件
spring:
kafka:
bootstrap-servers:X.X.X.X:9092
producer:
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
consumer:
group-id: test
auto-offset-reset: earliest
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
properties:
spring:
json:
trusted:
packages: com.laravelshao.springboot.kafka
3.創(chuàng)建消息對象
public class Message {
private Integer id;
private String msg;
public Message() {
}
public Message(Integer id, String msg) {
this.id = id;
this.msg = msg;
}
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public String getMsg() {
return msg;
}
public void setMsg(String msg) {
this.msg = msg;
}
@Override
public String toString() {
return "Message{" +
"id=" + id +
", msg='" + msg + '\'' +
'}';
}
}4.創(chuàng)建生產者
package com.laravelshao.springboot.kafka;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
/**
* Created by shaoqinghua on 2018/3/23.
*/
@Component
public class Producer {
private static Logger log = LoggerFactory.getLogger(Producer.class);
@Autowired
private KafkaTemplate kafkaTemplate;
public void send(String topic, Message message) {
kafkaTemplate.send(topic, message);
log.info("Producer->topic:{}, message:{}", topic, message);
}
}5.創(chuàng)建消費者，使用@ KafkaListener注解監(jiān)聽主題
package com.laravelshao.springboot.kafka;
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;
/**
* Created by shaoqinghua on 2018/3/23.
*/
@Component
public class Consumer {
private static Logger log = LoggerFactory.getLogger(Consumer.class);
@KafkaListener(topics = "test_topic")
public void receive(ConsumerRecord consumerRecord) {
log.info("Consumer->topic:{}, value:{}", consumerRecord.topic(), consumerRecord.value());
}
}6.發(fā)送消費測試
package com.laravelshao.springboot;
import com.laravelshao.springboot.kafka.Message;
import com.laravelshao.springboot.kafka.Producer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;
@SpringBootApplication
public class IntegrationKafkaApplication {
public static void main(String[] args) throws InterruptedException {
ApplicationContext context = SpringApplication.run(IntegrationKafkaApplication.class, args);
Producer producer = context.getBean(Producer.class);
for (int i = 1; i < 10; i++) {
producer.send("test_topic", new Message(i, "test topic message " + i));
Thread.sleep(2000);
}
}
}可以依次看到發(fā)送消息，消費消息
異常問題
反序列化異常(自定義的消息對象不在kafka信任的包路徑下)？
[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] ERROR org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.719 Container exception
org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition test_topic-0 at offset 9. If needed, please seek past the record to continue consumption.
Caused by: java.lang.IllegalArgumentException: The class 'com.laravelshao.springboot.kafka.Message' is not in the trusted packages: [java.util, java.lang]. If you believe this class is safe to deserialize, please provide its name. If the serialization is only done by a trusted source, you can also enable trust all (*).
at org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper.getClassIdType(DefaultJackson2JavaTypeMapper.java:139)
at org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper.toJavaType(DefaultJackson2JavaTypeMapper.java:113)
at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:191)
at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:923)
at org.apache.kafka.clients.consumer.internals.Fetcher.access$2600(Fetcher.java:93)
at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.fetchRecords(Fetcher.java:1100)
at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.access$1200(Fetcher.java:949)
at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:570)
at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:531)
at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1103)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:667)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.lang.Thread.run(Thread.java:745)
解決方法：將當前包添加到kafka信任的包路徑下
spring:
kafka:
consumer:
properties:
spring:
json:
trusted:
packages: com.laravelshao.springboot.kafka
關于Kafka怎么在Spring Boot中使用就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。
文章題目：Kafka怎么在SpringBoot中使用
轉載來于：
http://weahome.cn/article/pigsop.html
重慶分公司
重慶分公司
