摘要：慕課網(wǎng)流處理平臺(tái)學(xué)習(xí)總結(jié)時(shí)間年月日星期日說明本文部分內(nèi)容均來自慕課網(wǎng)。

慕課網(wǎng)《Kafka流處理平臺(tái)》學(xué)習(xí)總結(jié)

時(shí)間：2018年09月09日星期日

說明：本文部分內(nèi)容均來自慕課網(wǎng)。@慕課網(wǎng)：https://www.imooc.com

教學(xué)源碼：無

學(xué)習(xí)源碼：https://github.com/zccodere/s...

課程介紹

Kafka概念解析

Kafka結(jié)構(gòu)設(shè)計(jì)

Kafka場(chǎng)景應(yīng)用

Kafka高級(jí)特性

LinkedIn 開源

Databus 分布式數(shù)據(jù)同步系統(tǒng)

Cubert 高性能計(jì)算引擎

ParSeq Java異步處理框架

Kafka 分布式發(fā)布訂閱消息系統(tǒng)，流處理平臺(tái)

Kafka發(fā)展歷程

LinkedIn 開發(fā)

2011年初開源，加入Apache基金會(huì)

2012年從Apache Incubator畢業(yè)

Apache頂級(jí)開源項(xiàng)目

Kafka的特性

可以發(fā)布和訂閱且記錄數(shù)據(jù)的流，類似于消息隊(duì)列

數(shù)據(jù)流存儲(chǔ)的平臺(tái)，具備容錯(cuò)能力

在數(shù)據(jù)產(chǎn)生時(shí)就可以進(jìn)行處理

Kafka通常被用于

構(gòu)建實(shí)時(shí)數(shù)據(jù)流管道

構(gòu)建實(shí)時(shí)數(shù)據(jù)流處理

Kafka是什么

面向于數(shù)據(jù)流的生產(chǎn)、轉(zhuǎn)換、存儲(chǔ)、消費(fèi)整體的流處理平臺(tái)

Kafka不僅僅是一個(gè)消息隊(duì)列

Producer：數(shù)據(jù)生產(chǎn)者

消息和數(shù)據(jù)的生產(chǎn)者

向Kafka的一個(gè)topic發(fā)布消息的進(jìn)程或代碼或服務(wù)

Consumer：數(shù)據(jù)消費(fèi)者

消息和數(shù)據(jù)的消費(fèi)者

向Kafka訂閱數(shù)據(jù)（topic）并且處理其發(fā)布的消息的進(jìn)程或代碼或服務(wù)

Consumer Group：消費(fèi)者組

對(duì)于同一個(gè)topic，會(huì)廣播給不同的Group

一個(gè)Group中，只有一個(gè)Consumer可以消費(fèi)該消息

Broker：服務(wù)節(jié)點(diǎn)

Kafka集群中的每個(gè)Kafka節(jié)點(diǎn)

Topic：主題

Kafka消息的類別

對(duì)數(shù)據(jù)進(jìn)行區(qū)分、隔離

Partition：分區(qū)

Kafka中數(shù)據(jù)存儲(chǔ)的基本單元

一個(gè)topic數(shù)據(jù)，會(huì)被分散存儲(chǔ)到多個(gè)Partition

一個(gè)Partition只會(huì)存在一個(gè)Broker上

每個(gè)Partition是有序的

Replication：分區(qū)的副本

同一個(gè)Partition可能會(huì)有多個(gè)Replication

多個(gè)Replication之間數(shù)據(jù)是一樣的

Replication Leader：副本的老大

一個(gè)Partition的多個(gè)Replication上

需要一個(gè)Leader負(fù)責(zé)該P(yáng)artition上與Producer和Consumer交互

Replication Manager：副本的管理者

負(fù)責(zé)管理當(dāng)前Broker所有分區(qū)和副本的信息

處理KafkaController發(fā)起的一些請(qǐng)求

副本狀態(tài)的切換

添加、讀取消息等

Partition：分區(qū)

每一個(gè)Topic被切分為多個(gè)Partition

消費(fèi)者數(shù)目少于或等于Partition的數(shù)目

Broker Group中的每一個(gè)Broker保存Topic的一個(gè)或多個(gè)Partition

Consumer Group中的僅有一個(gè)Consumer讀取Topic的一個(gè)或多個(gè)Partition，并且是惟一的Consumer

Replication：分區(qū)的副本

當(dāng)集群中有Broker掛掉的情況，系統(tǒng)可以主動(dòng)地使Replication提供服務(wù)

系統(tǒng)默認(rèn)設(shè)置每一個(gè)Topic的Replication系數(shù)為1，可以在創(chuàng)建Topic時(shí)多帶帶設(shè)置

Replication的基本單位是Topic的Partition

所有的讀和寫都從Replication Leader進(jìn)行，Replication Followers只是作為備份

