Kafka使用入門教程

時間  2019-11-29
標籤 kafka 使用 入門 教程 欄目 Kafka 简体版
原文   原文鏈接

轉載自http://www.linuxidc.com/Linux/2014-07/104470.htmjava

介紹

Kafka是一個分佈式的、可分區的、可複製的消息系統。它提供了普通消息系統的功能，但具備本身獨特的設計。這個獨特的設計是什麼樣的呢？
 
首先讓咱們看幾個基本的消息系統術語：
  • Kafka將消息以topic爲單位進行概括。
  • 將向Kafka topic發佈消息的程序成爲producers.
  • 將預訂topics並消費消息的程序成爲consumer.
  • Kafka以集羣的方式運行，能夠由一個或多個服務組成，每一個服務叫作一個broker.
producers經過網絡將消息發送到Kafka集羣，集羣向消費者提供消息，以下圖所示：
客戶端和服務端經過TCP協議通訊。Kafka提供了Java客戶端，而且對 多種語言都提供了支持。
 

相關閱讀：node

分佈式發佈訂閱消息系統 Kafka 架構設計 http://www.linuxidc.com/Linux/2013-11/92751.htmlinux

Apache Kafka 代碼實例 http://www.linuxidc.com/Linux/2013-11/92754.htm正則表達式

Apache Kafka 教程筆記 http://www.linuxidc.com/Linux/2014-01/94682.htmapache

Topics 和Logs

先來看一下Kafka提供的一個抽象概念:topic.
一個topic是對一組消息的概括。對每一個topic，Kafka 對它的日誌進行了分區，以下圖所示：
每一個分區都由一系列有序的、不可變的消息組成，這些消息被連續的追加到分區中。分區中的每一個消息都有一個連續的序列號叫作offset,用來在分區中惟一的標識這個消息。
 
在一個可配置的時間段內，Kafka集羣保留全部發布的消息，無論這些消息有沒有被消費。好比，若是消息的保存策略被設置爲2天，那麼在一個消 息被髮布的兩天時間內，它都是能夠被消費的。以後它將被丟棄以釋放空間。Kafka的性能是和數據量無關的常量級的，因此保留太多的數據並非問題。
 
實際上每一個consumer惟一須要維護的數據是消息在日誌中的位置，也就是offset.這個offset有consumer來維護：通常情 況下隨着consumer不斷的讀取消息，這offset的值不斷增長，但其實consumer能夠以任意的順序讀取消息，好比它能夠將offset設置 成爲一箇舊的值來重讀以前的消息。
 
以上特色的結合，使Kafka consumers很是的輕量級：它們能夠在不對集羣和其餘consumer形成影響的狀況下讀取消息。你可使用命令行來"tail"消息而不會對其餘正在消費消息的consumer形成影響。
 
將日誌分區能夠達到如下目的：首先這使得每一個日誌的數量不會太大，能夠在單個服務上保存。另外每一個分區能夠單獨發佈和消費，爲併發操做topic提供了一種可能。

分佈式

每一個分區在Kafka集羣的若干服務中都有副本，這樣這些持有副本的服務能夠共同處理數據和請求，副本數量是能夠配置的。副本使Kafka具有了容錯能力。
每一個分區都由一個服務器做爲「leader」，零或若干服務器做爲「followers」,leader負責處理消息的讀和 寫，followers則去複製leader.若是leader down了，followers中的一臺則會自動成爲leader。集羣中的每一個服務都會同時扮演兩個角色：做爲它所持有的一部分分區的leader，同 時做爲其餘分區的followers，這樣集羣就會據有較好的負載均衡。

Producers

Producer將消息發佈到它指定的topic中,並負責決定發佈到哪一個分區。一般簡單的由負載均衡機制隨機選擇分區，但也能夠經過特定的分區函數選擇分區。使用的更多的是第二種。

Consumers

發佈消息一般有兩種模式：隊列模式（queuing）和發佈-訂閱模式(publish-subscribe)。隊列模式中，consumers能夠同時從服務端讀取消息，每一個消息只被其中一個consumer讀到；發佈-訂閱模式中消息被廣播到全部的consumer中。

Consumers能夠加入一個consumer 組，共同競爭一個topic，topic中的消息將被分發到組中的一個成員中。同一組中的consumer能夠在不一樣的程序中，也能夠在不一樣的機器上。若是全部的consumer都在一個組中，這就成爲了傳統的隊列模式，在各consumer中實現負載均衡。

若是全部的consumer都不在不一樣的組中，這就成爲了發佈-訂閱模式，全部的消息都被分發到全部的consumer中。

更常見的是，每一個topic都有若干數量的consumer組，每一個組都是一個邏輯上的「訂閱者」，爲了容錯和更好的穩定性，每一個組由若干consumer組成。這其實就是一個發佈-訂閱模式，只不過訂閱者是個組而不是單個consumer。
由兩個機器組成的集羣擁有4個分區 (P0-P3) 2個consumer組. A組有兩個consumerB組有4個
 
相比傳統的消息系統，Kafka能夠很好的保證有序性。
傳統的隊列在服務器上保存有序的消息，若是多個consumers同時從這個服務器消費消息，服務器就會以消息存儲的順序向consumer分 發消息。雖然服務器按順序發佈消息，可是消息是被異步的分發到各consumer上，因此當消息到達時可能已經失去了原來的順序，這意味着併發消費將致使 順序錯亂。爲了不故障，這樣的消息系統一般使用「專用consumer」的概念，其實就是隻容許一個消費者消費消息，固然這就意味着失去了併發性。
 
在這方面Kafka作的更好，經過分區的概念，Kafka能夠在多個consumer組併發的狀況下提供較好的有序性和負載均衡。將每一個分區分 只分發給一個consumer組，這樣一個分區就只被這個組的一個consumer消費，就能夠順序的消費這個分區的消息。由於有多個分區，依然能夠在多 個consumer組之間進行負載均衡。注意consumer組的數量不能多於分區的數量，也就是有多少分區就容許多少併發消費。
 
