怎么使用Spring Boot Kafka

本篇內(nèi)容介紹了“怎么使用Spring Boot  Kafka”的有關(guān)知識(shí)，在實(shí)際案例的操作過(guò)程中，不少人都會(huì)遇到這樣的困境，接下來(lái)就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

Kafka集群安裝、配置和啟動(dòng)

Kafka需要依賴zookeeper，并且自身集成了zookeeper，zookeeper至少需要3個(gè)節(jié)點(diǎn)保證集群高可用，下面是在單機(jī)linux下創(chuàng)建kafka3個(gè)節(jié)點(diǎn)偽集群模式。

1、下載包

下載地址：http://kafka.apache.org/downloads

2、解壓包

tar -zxvf kafka_2.11-1.0.0.tgz\mv kafka_2.11-1.0.0 kafka1\mv kafka_2.11-1.0.0 kafka2\mv kafka_2.11-1.0.0 kafka3

3、創(chuàng)建ZK集群

修改ZK配置文件：kafka1-3/config/zookeeper.properties分別修改對(duì)應(yīng)的參數(shù)。

dataDir=/usr/local/kafka/zookeeper1 dataLogDir=/usr/local/kafka/zookeeper/log clientPort=2181 maxClientCnxns=0 tickTime=2000 initLimit=100 syncLimit=5 server.1=127.0.0.1:2888:3888 server.2=127.0.0.1:4888:5888 server.3=127.0.0.1:6888:7888

/usr/local/kafka/zookeeper1-3目錄下分別創(chuàng)建myid文件，內(nèi)容對(duì)應(yīng)1~3

啟動(dòng)ZK，分別進(jìn)行Kafka1-3目錄:

bin/zookeeper-server-start.sh config/zookeeper.properties &

啟動(dòng)報(bào)文件失敗，需要手動(dòng)創(chuàng)建文件目錄并賦予對(duì)應(yīng)的權(quán)限。

4、創(chuàng)建Kafka集群

配置文件：kafka1-3/config/server.properties分別修改對(duì)應(yīng)的參數(shù)。

broker.id=1  zookeeper.connect=localhost:2181,localhost:2182,localhost:2183 listeners=PLAINTEXT://192.168.12.11:9091  log.dirs=/tmp/kafka-logs-1

啟動(dòng)Kafka，分別進(jìn)行Kafka1-3目錄:

bin/kafka-server-start.sh config/server.properties &

啟動(dòng)報(bào)文件失敗，需要手動(dòng)創(chuàng)建文件目錄并賦予對(duì)應(yīng)的權(quán)限。

5、集群測(cè)試

在kafka1上面發(fā)送消息：

bin/kafka-console-producer.sh --broker-list localhost:9091 --topic test

在kafka2、kafka3消費(fèi)消息：

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic

Spring Boot 集成 Kafka 實(shí)戰(zhàn)

1、添加spring-kafka依賴

<spring-kafka.version>2.1.0.RELEASE</spring-kafka.version>  <!-- spring-kafka--> <dependency>     <groupId>org.springframework.kafka</groupId>     <artifactId>spring-kafka</artifactId>     <version>${spring-kafka.version}</version> </dependency>

2、添加Spring Boot的自動(dòng)配置

自動(dòng)配置類：

• org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration

配置屬性類：

• org.springframework.boot.autoconfigure.kafka.KafkaProperties

Spring:   kafka:     bootstrap-servers:       - 192.168.101.137:9091       - 192.168.101.137:9092       - 192.168.101.137:9093     producer:       retries: 0       batch-size: 16384       buffer-memory: 33554432       key-serializer: org.apache.kafka.common.serialization.StringSerializer       value-serializer: org.apache.kafka.common.serialization.StringSerializer     consumer:       group-id: foo       auto-offset-reset: earliest       enable-auto-commit: true       auto-commit-interval: 100       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

3、發(fā)送消息

@Autowired private KafkaTemplate kafkaTemplate;  @GetMapping("/send") public Object send(String msg) {     kafkaTemplate.send("test", "name", msg);     return "send ok"; }

4、接收消息

在任何bean里面，添加@KafkaListener，支持消息接收。

@KafkaListener(topics = "test") public void processMessage(String content) {     logger.info("收到消息, topic:test, msg:{}", content); }

“怎么使用Spring Boot  Kafka”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！