Replication Followers必須能夠及時(shí)復(fù)制Replication Leader的數(shù)據(jù)

增加容錯(cuò)性與可擴(kuò)展性

Kafka功能結(jié)構(gòu)

Kafka數(shù)據(jù)流勢(shì)

Kafka消息結(jié)構(gòu)

Offset：當(dāng)前消息所處于的偏移

Length：消息的長(zhǎng)度

CRC32：校驗(yàn)字段，用于校驗(yàn)當(dāng)前信息的完整性

Magic：很多分布式系統(tǒng)都會(huì)設(shè)計(jì)該字段，固定的數(shù)字，用于快速判定當(dāng)前信息是否為Kafka消息

attributes：可選字段，消息的屬性

Timestamp：時(shí)間戳

Key Length：Key的長(zhǎng)度

Key：Key

Value Length：Value的長(zhǎng)度

Value：Value

Kafka特點(diǎn)：分布式

多分區(qū)

多副本

多訂閱者

基于Zookeeper調(diào)度

Kafka特點(diǎn)：高性能

高吞吐量

低延遲

高并發(fā)

時(shí)間復(fù)雜度為O(1)

Kafka特點(diǎn)：持久性與擴(kuò)展性

數(shù)據(jù)可持久化

容錯(cuò)性

支持在線水平擴(kuò)展

消息自動(dòng)平衡

Kafka應(yīng)用場(chǎng)景

消息隊(duì)列

行為跟蹤

元信息監(jiān)控

日志收集

流處理

事件源

持久性日志（commit log）

Kafka簡(jiǎn)單案例

部署啟動(dòng)

簡(jiǎn)單生產(chǎn)者

簡(jiǎn)單消費(fèi)者

學(xué)習(xí)筆記

1.下載與安裝
Zookeeper下載：https://zookeeper.apache.org/releases.html#download
Kafka下載：http://kafka.apache.org/downloads
安裝：解壓、配置環(huán)境變量

2.Zookeeper啟動(dòng)
解壓：tar -zxf zookeeper-3.4.12.tar.gz
目錄：cd zookeeper-3.4.12/bin
啟動(dòng)：./zkServer.sh start /home/zc/server/kafka_2.12-2.0.0/config/zookeeper.properties

3.Kafka啟動(dòng)
解壓：tar -zxf kafka_2.12-2.0.0.tgz
目錄：cd kafka_2.12-2.0.0
啟動(dòng)：sudo bin/kafka-server-start.sh  config/server.properties

4.使用控制臺(tái)操作生產(chǎn)者與消費(fèi)者
創(chuàng)建Topic：sudo ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic myimooc-kafka-topic
查看Topic：sudo ./bin/kafka-topics.sh --list --zookeeper localhost:2181
啟動(dòng)生產(chǎn)者：sudo ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic myimooc-kafka-topic
啟動(dòng)消費(fèi)者：sudo ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic myimooc-kafka-topic --from-beginning
生產(chǎn)消息：first message
生產(chǎn)消息：second message

創(chuàng)建49-kafka-example的maven工程pom如下

    
        49-kafka
        com.myimooc
        1.0-SNAPSHOT
    
    4.0.0

    49-kafka-example

    
        2.0.4.RELEASE
    

    
        
            
                org.springframework.boot
                spring-boot-parent
                ${spring.boot.version}
                pom
                import
            
        
    

    
        
            org.springframework.boot
            spring-boot-starter-web
        
        
            org.springframework.kafka
            spring-kafka
        
        
            com.alibaba
            fastjson
            1.2.36
        
        
            org.springframework.boot
            spring-boot-starter-test
            test
        
    

    
        
            
                org.springframework.boot
                spring-boot-maven-plugin
            
        
    

1.編寫MessageEntity

package com.myimooc.kafka.example.common;

import java.util.Objects;

/**
 * 

 * 標(biāo)題: 消息實(shí)體

 * 描述: 消息實(shí)體

 * 時(shí)間: 2018/09/09

 *
 * @author zc
 */
public class MessageEntity {
    /**
     * 標(biāo)題
     */
    private String title;
    /**
     * 內(nèi)容
     */
    private String body;

    @Override
    public String toString() {
        return "MessageEntity{" +
                "title="" + title + """ +
                ", body="" + body + """ +
                "}";
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        MessageEntity that = (MessageEntity) o;
        return Objects.equals(title, that.title) &&
                Objects.equals(body, that.body);
    }

    @Override
    public int hashCode() {
        return Objects.hash(title, body);
    }

    public String getTitle() {
        return title;
    }

    public void setTitle(String title) {
        this.title = title;
    }

    public String getBody() {
        return body;
    }

    public void setBody(String body) {
        this.body = body;
    }
}

2.編寫SimpleProducer