Kafka只能保證一個分區以內消息的有序性，在不一樣的分區之間是不能夠的，這已經能夠知足大部分應用的需求。若是須要topic中全部消息的有序性，那就只能讓這個topic只有一個分區，固然也就只有一個consumer組消費它。

接下來一步一步搭建Kafka運行環境。

Step 1: 下載Kafka

點擊下載最新的版本並解壓.

> tar -xzf kafka_2.9.2-0.8.1.1.tgz
> cd kafka_2.9.2-0.8.1.1

Step 2: 啓動服務

Kafka用到了Zookeeper，全部首先啓動Zookper，下面簡單的啓用一個單實例的Zookkeeper服務。能夠在命令的結尾加個&符號，這樣就能夠啓動後離開控制檯。 
> bin/zookeeper-server-start.sh config/zookeeper.properties &
[2013-04-22 15:01:37,495] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
...
如今啓動Kafka: 
> bin/kafka-server-start.sh config/server.properties
[2013-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties)
[2013-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties)
...

Step 3: 建立 topic

建立一個叫作「test」的topic，它只有一個分區，一個副本。 
> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
能夠經過list命令查看建立的topic: 
> bin/kafka-topics.sh --list --zookeeper localhost:2181
test
除了手動建立topic，還能夠配置broker讓它自動建立topic.

Step 4:發送消息.

Kafka 使用一個簡單的命令行producer，從文件中或者從標準輸入中讀取消息併發送到服務端。默認的每條命令將發送一條消息。

運行producer並在控制檯中輸一些消息，這些消息將被髮送到服務端： 
> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test 
This is a messageThis is another message
ctrl+c能夠退出發送。

Step 5: 啓動consumer

Kafka also has a command line consumer that will dump out messages to standard output.
Kafka也有一個命令行consumer能夠讀取消息並輸出到標準輸出： 
> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
This is a message
This is another message

你在一個終端中運行consumer命令行，另外一個終端中運行producer命令行，就能夠在一個終端輸入消息，另外一個終端讀取消息。
這兩個命令都有本身的可選參數，能夠在運行的時候不加任何參數能夠看到幫助信息。
 

Step 6: 搭建一個多個broker的集羣

剛纔只是啓動了單個broker，如今啓動有3個broker組成的集羣，這些broker節點也都是在本機上的：
首先爲每一個節點編寫配置文件：
 
> cp config/server.properties config/server-1.properties 
> cp config/server.properties config/server-2.properties
在拷貝出的新文件中添加如下參數： 
config/server-1.properties:
    broker.id=1
    port=9093
    log.dir=/tmp/kafka-logs-1
 
config/server-2.properties:
    broker.id=2
    port=9094
    log.dir=/tmp/kafka-logs-2
broker.id在集羣中惟一的標註一個節點，由於在同一個機器上，因此必須制定不一樣的端口和日誌文件，避免數據被覆蓋。
 
We already have Zookeeper and our single node started, so we just need to start the two new nodes:
剛纔已經啓動可Zookeeper和一個節點，如今啓動另外兩個節點： 
> bin/kafka-server-start.sh config/server-1.properties &
...
> bin/kafka-server-start.sh config/server-2.properties &
...
建立一個擁有3個副本的topic: 
> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
如今咱們搭建了一個集羣，怎麼知道每一個節點的信息呢？運行「"describe topics」命令就能夠了： 
> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic       PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: my-replicated-topic      Partition: 0    Leader: 1       Replicas: 1,2,0 Isr: 1,2,0

下面解釋一下這些輸出。第一行是對全部分區的一個描述，而後每一個分區都會對應一行，由於咱們只有一個分區因此下面就只加了一行。
  • leader：負責處理消息的讀和寫，leader是從全部節點中隨機選擇的.
  • replicas：列出了全部的副本節點，無論節點是否在服務中.
  • isr：是正在服務中的節點.
在咱們的例子中，節點1是做爲leader運行。

向topic發送消息：api

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic
...
my test message 1my test message 2^C
消費這些消息： 
> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2
^C
測試一下容錯能力.Broker 1做爲leader運行，如今咱們kill掉它： 
> ps | grep server-1.properties7564 ttys002    0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.6/Home/bin/java...
> kill -9 7564
另一個節點被選作了leader,node 1 再也不出如今 in-sync 副本列表中： 
> bin/kafka-topics.sh --describe --zookeeper localhost:218192 --topic my-replicated-topic
Topic:my-replicated-topic       PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: my-replicated-topic      Partition: 0    Leader: 2       Replicas: 1,2,0 Isr: 2,0
雖然最初負責續寫消息的leader down掉了，但以前的消息仍是能夠消費的： 
> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2
^C
看來Kafka的容錯機制仍是不錯的。

下面咱們來搭建kafka的開發環境。
添加依賴
搭建開發環境須要引入kafka的jar包，一種方式是將Kafka安裝包中lib下的jar包加入到項目的classpath中，這種比較簡單了。不過咱們使用另外一種更加流行的方式：使用maven管理jar包依賴。
建立好maven項目後，在pom.xml中添加如下依賴：

<dependency>
        <groupId> org.apache.kafka</groupId >
        <artifactId> kafka_2.10</artifactId >
        <version> 0.8.0</ version>
</dependency>數組

添 加依賴後你會發現有兩個jar包的依賴找不到。不要緊我都幫你想好了，點擊這裏下載這兩個jar包，解壓後你有兩種選擇，第一種是使用mvn的 install命令將jar包安裝到本地倉庫，另外一種是直接將解壓後的文件夾拷貝到mvn本地倉庫的com文件夾下，好比個人本地倉庫是d:\mvn,完 成後個人目錄結構是這樣的：
目錄結構
 
配置程序
首先是一個充當配置文件做用的接口,配置了Kafka的各類鏈接參數：

