輕松上手 Spring Boot & Kafka 實(shí)戰(zhàn)！

作者：Java領(lǐng)域佼佼者 2020-04-23 15:59:04

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

創(chuàng)新互聯(lián)建站服務(wù)項(xiàng)目包括博樂網(wǎng)站建設(shè)、博樂網(wǎng)站制作、博樂網(wǎng)頁制作以及博樂網(wǎng)絡(luò)營銷策劃等。多年來，我們專注于互聯(lián)網(wǎng)行業(yè)，利用自身積累的技術(shù)優(yōu)勢、行業(yè)經(jīng)驗(yàn)、深度合作伙伴關(guān)系等，向廣大中小型企業(yè)、政府機(jī)構(gòu)等提供互聯(lián)網(wǎng)行業(yè)的解決方案，博樂網(wǎng)站推廣取得了明顯的社會效益與經(jīng)濟(jì)效益。目前，我們服務(wù)的客戶以成都為中心已經(jīng)輻射到博樂省份的部分城市，未來相信會繼續(xù)擴(kuò)大服務(wù)區(qū)域并繼續(xù)獲得客戶的支持與信任！

Kafka集群安裝、配置和啟動

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分別修改對應(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，分別進(jìn)行Kafka1-3目錄:

  
 
 
 

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

啟動報(bào)文件失敗，需要手動創(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，分別進(jìn)行Kafka1-3目錄:

  
 
 
 

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

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

5、集群測試

在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依賴

  
 
 
 

   
  
  
  
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); 
   
  
  
  
} 
  
 
 
 

文章名稱：輕松上手SpringBoot&Kafka實(shí)戰(zhàn)！
標(biāo)題URL：http://uogjgqi.cn/article/djpgpsc.html