package com.myimooc.kafka.example.producer;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * 

 * 標(biāo)題: 生產(chǎn)者

 * 描述: 生產(chǎn)者

 * 時(shí)間: 2018/09/09

 *
 * @author zc
 */
@Component
public class SimpleProducer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void send(String topic, String key, Object entity) {
        logger.info("發(fā)送消息入?yún)ⅲ簕}", entity);
        ProducerRecord record = new ProducerRecord<>(
                topic,
                key,
                JSON.toJSONString(entity)
        );

        long startTime = System.currentTimeMillis();
        ListenableFuture> future = this.kafkaTemplate.send(record);
        future.addCallback(new ListenableFutureCallback>() {
            @Override
            public void onFailure(Throwable ex) {
                logger.error("消息發(fā)送失敗：{}", ex);
            }

            @Override
            public void onSuccess(SendResult result) {
                long elapsedTime = System.currentTimeMillis() - startTime;

                RecordMetadata metadata = result.getRecordMetadata();
                StringBuilder record = new StringBuilder(128);
                record.append("message(")
                        .append("key = ").append(key).append(",")
                        .append("message = ").append(entity).append(")")
                        .append("send to partition(").append(metadata.partition()).append(")")
                        .append("with offset(").append(metadata.offset()).append(")")
                        .append("in ").append(elapsedTime).append(" ms");
                logger.info("消息發(fā)送成功：{}", record.toString());
            }
        });
    }
}

3.編寫SimpleConsumer

package com.myimooc.kafka.example.consumer;

import com.alibaba.fastjson.JSONObject;
import com.myimooc.kafka.example.common.MessageEntity;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * 

 * 標(biāo)題: 消費(fèi)者

 * 描述: 消費(fèi)者

 * 時(shí)間: 2018/09/09

 *
 * @author zc
 */
@Component
public class SimpleConsumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = "${kafka.topic.default}")
    public void listen(ConsumerRecord record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        //判斷是否NULL
        Optional kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            //獲取消息
            Object message = kafkaMessage.get();

            MessageEntity messageEntity = JSONObject.parseObject(message.toString(), MessageEntity.class);

            logger.info("接收消息Topic：{}", topic);
            logger.info("接收消息Record：{}", record);
            logger.info("接收消息Message：{}", messageEntity);
        }
    }

}

4.編寫Response

package com.myimooc.kafka.example.common;

import java.io.Serializable;

/**
 * 

 * 標(biāo)題: REST請(qǐng)求統(tǒng)一響應(yīng)對(duì)象

 * 描述: REST請(qǐng)求統(tǒng)一響應(yīng)對(duì)象

 * 時(shí)間: 2018/09/09

 *
 * @author zc
 */
public class Response implements Serializable {

    private static final long serialVersionUID = -972246069648445912L;
    /**
     * 響應(yīng)編碼
     */
    private int code;
    /**
     * 響應(yīng)消息
     */
    private String message;

    public Response() {
    }

    public Response(int code, String message) {
        this.code = code;
        this.message = message;
    }

    @Override
    public String toString() {
        return "Response{" +
                "code=" + code +
                ", message="" + message + """ +
                "}";
    }

    public int getCode() {
        return code;
    }

    public void setCode(int code) {
        this.code = code;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }
}

5.編寫ErrorCode

package com.myimooc.kafka.example.common;

/**
 * 

 * 標(biāo)題: 錯(cuò)誤編碼

 * 描述: 錯(cuò)誤編碼

 * 時(shí)間: 2018/09/09

 *
 * @author zc
 */
public class ErrorCode {
    /**
     * 成功
     */
    public final static int SUCCESS = 200;
    /**
     * 失敗
     */
    public final static int EXCEPTION = 500;

}

6.編寫ProducerController

package com.myimooc.kafka.example.controller;

import com.alibaba.fastjson.JSON;
import com.myimooc.kafka.example.common.ErrorCode;
import com.myimooc.kafka.example.common.MessageEntity;
import com.myimooc.kafka.example.common.Response;
import com.myimooc.kafka.example.producer.SimpleProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.*;

/**
 * 

 * 標(biāo)題: 生產(chǎn)者Controller

 * 描述: 生產(chǎn)者Controller

 * 時(shí)間: 2018/09/09

 *
 * @author zc
 */
@RestController
@RequestMapping("/producer")
public class ProducerController {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private SimpleProducer simpleProducer;

    @Value("${kafka.topic.default}")
    private String topic;

    private static final String KEY = "key";

    @PostMapping("/send")
    public Response sendKafka(@RequestBody MessageEntity message) {
        try {
            logger.info("kafka的消息：{}", JSON.toJSONString(message));
            this.simpleProducer.send(topic, KEY, message);
            logger.info("kafka消息發(fā)送成功！");
            return new Response(ErrorCode.SUCCESS,"kafka消息發(fā)送成功");
        } catch (Exception ex) {
            logger.error("kafka消息發(fā)送失敗：", ex);
            return new Response(ErrorCode.EXCEPTION,"kafka消息發(fā)送失敗");
        }
    }
}

