摘要：是一個日志收集器，支持非常多的輸入源和輸出源。這個庫支持展開文件路徑，而且會記錄一個叫的數據庫文件來跟蹤被監聽的日志文件的當前讀取位置。

對于Zookeeper我們用一條簡單的命令來測試一下：

echo ruok|nc localhost 2181

你應該可以看到：

imok

Kafka 是由 Linked 開發并且開源的一套分布式的流平臺，它類似于一個消息隊列。 Kafka 的優勢就是適合構建實時的流數據管道，并且可靠地獲取系統和應用程序之間的數據， 還可以實時的對數據流進行轉換。Kafka 的使用場景很多， 特別是在需要高吞吐量的系統上。首先來理解一下 Kafka 的幾個基本的概念：

Topic， Kafka 將消息進行分類，每一類的消息稱之為一個主題(Topic).

Producer， 發布消息的對象稱之為主題生產者(Producer)

Consumer， 訂閱消息并處理消息的對象稱之為主題消費者(Consumers)

Broker， 已發布的消息保存在一組服務器中，稱之為Kafka集群。集群中的每一個服務器都是一個代理(Broker)。 消費者可以訂閱一個或多個主題（topic），并從Broker拉數據，從而消費這些已發布的消息。

我們可以登錄到 Kafka 容器中，做一些簡單的小測試，登錄到容器中的命令:

docker-compose exec kafka bash

首先來創建一個名為test的 Topic ，只有一個分區和一個備份：

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

創建好之后，可以通過運行以下命令，查看topic信息：

kafka-topics.sh --list --zookeeper zookeeper:2181

Kafka 默認有提供了一個命令行的工具用來發送消息，一行是一條消息，運行 Producer 然后輸入一些信息：

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

Kafka 也提供了一個消費消息的命令行工具，將存儲的信息輸出出來。

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

ElasticStack技術棧包括的三個軟件，而且一開始就是5.0的大版本，主要目的是為了讓其他的軟件版本號一致，方便各個軟件的對應。不過到現在已經發展到了6.0的版本了，我們這次就是用這個版本。具體的更新詳細可以查看這里Elasticsearch，這里只說說基本的概念。

Elasticsearch之所以可以方便的進行查詢過濾排序是因為和 MongoDB 一樣存儲的是整個文檔，然后還會根據文檔的內容進行索引，而且文檔的格式用的還是很友好的 JSON，比如一個簡單的用戶對象可以這么表示：

{
    "name": "Xiao Ming",
    "phone": "10086",
    "age": "25",
    "info": {
        "site" : "https://sunnyshift.com"
        "likes": ["games", "music"]
    }
}

在 Elasticsearch 中有四個名詞，索引(idnex)、類型(type)、文檔(document)、字段(field)，這四個字段類似于數據庫中的數據庫(database), 表(table), 行(row), 列(column), Elasticsearch 可以包含多個索引，每個索引可以包含多個類型，每個類型可以包含多個文檔，每個文檔可以包含多個字段，把他想成數據庫就行了。

我們來簡單操作一下，先來創建一個索引，并且插入幾條數據：

PUT /banji/xuesheng/1
{
    "name": "Xiao Ming",
    "age": "12"
}
PUT /banji/xuesheng/2
{
    "name": "Xiao Hong",
    "age": "16"
}

具體操作可以使用 Kibana 上面的 Dev Tools 來運行命令：

有了數據之后就可以來簡單的檢索一下：

這里只是介紹了一下簡單的入門使用，之后可以像一些其他的操作比如過濾、組合、全文、啥的都不在話下，甚至還可以高亮搜索結果。

Logstash 是一個日志收集器，支持非常多的輸入源和輸出源。我們簡單的整合一下 ElasticStack 這個技術棧，這里用 Nginx 的訪問日志來測試一下。Logstash 最新的是有一個 Beats 的組件來收集數據，這里先不考慮而是使用傳統的配置文件，我們會直接把配置寫到配置文件中。

Logstash 使用一個名叫 FileWatch 的 Ruby Gem 庫來監聽文件變化。這個庫支持 glob 展開文件路徑，而且會記錄一個叫 .sincedb 的數據庫文件來跟蹤被監聽的日志文件的當前讀取位置。

使用 Dcoker 后配置文件路徑在logstash/config下，可以在該目錄下放置多個配置文件，最后 Logstash 會幫我們合并成一個。先來定義一個讀取 Nginx 訪問日志的配置：

input {
    file {
        path => [ "/var/log/nginx/access.log"]
        type => "accesslog"
    }
}
 
output {
    # 輸出到 elasticsearch，索引名為 access-log
    elasticsearch { 
        hosts => "localhost:9200"
        index => "access-log"
        user => "elastic"
        password => "changeme"
    }
}

要注意的是這里寫的 Nginx 的路徑是相對于 Logstash 容器的，所以要把該路徑掛載到容器中。然后重啟 Logstash 就可以去 Kibana 中查看日志了, 附上 Logstash 重啟命令。如果一切正常，那么就可以去 Kibana 中日志了。

docker-compose stop logstash
docker-compose up -d logstash

Kibana 是一個可視化的 UI 界面，并且可以檢索聚合分詞搜索在 Elasticsearch 中的數據，并且還可以以非常精美的圖標來展示統計數據。我們需要在 Management 頁面中告訴 Kibana 我們剛才創建的 access-log 索引。

到這里就完成了收集 Nginx 系統日志并展示出來，對于 Kibana 還有很多用法，就不一一撰述了。