kafka環境搭建1-單機

kafka是基於scala語言開發，所以需要java執行環境，下載前請先確認是否已經安裝並配置java環境

下載安裝

• 解壓縮

tar -xzf kafka_2.11-0.9.0.0.tgz

配置訪問地址：

config/server.properties

• #內網地址：192.168.85.171
• listeners=PLAINTEXT://192.168.85.171:9092
• #外網地址：192.168.85.171
• advertised.listeners=PLAINTEXT://192.168.85.171:9092

開放默認埠：

/sbin/iptables 
 
-I INPUT -p tcp --dport 9092 -j ACCEPT

/sbin/iptables -I INPUT -p tcp --dport 2181 -j ACCEPT

啟動

首先要進入kafka目錄

• 啟動zookeeper server（kafka自帶的zookeeper）

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

注意 & 號的使用：命令結尾新增 & 號，可以在執行完命令後退出當前命令環境卻不會結束程序

• 啟動kafka server

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

建立主題

kafka生產和消費資料，必須基於主題topic。主題其實就是對訊息的分類。

• 建立主題：名稱為“test”、複製數目為1、partitions為1的topic主題

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

replication-factor ： 複製數目，提供failover機制；1代表只在一個broker上有資料記錄，一般值都大於1，代表一份資料會自動同步到其他的多個broker，防止某個broker宕機後資料丟失。

• 檢視已經創建的主題

bin/kafka-topics.sh --list --zookeeper localhost:2181

啟動生產者和消費者

生產者產生（輸入）資料，消費者消費（輸出）資料

• 啟動生產者

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

啟動後，在命令列下每輸入一些字串按下回車時，就作為一個訊息併發送的kafka

• 啟動消費者

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

啟動消費者時，建議另開一個ssh視窗，方便一遍通過生產者命令列輸入訊息，一遍觀察消費者消費的資料

當在生產者下輸入訊息並回車後，在消費者視窗下就能立即看到對應的訊息，這就說明環境搭建成功。

Springboot +kafka

pom

<dependency>

   <groupId>org.springframework.kafka</groupId>

   <artifactId>spring-kafka</artifactId>

   <version>1.2.2.RELEASE</version>

</dependency>

application.properties

# kafka

spring.kafka.bootstrap-servers=192.168.85.171:9092

spring.kafka.consumer.group-id=group1

spring.kafka.consumer.auto-offset-reset=earliest

controller

package com.kaicom.nettytest.kafka;



import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.SpringApplication;

import org.springframework.kafka.annotation.KafkaListener;

import org.springframework.kafka.core.KafkaTemplate;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.ResponseBody;

import org.springframework.web.bind.annotation.RestController;



/**

 * @Author: BillYu

 * @Description:kafka連線測試
* @Date: Created in 下午3:52 2018/6/4.

 */

@RestController

public class SampleController {

    private final Logger log = LoggerFactory.getLogger(SampleController.class);







    @Autowired

    private KafkaTemplate<String, String> template;



    @RequestMapping("/send")

    @ResponseBody

    String send(String topic, String key, String data) {

        template.send(topic, key, data);

        return "success";

    }







    @KafkaListener(id = "test", topics = "test")

    public void listenTest(ConsumerRecord<?, ?> cr) throws Exception {

        System.out.println("===>listen");

        log.info("{} - {} : {}", cr.topic(), cr.key(), cr.value());

    }









}

視覺化監控安裝

 一、KafkaOffsetMonitor簡述

KafkaOffsetMonitor是Kafka的一款客戶端消費監控工具，用來實時監控Kafka服務的Consumer以及它們所在的Partition中的Offset，我們可以瀏覽當前的消費者組，並且每個Topic的所有Partition的消費情況都可以一目瞭然。

二、KafkaOffsetMonitor下載

KafkaOffsetMonitor託管在Github上，可以通過Github下載。
下載地址：https://github.com/quantifind/KafkaOffsetMonitor/releases

或者下載百度網盤：連結：https://pan.baidu.com/s/1geEBEvT 密碼：jaeu

github提供下載的jar包應用了google網站提供的js檔案，由於被牆的原因導致web介面無法訪問。可以直接使用別人修改後的jar包，也可以從github上拉取原始碼然後修改打包。

三、KafkaOffsetMonitor啟動

將下載下來的KafkaOffsetMonitor jar包上傳到linux上，可以新建一個目錄KafkaMonitor，用於存放KafkaOffsetMonitor-assembly-0.2.0.jar進入到KafkaMonitor目錄下，通過java編譯命令來執行這個jar包：

[[email protected] KafkaMonitor]# java -cp KafkaOffsetMonitor-assembly-0.2.0.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb --zk 10.0.0.50:12181,10.0.0.60:12181,10.0.0.70:12181 --port 8088  --refresh 5.seconds --retain 1.days

按回車後，可以看到控制檯輸出：

serving resources from: jar:file:/data/KafkaMonitor/KafkaOffsetMonitor-assembly-0.2.0.jar!/offsetapp

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".

SLF4J: Defaulting to no-operation (NOP) logger implementation

SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

2018-01-05 21:17:36.267:INFO:oejs.Server:jetty-7.x.y-SNAPSHOT

log4j:WARN No appenders could be found for logger (org.I0Itec.zkclient.ZkConnection).

log4j:WARN Please initialize the log4j system properly.

log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

2018-01-05 21:17:36.630:INFO:oejsh.ContextHandler:started o.e.j.s.ServletContextHandler{/,jar:file:/data/KafkaMonitor/KafkaOffsetMonitor-assembly-0.2.0.jar!/offsetapp}

2018-01-05 21:17:36.662:INFO:oejs.AbstractConnector:Started [email protected]0.0.0.0:8088

如果沒有指定埠，則預設會開啟一個隨機埠。

引數說明:

zk ：zookeeper主機地址，如果有多個，用逗號隔開

port ：應用程式埠

refresh ：應用程式在資料庫中重新整理和儲存點的頻率

retain ：在db中保留多長時間

dbName ：儲存的資料庫檔名，預設為offsetapp

為了更方便的啟動KafkaOffsetMonitor，可以寫一個啟動指令碼來直接執行，我這裡新建一個名為：kafka-monitor-start.sh的指令碼，然後編輯這個指令碼：

[[email protected] KafkaMonitor]# vim kafka-monitor-start.sh 

java -Xms512M -Xmx512M -Xss1024K -XX:PermSize=256m -XX:MaxPermSize=512m  -cp KafkaOffsetMonitor-assembly-0.2.0.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb \

--port 8088 \

--zk 10.0.0.50:12181,10.0.0.60:12181,10.0.0.70:12181 \

--refresh 5.minutes \

--retain 1.day >/dev/null 2>&1;

然後退出儲存即可，接下來修改一下kafka-monitor-start.sh的許可權

[[email protected] KafkaMonitor]# chmod +x kafka-monitor-start.sh 

啟動KafkaOffsetMonitor：

[[email protected] KafkaMonitor]# nohup /data/KafkaMonitor/kafka-monitor-start.sh &

[1] 6551

[[email protected] KafkaMonitor]# lsof -i:8088

COMMAND  PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME

java    6552 root   16u  IPv6  26047      0t0  TCP *:radan-http (LISTEN)

四、KafkaOffsetMonitor Web UI

在遊覽器中輸入：http://ip:port即可以檢視KafkaOffsetMonitor Web UI,如下圖：