package com.sohu.kafkademon;緩存

public interface KafkaProperties
{
    final static String zkConnect = "10.22.10.139:2181";
    final static String groupId = "group1";
    final static String topic = "topic1";
    final static String kafkaServerURL = "10.22.10.139";
    final static int kafkaServerPort = 9092;
    final static int kafkaProducerBufferSize = 64 * 1024;
    final static int connectionTimeOut = 20000;
    final static int reconnectInterval = 10000;
    final static String topic2 = "topic2";
    final static String topic3 = "topic3";
    final static String clientId = "SimpleConsumerDemoClient";
}性能優化

producer
 

package com.sohu.kafkademon;服務器

import java.util.Properties;

import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 * @author leicui bourne_cui@163.com
 */
public class KafkaProducer extends Thread
{
    private final kafka.javaapi.producer.Producer<Integer, String> producer;
    private final String topic;
    private final Properties props = new Properties();

    public KafkaProducer(String topic)
    {
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("metadata.broker.list", "10.22.10.139:9092");
        producer = new kafka.javaapi.producer.Producer<Integer, String>(new ProducerConfig(props));
        this.topic = topic;
    }

    @Override
    public void run() {
        int messageNo = 1;
        while (true)
        {
            String messageStr = new String("Message_" + messageNo);
            System.out.println("Send:" + messageStr);
            producer.send(new KeyedMessage<Integer, String>(topic, messageStr));
            messageNo++;
            try {
                sleep(3000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }

}

consumer

package com.sohu.kafkademon;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

/**
 * @author leicui bourne_cui@163.com
 */
public class KafkaConsumer extends Thread
{
    private final ConsumerConnector consumer;
    private final String topic;

    public KafkaConsumer(String topic)
    {
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
                createConsumerConfig());
        this.topic = topic;
    }

    private static ConsumerConfig createConsumerConfig()
    {
        Properties props = new Properties();
        props.put("zookeeper.connect", KafkaProperties.zkConnect);
        props.put("group.id", KafkaProperties.groupId);
        props.put("zookeeper.session.timeout.ms", "40000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        return new ConsumerConfig(props);
    }

    @Override
    public void run() {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(1));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
        ConsumerIterator<byte[], byte[]> it = stream.iterator();
        while (it.hasNext()) {
            System.out.println("receive：" + new String(it.next().message()));
            try {
                sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

簡單的發送接收
 
運行下面這個程序，就能夠進行簡單的發送接收消息了：

package com.sohu.kafkademon;

/**
 * @author leicui bourne_cui@163.com
 */
public class KafkaConsumerProducerDemo
{
    public static void main(String[] args)
    {
        KafkaProducer producerThread = new KafkaProducer(KafkaProperties.topic);
        producerThread.start();

        KafkaConsumer consumerThread = new KafkaConsumer(KafkaProperties.topic);
        consumerThread.start();
    }
}

高級別的consumer
 
下面是比較負載的發送接收的程序：

package com.sohu.kafkademon;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

/**
 * @author leicui bourne_cui@163.com
 */
public class KafkaConsumer extends Thread
{
    private final ConsumerConnector consumer;
    private final String topic;

    public KafkaConsumer(String topic)
    {
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
                createConsumerConfig());
        this.topic = topic;
    }

    private static ConsumerConfig createConsumerConfig()
    {
        Properties props = new Properties();
        props.put("zookeeper.connect", KafkaProperties.zkConnect);
        props.put("group.id", KafkaProperties.groupId);
        props.put("zookeeper.session.timeout.ms", "40000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        return new ConsumerConfig(props);
    }

    @Override
    public void run() {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(1));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
        ConsumerIterator<byte[], byte[]> it = stream.iterator();
        while (it.hasNext()) {
            System.out.println("receive：" + new String(it.next().message()));
            try {
                sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

不要畏懼文件系統!

Kafka大量依賴文件系統去存儲和緩存消息。對於硬盤有個傳統的觀念是硬盤老是很慢，這使不少人懷疑基於文件系統的架構可否提供優異的性能。實際上硬盤的快慢徹底取決於使用它的方式。設計良好的硬盤架構能夠和內存同樣快。

在6塊7200轉的SATA RAID-5磁盤陣列的線性寫速度差很少是600MB/s，可是隨即寫的速度倒是100k/s，差了差很少6000倍。現代的操做系統都對次作了大量的優 化，使用了 read-ahead 和 write-behind的技巧，讀取的時候成塊的預讀取數據，寫的時候將各類微小瑣碎的邏輯寫入組織合併成一次較大的物理寫入。對此的深刻討論能夠查看 這裏，它們發現線性的訪問磁盤，不少時候比隨機的內存訪問快得多。

爲了提升性能，現代操做系統每每使用內存做爲磁盤的緩存，現代操做系統樂於把全部空閒內存用做磁盤緩存，雖然這可能在緩存回收和從新分配時犧牲一些 性能。全部的磁盤讀寫操做都會通過這個緩存，這不太可能被繞開除非直接使用I/O。因此雖然每一個程序都在本身的線程裏只緩存了一份數據，但在操做系統的緩 存裏還有一份，這等於存了兩份數據。

另外再來討論一下JVM,如下兩個事實是衆所周知的：

•Java對象佔用空間是很是大的，差很少是要存儲的數據的兩倍甚至更高。

•隨着堆中數據量的增長，垃圾回收回變的愈來愈困難。

基於以上分析，若是把數據緩存在內存裏，由於須要存儲兩份，不得不使用兩倍的內存空間，Kafka基於JVM，又不得不將空間再次加倍,再加上要避 免GC帶來的性能影響，在一個32G內存的機器上，不得不使用到28-30G的內存空間。而且當系統重啓的時候，又必需要將數據刷到內存中（ 10GB 內存差很少要用10分鐘），就算使用冷刷新（不是一次性刷進內存，而是在使用數據的時候沒有就刷到內存）也會致使最初的時候新能很是慢。可是使用文件系 統，即便系統重啓了，也不須要刷新數據。使用文件系統也簡化了維護數據一致性的邏輯。

因此與傳統的將數據緩存在內存中而後刷到硬盤的設計不一樣，Kafka直接將數據寫到了文件系統的日誌中。

常量時間的操做效率

在大多數的消息系統中，數據持久化的機制每每是爲每一個cosumer提供一個B樹或者其餘的隨機讀寫的數據結構。B樹固然是很棒的，可是也帶了一些 代價：好比B樹的複雜度是O(log N)，O(log N)一般被認爲就是常量複雜度了，但對於硬盤操做來講並不是如此。磁盤進行一次搜索須要10ms，每一個硬盤在同一時間只能進行一次搜索，這樣併發處理就成了 問題。雖然存儲系統使用緩存進行了大量優化，可是對於樹結構的性能的觀察結果卻代表，它的性能每每隨着數據的增加而線性降低，數據增加一倍，速度就會下降 一倍。

直觀的講，對於主要用於日誌處理的消息系統，數據的持久化能夠簡單的經過將數據追加到文件中實現，讀的時候從文件中讀就行了。這樣作的好處是讀和寫都是 O(1) 的，而且讀操做不會阻塞寫操做和其餘操做。這樣帶來的性能優點是很明顯的，由於性能和數據的大小沒有關係了。

既然可使用幾乎沒有容量限制（相對於內存來講）的硬盤空間創建消息系統，就能夠在沒有性能損失的狀況下提供一些通常消息系統不具有的特性。好比， 通常的消息系統都是在消息被消費後當即刪除，Kafka卻能夠將消息保存一段時間（好比一星期），這給consumer提供了很好的機動性和靈活性，這點 在從此的文章中會有詳述。

 

以前討論了consumer和producer是怎麼工做的，如今來討論一下數據傳輸方面。數據傳輸的事務定義一般有如下三種級別：

  1. 最多一次: 消息不會被重複發送，最多被傳輸一次，但也有可能一次不傳輸。
  2. 最少一次: 消息不會被漏發送，最少被傳輸一次，但也有可能被重複傳輸.
  3. 精確的一次（Exactly once）: 不會漏傳輸也不會重複傳輸,每一個消息都傳輸被一次並且僅僅被傳輸一次，這是你們所指望的。

大多數消息系統聲稱能夠作到「精確的一次」，可是仔細閱讀它們的的文檔能夠看到裏面存在誤導，好比沒有說明當consumer或producer失 敗時怎麼樣，或者當有多個consumer並行時怎麼樣，或寫入硬盤的數據丟失時又會怎麼樣。kafka的作法要更先進一些。當發佈消息時，Kafka有 一個「committed」的概念，一旦消息被提交了，只要消息被寫入的分區的所在的副本broker是活動的，數據就不會丟失。關於副本的活動的概念， 下節文檔會討論。如今假設broker是不會down的。
若是producer發佈消息時發生了網絡錯誤，但又不肯定實在提交以前發生的仍是提交以後發生的，這種狀況雖然不常見，可是必須考慮進去，如今Kafka版本尚未解決這個問題，未來的版本正在努力嘗試解決。
並 不是全部的狀況都須要「精確的一次」這樣高的級別，Kafka容許producer靈活的指定級別。好比producer能夠指定必須等待消息被提交的通 知，或者徹底的異步發送消息而不等待任何通知，或者僅僅等待leader聲明它拿到了消息（followers沒有必要）。

如今從consumer的方面考慮這個問題，全部的副本都有相同的日誌文件和相同的offset，consumer維護本身消費的消息的 offset，若是consumer不會崩潰固然能夠在內存中保存這個值，固然誰也不能保證這點。若是consumer崩潰了，會有另一個 consumer接着消費消息，它須要從一個合適的offset繼續處理。這種狀況下能夠有如下選擇：

 

    • consumer能夠先讀取消息，而後將offset寫入日誌文件中，而後再處理消息。這存在一種可能就是在存儲offset後還沒處理消息就crash了，新的consumer繼續從這個offset處理，那麼就會有些消息永遠不會被處理，這就是上面說的「最多一次」。
    • consumer能夠先讀取消息，處理消息，最後記錄offset，固然若是在記錄offset以前就crash了，新的consumer會重複的消費一些消息，這就是上面說的「最少一次」。
    • 「精確一次」能夠經過將提交分爲兩個階段來解決：保存了offset後提交一次，消息處理成功以後再提交一次。可是還有個更簡單的作法：將消息的offset和消息被處理後的結果保存在一塊兒。好比用Hadoop ETL處理消息時，將處理後的結果和offset同時保存在HDFS中，這樣就能保證消息和offser同時被處理了。

Kafka在提升效率方面作了很大努力。Kafka的一個主要使用場景是處理網站活動日誌，吞吐量是很是大的，每一個頁面都會產生好屢次寫操做。讀方面，假設每一個消息只被消費一次，讀的量的也是很大的，Kafka也儘可能使讀的操做更輕量化。

咱們以前討論了磁盤的性能問題，線性讀寫的狀況下影響磁盤性能問題大約有兩個方面：太多的瑣碎的I/O操做和太多的字節拷貝。I/O問題發生在客戶端和服務端之間，也發生在服務端內部的持久化的操做中。
消息集（message set）
爲 了避免這些問題，Kafka創建了「消息集（message set）」的概念，將消息組織到一塊兒，做爲處理的單位。以消息集爲單位處理消息，比以單個的消息爲單位處理，會提高很多性能。Producer把消息集一 塊發送給服務端，而不是一條條的發送；服務端把消息集一次性的追加到日誌文件中，這樣減小了瑣碎的I/O操做。consumer也能夠一次性的請求一個消 息集。
另一個性能優化是在字節拷貝方面。在低負載的狀況下這不是問題，可是在高負載的狀況下它的影響仍是很大的。爲了不這個問題，Kafka使用了標準的二進制消息格式，這個格式能夠在producer,broker和producer之間共享而無需作任何改動。
zero copy
Broker 維護的消息日誌僅僅是一些目錄文件，消息集以固定隊的格式寫入到日誌文件中，這個格式producer和consumer是共享的，這使得Kafka能夠 一個很重要的點進行優化：消息在網絡上的傳遞。現代的unix操做系統提供了高性能的將數據從頁面緩存發送到socket的系統函數，在linux中，這 個函數是sendfile.
爲了更好的理解sendfile的好處，咱們先來看下通常將數據從文件發送到socket的數據流向：

  1. 操做系統把數據從文件拷貝內核中的頁緩存中
  2. 應用程序從頁緩存從把數據拷貝本身的內存緩存中
  3. 應用程序將數據寫入到內核中socket緩存中
  4. 操做系統把數據從socket緩存中拷貝到網卡接口緩存，從這裏發送到網絡上。


這顯然是低效率的，有4次拷貝和2次系統調用。Sendfile經過直接將數據從頁面緩存發送網卡接口緩存，避免了重複拷貝，大大的優化了性能。
在一個多consumers的場景裏，數據僅僅被拷貝到頁面緩存一次而不是每次消費消息的時候都重複的進行拷貝。這使得消息以近乎網絡帶寬的速率發送出去。這樣在磁盤層面你幾乎看不到任何的讀操做，由於數據都是從頁面緩存中直接發送到網絡上去了。
這篇文章詳細介紹了sendfile和zero-copy技術在Java方面的應用。
數據壓縮
很 多時候，性能的瓶頸並不是CPU或者硬盤而是網絡帶寬，對於須要在數據中心之間傳送大量數據的應用更是如此。固然用戶能夠在沒有Kafka支持的狀況下各自 壓縮本身的消息，可是這將致使較低的壓縮率，由於相比於將消息單獨壓縮，將大量文件壓縮在一塊兒才能起到最好的壓縮效果。
Kafka採用了端到端的 壓縮：由於有「消息集」的概念，客戶端的消息能夠一塊兒被壓縮後送到服務端，並以壓縮後的格式寫入日誌文件，以壓縮的格式發送到consumer，消息從 producer發出到consumer拿到都被是壓縮的，只有在consumer使用的時候才被解壓縮，因此叫作「端到端的壓縮」。
Kafka支持GZIP和Snappy壓縮協議。

 

Kafka Producer

消息發送

producer直接將數據發送到broker的leader(主節點)，不須要在多個節點進行分發。爲了幫助producer作到這點，全部的 Kafka節點均可以及時的告知:哪些節點是活動的，目標topic目標分區的leader在哪。這樣producer就能夠直接將消息發送到目的地了。

客戶端控制消息將被分發到哪一個分區。能夠經過負載均衡隨機的選擇，或者使用分區函數。Kafka容許用戶實現分區函數，指定分區的key，將消息 hash到不一樣的分區上(固然有須要的話，也能夠覆蓋這個分區函數本身實現邏輯).好比若是你指定的key是user id，那麼同一個用戶發送的消息都被髮送到同一個分區上。通過分區以後，consumer就能夠有目的的消費某個分區的消息。

異步發送

批量發送能夠頗有效的提升發送效率。Kafka producer的異步發送模式容許進行批量發送，先將消息緩存在內存中，而後一次請求批量發送出去。這個策略能夠配置的，好比能夠指定緩存的消息達到某 個量的時候就發出去，或者緩存了固定的時間後就發送出去（好比100條消息就發送，或者每5秒發送一次）。這種策略將大大減小服務端的I/O次數。

既然緩存是在producer端進行的，那麼當producer崩潰時，這些消息就會丟失。Kafka0.8.1的異步發送模式還不支持回調，就不能在發送出錯時進行處理。Kafka 0.9可能會增長這樣的回調函數。見Proposed Producer API.

Kafka Consumer

Kafa consumer消費消息時，向broker發出"fetch"請求去消費特定分區的消息。consumer指定消息在日誌中的偏移量（offset）， 就能夠消費從這個位置開始的消息。customer擁有了offset的控制權，能夠向後回滾去從新消費以前的消息，這是頗有意義的。

推仍是拉？

Kafka最初考慮的問題是，customer應該從brokes拉取消息仍是brokers將消息推送到consumer，也就是pull還 push。在這方面，Kafka遵循了一種大部分消息系統共同的傳統的設計：producer將消息推送到broker，consumer從broker 拉取消息。
一些消息系統好比Scribe和Apache Flume採用了push模式，將消息推送到下游的consumer。這樣作有好處也有壞處：由broker決定消息推送的速率，對於不一樣消費速率的 consumer就不太好處理了。消息系統都致力於讓consumer以最大的速率最快速的消費消息，但不幸的是，push模式下，當broker推送的 速率遠大於consumer消費的速率時，consumer恐怕就要崩潰了。最終Kafka仍是選取了傳統的pull模式。
Pull模式的另一 個好處是consumer能夠自主決定是否批量的從broker拉取數據。Push模式必須在不知道下游consumer消費能力和消費策略的狀況下決定 是當即推送每條消息仍是緩存以後批量推送。若是爲了不consumer崩潰而採用較低的推送速率，將可能致使一次只推送較少的消息而形成浪費。Pull 模式下，consumer就能夠根據本身的消費能力去決定這些策略。
Pull有個缺點是，若是broker沒有可供消費的消息，將致使consumer不斷在循環中輪詢，直到新消息到t達。爲了不這點，Kafka有個參數可讓consumer阻塞知道新消息到達(固然也能夠阻塞知道消息的數量達到某個特定的量這樣就能夠批量發送)。

消費狀態跟蹤

對消費消息狀態的記錄也是很重要的。
大部分消息系統在broker端的維護消息被消費的記錄：一個消息被分發到consumer後broker就立刻進行標記或者等待customer的通知後進行標記。這樣也能夠在消息在消費後立馬就刪除以減小空間佔用。
但 是這樣會不會有什麼問題呢？若是一條消息發送出去以後就當即被標記爲消費過的，一旦consumer處理消息時失敗了（好比程序崩潰）消息就丟失了。爲了 解決這個問題，不少消息系統提供了另一個個功能：當消息被髮送出去以後僅僅被標記爲已發送狀態，當接到consumer已經消費成功的通知後才標記爲已 被消費的狀態。這雖然解決了消息丟失的問題，但產生了新問題，首先若是consumer處理消息成功了可是向broker發送響應時失敗了，這條消息將被 消費兩次。第二個問題時，broker必須維護每條消息的狀態，而且每次都要先鎖住消息而後更改狀態而後釋放鎖。這樣麻煩又來了，且不說要維護大量的狀態 數據，好比若是消息發送出去但沒有收到消費成功的通知，這條消息將一直處於被鎖定的狀態，
Kafka採用了不一樣的策略。Topic被分紅了若干分 區，每一個分區在同一時間只被一個consumer消費。這意味着每一個分區被消費的消息在日誌中的位置僅僅是一個簡單的整數：offset。這樣就很容易標 記每一個分區消費狀態就很容易了，僅僅須要一個整數而已。這樣消費狀態的跟蹤就很簡單了。
這帶來了另一個好處：consumer能夠把 offset調成一個較老的值，去從新消費老的消息。這對傳統的消息系統來講看起來有些難以想象，但確實是很是有用的，誰規定了一條消息只能被消費一次 呢？consumer發現解析數據的程序有bug，在修改bug後再來解析一次消息，看起來是很合理的額呀！

離線處理消息

高級的數據持久化容許consumer每一個隔一段時間批量的將數據加載到線下系統中好比Hadoop或者數據倉庫。這種狀況下，Hadoop能夠將加載任務分拆，拆成每一個broker或每一個topic或每一個分區一個加載任務。Hadoop具備任務管理功能，當一個任務失敗了就能夠重啓而不用擔憂數據被從新加載，只要從上次加載的位置繼續加載消息就能夠了。

 

Kafka容許topic的分區擁有若干副本，這個數量是能夠配置的，你能夠爲每一個topci配置副本的數量。Kafka會自動在每一個個副本上備份數據，因此當一個節點down掉時數據依然是可用的。

Kafka的副本功能不是必須的，你能夠配置只有一個副本，這樣其實就至關於只有一份數據。

建立副本的單位是topic的分區，每一個分區都有一個leader和零或多個followers.全部的讀寫操做都由leader處理，通常分區的 數量都比broker的數量多的多，各分區的leader均勻的分佈在brokers中。全部的followers都複製leader的日誌，日誌中的消 息和順序都和leader中的一致。flowers向普通的consumer那樣從leader那裏拉取消息並保存在本身的日誌文件中。
許多分佈式的消息系統自動的處理失敗的請求，它們對一個節點是否
着（alive）」有着清晰的定義。Kafka判斷一個節點是否活着有兩個條件：

  1. 節點必須能夠維護和ZooKeeper的鏈接，Zookeeper經過心跳機制檢查每一個節點的鏈接。
  2. 若是節點是個follower,他必須能及時的同步leader的寫操做，延時不能過久。

符合以上條件的節點準確的說應該是「同步中的（in sync）」，而不是模糊的說是「活着的」或是「失敗的」。Leader會追蹤全部「同步中」的節點，一旦一個down掉了，或是卡住了，或是延時太 久，leader就會把它移除。至於延時多久算是「過久」，是由參數replica.lag.max.messages決定的，怎樣算是卡住了，怎是由參 數replica.lag.time.max.ms決定的。
只有當消息被全部的副本加入到日誌中時，纔算是「committed」，只有 committed的消息纔會發送給consumer，這樣就不用擔憂一旦leader down掉了消息會丟失。Producer也能夠選擇是否等待消息被提交的通知，這個是由參數request.required.acks決定的。

Kafka保證只要有一個「同步中」的節點，「committed」的消息就不會丟失。

Leader的選擇

Kafka的核心是日誌文件，日誌文件在集羣中的同步是分佈式數據系統最基礎的要素。

若是leaders永遠不會down的話咱們就不須要followers了！一旦leader down掉了，須要在followers中選擇一個新的leader.可是followers自己有可能延時過久或者crash，因此必須選擇高質量的 follower做爲leader.必須保證，一旦一個消息被提交了，可是leader down掉了，新選出的leader必須能夠提供這條消息。大部分的分佈式系統採用了多數投票法則選擇新的leader,對於多數投票法則，就是根據全部 副本節點的情況動態的選擇最適合的做爲leader.Kafka並非使用這種方法。

Kafaka動態維護了一個同步狀態的副本的集合（a set of in-sync replicas），簡稱ISR，在這個集合中的節點都是和leader保持高度一致的，任何一條消息必須被這個集合中的每一個節點讀取並追加到日誌中了， 纔回通知外部這個消息已經被提交了。所以這個集合中的任何一個節點隨時均可以被選爲leader.ISR在ZooKeeper中維護。ISR中有f+1個 節點，就能夠容許在f個節點down掉的狀況下不會丟失消息並正常提供服。ISR的成員是動態的，若是一個節點被淘汰了，當它從新達到「同步中」的狀態 時，他能夠從新加入ISR.這種leader的選擇方式是很是快速的，適合kafka的應用場景。

一個邪惡的想法：若是全部節點都down掉了怎麼辦？Kafka對於數據不會丟失的保證，是基於至少一個節點是存活的，一旦全部節點都down了，這個就不能保證了。
實際應用中，當全部的副本都down掉時，必須及時做出反應。能夠有如下兩種選擇:

  1. 等待ISR中的任何一個節點恢復並擔任leader。
  2. 選擇全部節點中（不僅是ISR）第一個恢復的節點做爲leader.

這是一個在可用性和連續性之間的權衡。若是等待ISR中的節點恢復，一旦ISR中的節點起不起來或者數據都是了，那集羣就永遠恢復不了了。若是等待 ISR意外的節點恢復，這個節點的數據就會被做爲線上數據，有可能和真實的數據有所出入，由於有些數據它可能還沒同步到。Kafka目前選擇了第二種策 略，在將來的版本中將使這個策略的選擇可配置，能夠根據場景靈活的選擇。

這種窘境不僅Kafka會遇到，幾乎全部的分佈式數據系統都會遇到。

副本管理

以上僅僅以一個topic一個分區爲例子進行了討論，但實際上一個Kafka將會管理成千上萬的topic分區.Kafka儘可能的使全部分區均勻的分佈到集羣全部的節點上而不是集中在某些節點上，另外主從關係也儘可能均衡這樣每一個幾點都會擔任必定比例的分區的leader.

優化leader的選擇過程也是很重要的，它決定了系統發生故障時的空窗期有多久。Kafka選擇一個節點做爲「controller」,當發現有 節點down掉的時候它負責在游泳分區的全部節點中選擇新的leader,這使得Kafka能夠批量的高效的管理全部分區節點的主從關係。若是 controller down掉了，活着的節點中的一個會備切換爲新的controller.

 

Kafka Producer APIs

Procuder API有兩種：kafka.producer.SyncProducer和kafka.producer.async.AsyncProducer.它們都實現了同一個接口：

  1. class Producer {
  3. /* 將消息發送到指定分區 */
  4. publicvoid send(kafka.javaapi.producer.ProducerData<K,V> producerData);
  6. /* 批量發送一批消息 */
  7. publicvoid send(java.util.List<kafka.javaapi.producer.ProducerData<K,V>> producerData);
  9. /* 關閉producer */
  10. publicvoid close();
  12. }

 

Producer API提供瞭如下功能：

  1. 能夠將多個消息緩存到本地隊列裏，而後異步的批量發送到broker，能夠經過參數producer.type=async作到。緩存的大小能夠經過一些參數指定：queue.time和batch.size。一個後臺線程（(kafka.producer.async.ProducerSendThread）從隊列中取出數據並讓kafka.producer.EventHandler將消息發送到broker，也能夠經過參數event.handler定製handler，在producer端處理數據的不一樣的階段註冊處理器，好比能夠對這一過程進行日誌追蹤，或進行一些監控。只需實現kafka.producer.async.CallbackHandler接口，並在callback.handler中配置。
  2. 本身編寫Encoder來序列化消息，只需實現下面這個接口。默認的Encoder是kafka.serializer.DefaultEncoder。
    1. interface Encoder<T> {
    2. public Message toMessage(T data);
    3. }
  3. 提供了基於Zookeeper的broker自動感知能力，能夠經過參數zk.connect實現。若是不使用Zookeeper，也可使用broker.list參數指定一個靜態的brokers列表，這樣消息將被隨機的發送到一個broker上，一旦選中的broker失敗了，消息發送也就失敗了。
  4. 經過分區函數kafka.producer.Partitioner類對消息分區。
    1. interface Partitioner<T> {
    2. int partition(T key, int numPartitions);
    3. }
    分區函數有兩個參數：key和可用的分區數量，從分區列表中選擇一個分區並返回id。默認的分區策略是hash(key)%numPartitions.若是key是null,就隨機的選擇一個。能夠經過參數partitioner.class定製分區函數。

KafKa Consumer APIs

Consumer API有兩個級別。低級別的和一個指定的broker保持鏈接，並在接收完消息後關閉鏈接，這個級別是無狀態的，每次讀取消息都帶着offset。

高級別的API隱藏了和brokers鏈接的細節，在沒必要關心服務端架構的狀況下和服務端通訊。還能夠本身維護消費狀態，並能夠經過一些條件指定訂閱特定的topic,好比白名單黑名單或者正則表達式。

低級別的API

  1. class SimpleConsumer {
  3. /*向一個broker發送讀取請求並獲得消息集 */
  4. public ByteBufferMessageSet fetch(FetchRequest request);
  6. /*向一個broker發送讀取請求並獲得一個相應集 */
  7. public MultiFetchResponse multifetch(List<FetchRequest> fetches);
  9. /**
  10. * 獲得指定時間以前的offsets
  11. * 返回值是offsets列表，以倒序排序
  12. * @param time: 時間，毫秒,
  13. * 若是指定爲OffsetRequest$.MODULE$.LATIEST_TIME(), 獲得最新的offset.
  14. * 若是指定爲OffsetRequest$.MODULE$.EARLIEST_TIME(),獲得最老的offset.
  15. */
  16. publiclong[] getOffsetsBefore(String topic, int partition, long time, int maxNumOffsets);
  17. }

低級別的API是高級別API實現的基礎，也是爲了一些對維持消費狀態有特殊需求的場景，好比Hadoop consumer這樣的離線consumer。

 

高級別的API

  1. /* 建立鏈接 */
  2. ConsumerConnector connector = Consumer.create(consumerConfig);
  4. interface ConsumerConnector {
  6. /**
  7. * 這個方法能夠獲得一個流的列表，每一個流都是MessageAndMetadata的迭代，經過MessageAndMetadata能夠拿到消息和其餘的元數據（目前以後topic）
  8. * Input: a map of <topic, #streams>
  9. * Output: a map of <topic, list of message streams>
  10. */
  11. public Map<String,List<KafkaStream>> createMessageStreams(Map<String,Int> topicCountMap);
  13. /**
  14. * 你也能夠獲得一個流的列表，它包含了符合TopicFiler的消息的迭代，
  15. * 一個TopicFilter是一個封裝了白名單或黑名單的正則表達式。
  16. */
  17. public List<KafkaStream> createMessageStreamsByFilter(
  18. TopicFilter topicFilter, int numStreams);
  20. /* 提交目前消費到的offset */
  21. public commitOffsets()
  23. /* 關閉鏈接 */
  24. public shutdown()
  25. }

 

這個API圍繞着由KafkaStream實現的迭代器展開，每一個流表明一系列從一個或多個分區多和broker上匯聚來的消息，每一個流由一個線程 處理，因此客戶端能夠在建立的時候經過參數指定想要幾個流。一個流是多個分區多個broker的合併，可是每一個分區的消息只會流向一個流。

每調用一次createMessageStreams都會將consumer註冊到topic上，這樣consumer和brokers之間的負載 均衡就會進行調整。API鼓勵每次調用建立更多的topic流以減小這種調整。createMessageStreamsByFilter方法註冊監聽可 以感知新的符合filter的tipic。

 

消息格式

消息由一個固定長度的頭部和可變長度的字節數組組成。頭部包含了一個版本號和CRC32校驗碼。

 

[plain] view plain copy print ?
  1. /**
  2. * 具備N個字節的消息的格式以下
  3. *
  4. * 若是版本號是0
  5. *
  6. * 1. 1個字節的 "magic" 標記
  7. *
  8. * 2. 4個字節的CRC32校驗碼
  9. *
  10. * 3. N - 5個字節的具體信息
  11. *
  12. * 若是版本號是1
  13. *
  14. * 1. 1個字節的 "magic" 標記
  15. *
  16. * 2.1個字節的參數容許標註一些附加的信息好比是否壓縮了，解碼類型等
  17. *
  18. * 3.4個字節的CRC32校驗碼
  19. *
  20. * 4. N - 6 個字節的具體信息
  21. *
  22. */

 

日誌

一個叫作「my_topic」且有兩個分區的的topic,它的日誌有兩個文件夾組成，my_topic_0和my_topic_1,每一個文件夾裏放着具體的數據文件，每一個數據文件都是一系列的日誌實體，每一個日誌實體有一個4個字節的整數N標註消息的長度，後邊跟着N個字節的消息。每 個消息均可以由一個64位的整數offset標註，offset標註了這條消息在發送到這個分區的消息流中的起始位置。每一個日誌文件的名稱都是這個文件第 一條日誌的offset.因此第一個日誌文件的名字就是00000000000.kafka.因此每相鄰的兩個文件名字的差就是一個數字S,S差很少就是 配置文件中指定的日誌文件的最大容量。

消息的格式都由一個統一的接口維護，因此消息能夠在producer,broker和consumer之間無縫的傳遞。存儲在硬盤上的消息格式以下所示：

  1. 消息長度: 4 bytes (value: 1+4+n)
  2. 版本號: 1 byte
  3. CRC校驗碼: 4 bytes
  4. 具體的消息: n bytes


 

寫操做

消息被不斷的追加到最後一個日誌的末尾，當日志的大小達到一個指定的值時就會產生一個新的文件。對於寫操做有兩個參數，一個規定了消息的數量達到這 個值時必須將數據刷新到硬盤上，另一個規定了刷新到硬盤的時間間隔，這對數據的持久性是個保證，在系統崩潰的時候只會丟失必定數量的消息或者一個時間段 的消息。

讀操做

 

讀操做須要兩個參數：一個64位的offset和一個S字節的最大讀取量。S一般比單個消息的大小要大，但在一些個別消息比較大的狀況下，S會小於 單個消息的大小。這種狀況下讀操做會不斷重試，每次重試都會將讀取量加倍，直到讀取到一個完整的消息。能夠配置單個消息的最大值，這樣服務器就會拒絕大小 超過這個值的消息。也能夠給客戶端指定一個嘗試讀取的最大上限，避免爲了讀到一個完整的消息而無限次的重試。

在實際執行讀取操縱時，首先須要定位數據所在的日誌文件，而後根據offset計算出在這個日誌中的offset(前面的的offset是整個分區 的offset),而後在這個offset的位置進行讀取。定位操做是由二分查找法完成的，Kafka在內存中爲每一個文件維護了offset的範圍。

 

下面是發送給consumer的結果的格式：

MessageSetSend (fetch result)

total length     : 4 bytes
error code       : 2 bytes
message 1        : x bytes
...
message n        : x bytes

MultiMessageSetSend (multiFetch result)

total length       : 4 bytes
error code         : 2 bytes
messageSetSend 1
...
messageSetSend n

刪除

日誌管理器容許定製刪除策略。目前的策略是刪除修改時間在N天以前的日誌（按時間刪除），也可使用另一個策略：保留最後的N GB數據的策略(按大小刪除)。爲了不在刪除時阻塞讀操做，採用了copy-on-write形式的實現，刪除操做進行時，讀取操做的二分查找功能實際 是在一個靜態的快照副本上進行的，這相似於Java的CopyOnWriteArrayList。

可靠性保證

日誌文件有一個可配置的參數M，緩存超過這個數量的消息將被強行刷新到硬盤。一個日誌矯正線程將循環檢查最新的日誌文件中的消息確認每一個消息都是合 法的。合法的標準爲：全部文件的大小的和最大的offset小於日誌文件的大小，而且消息的CRC32校驗碼與存儲在消息實體中的校驗碼一致。若是在某個 offset發現不合法的消息，從這個offset到下一個合法的offset之間的內容將被移除。

有兩種狀況必須考慮：1，當發生崩潰時有些數據塊未能寫入。2，寫入了一些空白數據塊。第二種狀況的緣由是，對於每一個文件，操做系統都有一個 inode（inode是指在許多「類Unix文件系統」中的一種數據結構。每一個inode保存了文件系統中的一個文件系統對象,包括文件、目錄、大小、 設備文件、socket、管道, 等等），但沒法保證更新inode和寫入數據的順序，當inode保存的大小信息被更新了，但寫入數據時發生了崩潰，就產生了空白數據塊。CRC校驗碼可 以檢查這些塊並移除，固然由於崩潰而未寫入的數據塊也就丟失了。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程