資料源管理 | Kafka叢集環境搭建，消息存儲機制詳解

一、Kafka叢集環境

1、環境版本

版本：kafka2.11，zookeeper3.4           

注意：這裡zookeeper3.4也是基于叢集模式部署。

2、解壓重命名

tar -zxvf kafka_2.11-0.11.0.0.tgz
mv kafka_2.11-0.11.0.0 kafka2.11           

建立日志目錄

[root@en-master kafka2.11]# mkdir logs           

注意：以上操作需要同步到叢集下其他服務上。

3、添加環境變量

vim /etc/profile
export KAFKA_HOME=/opt/kafka2.11
export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile           

4、修改核心配置

[root@en-master /opt/kafka2.11/config]# vim server.properties
-- 核心修改如下
# 唯一編号
broker.id=0
# 開啟topic删除
delete.topic.enable=true
# 日志位址
log.dirs=/opt/kafka2.11/logs
# zk叢集
zookeeper.connect=zk01:2181,zk02:2181,zk03:2181           

注意：broker.id安裝叢集服務個數編排即可，叢集下不能重複。

5、啟動kafka叢集

# 啟動指令
[root@node02 kafka2.11]# bin/kafka-server-start.sh -daemon config/server.properties
# 停止指令
[root@node02 kafka2.11]# bin/kafka-server-stop.sh
# 程序檢視
[root@node02 kafka2.11]# jps           

注意：這裡預設啟動了zookeeper叢集服務，并且叢集下的kafka分别啟動。

6、基礎管理指令

建立topic

bin/kafka-topics.sh --zookeeper zk01:2181 \
--create --replication-factor 3 --partitions 1 --topic one-topic           

參數說明：

• replication-factor 定義副本個數
• partitions 定義分區個數
• topic：定義topic名稱

檢視topic清單

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

修改topic分區

bin/kafka-topics.sh --zookeeper zk01:2181 --alter --topic one-topic --partitions 5           

檢視topic

bin/kafka-topics.sh --zookeeper zk01:2181 \
--describe --topic one-topic           

發送消息

bin/kafka-console-producer.sh \
--broker-list 192.168.72.133:9092 --topic one-topic           

消費消息

bin/kafka-console-consumer.sh \
--bootstrap-server 192.168.72.133:9092 --from-beginning --topic one-topic           

删除topic

bin/kafka-topics.sh --zookeeper zk01:2181 \
--delete --topic first           

7、Zk叢集用處

Kafka叢集中有一個broker會被選舉為Controller，Controller依賴Zookeeper環境，管理叢集broker的上下線，所有topic的分區副本配置設定和leader選舉等工作。

二、消息攔截案例

1、攔截器簡介

Kafka中間件的Producer攔截器主要用于實作消息發送的自定義控制邏輯。使用者可以在消息發送前以及回調邏輯執行前有機會對消息做一些自定義，比如消息修改等，發送狀态監控等，使用者可以指定多個攔截器按順序執行攔截。

核心方法

• configure：擷取配置資訊和初始化資料時調用；
• onSend：消息被序列化以及和計算分區前調用該方法，可以對消息做操作；
• onAcknowledgement：消息發送到Broker之後，或發送過程失敗時調用；
• close：關閉攔截器調用，執行一些資源清理工作；

注意：這裡說的攔截器是針對消息發送流程。

2、自定義攔截

定義方式：實作ProducerInterceptor接口即可。

攔截器一：在onSend方法中，對攔截的消息進行修改。

@Component
public class SendStartInterceptor implements ProducerInterceptor<String, String> {

    private final Logger LOGGER = LoggerFactory.getLogger("SendStartInterceptor");
    @Override
    public void configure(Map<String, ?> configs) {
        LOGGER.info("configs...");
    }
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // 修改消息内容
        return new ProducerRecord<>(record.topic(), record.partition(),
                                    record.timestamp(), record.key(),
                              "onSend：{" + record.value()+"}");
    }
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        LOGGER.info("onAcknowledgement...");
    }
    @Override
    public void close() {
        LOGGER.info("SendStart close...");
    }
}           

