建站
咨詢
售后
建站咨詢
新聞中心
輕松上手 Spring Boot & Kafka 實戰(zhàn)！
作者：Java領(lǐng)域佼佼者 2020-04-23 15:59:04
前端
Kafka Kafka需要依賴zookeeper，并且自身集成了zookeeper，zookeeper至少需要3個節(jié)點保證集群高可用，下面是在單機linux下創(chuàng)建kafka3個節(jié)點偽集群模式。
創(chuàng)新互聯(lián)建站服務(wù)項目包括博樂網(wǎng)站建設(shè)、博樂網(wǎng)站制作、博樂網(wǎng)頁制作以及博樂網(wǎng)絡(luò)營銷策劃等。多年來，我們專注于互聯(lián)網(wǎng)行業(yè)，利用自身積累的技術(shù)優(yōu)勢、行業(yè)經(jīng)驗、深度合作伙伴關(guān)系等，向廣大中小型企業(yè)、政府機構(gòu)等提供互聯(lián)網(wǎng)行業(yè)的解決方案，博樂網(wǎng)站推廣取得了明顯的社會效益與經(jīng)濟效益。目前，我們服務(wù)的客戶以成都為中心已經(jīng)輻射到博樂省份的部分城市，未來相信會繼續(xù)擴大服務(wù)區(qū)域并繼續(xù)獲得客戶的支持與信任！
Kafka集群安裝、配置和啟動
Kafka需要依賴zookeeper，并且自身集成了zookeeper，zookeeper至少需要3個節(jié)點保證集群高可用，下面是在單機linux下創(chuàng)建kafka3個節(jié)點偽集群模式。
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分別修改對應(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)容對應(yīng)1~3
啟動ZK，分別進行Kafka1-3目錄:
- bin/zookeeper-server-start.sh config/zookeeper.properties &
啟動報文件失敗，需要手動創(chuàng)建文件目錄并賦予對應(yīng)的權(quán)限。
4、創(chuàng)建Kafka集群
配置文件：kafka1-3/config/server.properties分別修改對應(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
啟動Kafka，分別進行Kafka1-3目錄:
- bin/kafka-server-start.sh config/server.properties &
啟動報文件失敗，需要手動創(chuàng)建文件目錄并賦予對應(yīng)的權(quán)限。
5、集群測試
在kafka1上面發(fā)送消息：
- bin/kafka-console-producer.sh --broker-list localhost:9091 --topic test
在kafka2、kafka3消費消息：
- bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic
Spring Boot 集成 Kafka 實戰(zhàn)
1、添加spring-kafka依賴
2.1.0.RELEASE org.springframework.kafka spring-kafka ${spring-kafka.version}
2、添加Spring Boot的自動配置
自動配置類：
- 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);
- }
新聞標(biāo)題：輕松上手SpringBoot&Kafka實戰(zhàn)！
文章出自：http://www.5511xx.com/article/djpgpsc.html
