二十六、Springboot整合kafka
(一)新增依賴
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
(二)配置檔案
spring: kafka: bootstrap-servers: 192.168.10.130:9092,192.168.10.130:9093,192.168.10.130:9094 #生產者 “值” 序列化(自定義ObjectSerializer類) producer: value-serializer: org.pc.serializer.ObjectSerializer consumer: group-id: cluster-group #消費者 “值” 反序列化(自定義ObjectDeserializer類) value-deserializer: org.pc.deserializer.ObjectDeserializer
(三)註冊KafkaTemplate Springboot自動裝載了KafkaTemplate,無需手動註冊
(四)使用Kafka生產者客戶端生產訊息
@RestController public class KafkaController { private KafkaTemplate<String, Object> kafkaTemplate; @Autowired public KafkaController(KafkaTemplate<String, Object> kafkaTemplate) { this.kafkaTemplate = kafkaTemplate; } @GetMapping("/message/send") public String sendMessage(@RequestParam String message){ //向kafka叢集中的"cluster-topic"傳送訊息 kafkaTemplate.send("cluster-topic", 0, "message", message); return message; } }
(五)使用Kafka消費者客戶端消費訊息
/**
* 消費者監聽器
* 作用:只要加上@KafkaListener,設定好topics,就可以訂閱該主題產生的訊息(相當於消費者客戶端)
*/
@Component
public class ConsumerListener {
@KafkaListener(topics = "cluster-topic")
public void consume(String message){
System.out.println(message);
}
}
(六)出現問題及其解決 1、當傳送Object型別訊息時,報錯"Can’t convert value of class org.pc.entity.User to class org.apache.kafka.common.serialization.StringSerializer specified in value.serializer"?
原因解析:在生產訊息時,Kafka的Java客戶端預設使用的是StringSerializer序列化類,但是他只能序列化String型別物件,要是其他型別物件,就會無法序列化。同理,在消費訊息時,需要反序列化,也會遇到這個問題。
解決辦法:自定義Kafka“生產者客戶端的序列化類”和“消費者客戶端的反序列化類”,目的實現對Object物件的序列化與反序列化。
(1)生產者客戶端 “值” 的序列化:
/**
* Object 序列化
* @author 鹹魚
* @date 2018/10/14 8:43
*/
public class ObjectSerializer implements Serializer<Object> {
@Override
public void configure(Map configs, boolean isKey) {
}
@Override
public byte[] serialize(String topic, Object data) {
byte[] dataArray = null;
try {
//1、建立OutputStream物件
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
//2、建立OutputStream的包裝物件ObjectOutputStream,PS:物件將寫到OutputStream流中
ObjectOutputStream objectOutputStream = new ObjectOutputStream(outputStream);
//3、將物件寫到OutputStream流中
objectOutputStream.writeObject(data);
//4、將OutputStream流轉換成位元組陣列
dataArray = outputStream.toByteArray();
} catch (IOException e) {
e.printStackTrace();
}
return dataArray;
}
@Override
public void close() {
}
}
(2)消費者客戶端 “值” 的反序列化:
/**
* Object 反序列化
* @author 鹹魚
* @date 2018/10/14 9:13
*/
public class ObjectDeserializer implements Deserializer<Object> {
@Override
public void configure(Map configs, boolean isKey) {
}
@Override
public Object deserialize(String topic, byte[] data) {
Object object = null;
try {
ByteArrayInputStream inputStream = new ByteArrayInputStream(data);
ObjectInputStream objectInputStream = new ObjectInputStream(inputStream);
object = objectInputStream.readObject();
} catch (Exception e) {
throw new RuntimeException(e);
}
return object;
}
@Override
public void close() {
}
}
(3)修改配置檔案
spring:
kafka:
bootstrap-servers: 192.168.10.130:9092,192.168.10.130:9093,192.168.10.130:9094
#生產者 “值” 序列化
producer:
value-serializer: org.pc.serializer.ObjectSerializer
consumer:
group-id: cluster-group
#消費者 “值” 反序列化
value-deserializer: org.pc.deserializer.ObjectDeserializer
(4)測試 生產訊息:
@PostMapping("/user")
public User saveUser(@RequestBody User user){
kafkaTemplate.send("object-topic", 0, "message", user);
return user;
}
消費訊息:
@KafkaListener(topics = "object-topic")
public void consume(Object object){
System.out.println(object);
}
2、報錯如下:javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=gx-test-20170629?
原因解析:如果使用了ConcurrentMessageListenerContainer 的實現,並且配置了併發度大於1,同時配置了kafka的 client.id屬性則會出現上述問題,而當你配置為1的時候不會出現上述log。
解決辦法:不配置client.id這一項,kakfa中會預設為多個執行緒生成id