攔截器二：在onAcknowledgement方法中，判斷消息是否發送成功。

@Component
public class SendOverInterceptor implements ProducerInterceptor<String, String> {

    private final Logger LOGGER = LoggerFactory.getLogger("SendOverInterceptor");
    @Override
    public void configure(Map<String, ?> configs) {
        LOGGER.info("configs...");
    }

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        LOGGER.info("record...{}", record.value());
        return record ;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception != null){
            LOGGER.info("Send Fail...exe-msg",exception.getMessage());
        }
        LOGGER.info("Send success...");
    }

    @Override
    public void close() {
        LOGGER.info("SendOver close...");
    }
}           

加載攔截器：基于一個KafkaProducer配置Bean，加入攔截器。

@Configuration
public class KafkaConfig {

    @Bean
    public Producer producer (){
        Properties props = new Properties();
        // 省略其他配置...
        // 添加攔截器
        List<String> interceptors = new ArrayList<>();
        interceptors.add("com.kafka.cluster.interceptor.SendStartInterceptor");
        interceptors.add("com.kafka.cluster.interceptor.SendOverInterceptor");
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
        return new KafkaProducer<>(props) ;
    }
}           

3、代碼案例

@RestController
public class SendMsgWeb {
    @Resource
    private KafkaProducer<String,String> producer ;
    @GetMapping("/sendMsg")
    public String sendMsg (){
        producer.send(new ProducerRecord<>("one-topic", "msgKey", "msgValue"));
        return "success" ;
    }
}           

基于上述自定義Bean類型，進行消息發送，關注攔截器中列印日志資訊。

三、Kafka存儲分析

說明：該過程基于上述案例producer.send方法追蹤的源碼執行流程，源碼中的過程相對清楚，涉及的核心流程如下。

1、消息生成過程

Producer發送消息采用的是異步發送的方式,消息發送過程如下：

• Producer發送消息之後，經過攔截器，序列化，事務判斷；
• 流程執行後，消息内容放入容器中；
• 容器在指定時間内如果裝滿(size),會喚醒Sender線程;
• 容器如果在指定時間内沒有裝滿，也會執行一次Sender線程喚醒；
• 喚醒Sender線程之後，把容器資料拉取到topic中；

絮叨一句：讀這些中間件的源碼，不僅能開闊思維，也會讓自己意識到平時寫的代碼可能真的叫搬磚。

2、存儲機制

Kafka中消息是以topic進行辨別分類，生産者面向topic生産消息，topic分區(partition)是實體上的存儲，基于消息日志檔案的方式。

• 每個partition對應于一個log檔案,發送的消息不斷追加到該log檔案末端；
• log檔案中存儲的就是producer生産的消息資料，采用分片和索引機制；
• partition分為多個segment。每個segment對應兩個(.index)和(.log)檔案；
• index檔案類型存儲的索引資訊；
• log檔案存儲消息的資料；
• 索引檔案中的中繼資料指向對應資料檔案中message的實體偏移位址；
• 消費者組中的每個消費者，都會實時記錄消費的消息offset位置；
• 當然消息消費出錯時，恢複是從上次的記錄位置繼續消費；

3、事務控制機制

Kafka支援消息的事務控制

Producer事務

跨分區跨會話的事務原理，引入全局唯一的TransactionID，并将Producer獲得的PID和TransactionID綁定。Producer重新開機後可以通過正在進行的TransactionID獲得原來的PID。

Kafka基于TransactionCoordinator元件管理Transaction，Producer通過和TransactionCoordinator互動獲得TransactionID對應的任務狀态。TransactionCoordinator将事務狀态寫入Kafka的内部Topic，即使整個服務重新開機，進行中的事務狀态可以得到恢複。

Consumer事務

Consumer消息消費，事務的保證強度很低，無法保證消息被精确消費，因為同一事務的消息可能會出現重新開機後已經被删除的情況。

四、源代碼位址

GitHub·位址
https://github.com/cicadasmile/data-manage-parent
GitEE·位址
https://gitee.com/cicadasmile/data-manage-parent