7.編寫application.properties

##----------kafka配置
## TOPIC
kafka.topic.default=myimooc-kafka-topic
# kafka地址
spring.kafka.bootstrap-servers=192.168.0.105:9092
# 生產(chǎn)者配置
spring.kafka.producer.retries=0
# 批量發(fā)送消息的數(shù)量
spring.kafka.producer.batch-size=4096
# 緩存容量
spring.kafka.producer.buffer-memory=40960
# 指定消息key和消息體的編解碼方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# 消費(fèi)者配置
spring.kafka.consumer.group-id=myimooc
spring.kafka.consumer.auto-commit-interval=100
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.enable-auto-commit=true
# 指定消息key和消息體的編解碼方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# 指定listener 容器中的線程數(shù)，用于提高并發(fā)量
spring.kafka.listener.concurrency=3

8.編寫ExampleApplication

package com.myimooc.kafka.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;

/**
 * 

 * 標(biāo)題: 啟動(dòng)類

 * 描述: 啟動(dòng)類

 * 時(shí)間: 2018/09/09

 *
 * @author zc
 */
@SpringBootApplication
@EnableKafka
public class ExampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(ExampleApplication.class, args);
    }

}

為什么要支持事務(wù)

滿足“讀取-處理-寫入”模式

流處理需求的不斷增強(qiáng)

不準(zhǔn)確的數(shù)據(jù)處理的容忍度不斷降低

數(shù)據(jù)傳輸?shù)氖聞?wù)定義

最多一次：消息不會(huì)被重復(fù)發(fā)送，最多被傳輸一次，但也有可能一次不傳輸

最少一次：消息不會(huì)被漏發(fā)送，最少被傳輸一次，但也有可能被重復(fù)傳輸

精確的一次（Exactly once）：不會(huì)漏傳輸也不會(huì)重復(fù)傳輸，每個(gè)消息都被傳輸一次且僅僅被傳輸一次，這是大家所期望的

事務(wù)保證

內(nèi)部重試問題：Procedure冪等處理

多分區(qū)原子寫入

避免僵尸實(shí)例

每個(gè)事務(wù)Procedure分配一個(gè) transactionl. id，在進(jìn)程重新啟動(dòng)時(shí)能夠識(shí)別相同的Procedure實(shí)例
Kafka增加了一個(gè)與transactionl.id相關(guān)的epoch，存儲(chǔ)每個(gè)transactionl.id內(nèi)部元數(shù)據(jù)
一旦epoch被觸發(fā)，任務(wù)具有相同的transactionl.id和更舊的epoch的Producer被視為僵尸，Kafka會(huì)拒絕來自這些Producer的后續(xù)事務(wù)性寫入

零拷貝簡(jiǎn)介

網(wǎng)絡(luò)傳輸持久性日志塊

Java Nio channel.transforTo()方法

Linux sendfile系統(tǒng)調(diào)用

文件傳輸?shù)骄W(wǎng)絡(luò)的公共數(shù)據(jù)路徑

第一次拷貝：操作系統(tǒng)將數(shù)據(jù)從磁盤讀入到內(nèi)核空間的頁(yè)緩存

第二次拷貝：應(yīng)用程序?qū)?shù)據(jù)從內(nèi)核空間讀入到用戶空間緩存中

第三次拷貝：應(yīng)用程序?qū)?shù)據(jù)寫回到內(nèi)核空間到socket緩存中

第四次拷貝：操作系統(tǒng)將數(shù)據(jù)從socket緩沖區(qū)復(fù)制到網(wǎng)卡緩沖區(qū)，以便將數(shù)據(jù)經(jīng)網(wǎng)絡(luò)發(fā)出

零拷貝過程（指內(nèi)核空間和用戶空間的交互拷貝次數(shù)為零）

第一次拷貝：操作系統(tǒng)將數(shù)據(jù)從磁盤讀入到內(nèi)核空間的頁(yè)緩存

將數(shù)據(jù)的位置和長(zhǎng)度的信息的描述符增加至內(nèi)核空間（socket緩存區(qū)）

第二次拷貝：操作系統(tǒng)將數(shù)據(jù)從內(nèi)核拷貝到網(wǎng)卡緩沖區(qū)，以便將數(shù)據(jù)經(jīng)網(wǎng)絡(luò)發(fā)出

文件傳輸?shù)骄W(wǎng)絡(luò)的公共數(shù)據(jù)路徑演變

課程總結(jié)

Kafka基礎(chǔ)概念與結(jié)構(gòu)

Kafka的特點(diǎn)

Kafka應(yīng)用場(chǎng)景

Kafka應(yīng)用案例

Kafka高級(jí)特性