日誌平臺（網關層） - 基於Openresty+ELKF+Kafka

時間 2019-11-10

標籤日誌平臺網關基於 openresty+elkf+kafka openresty elkf kafka 欄目 Kafka 简体版

原文原文鏈接

背景介紹

一、問題現狀與嘗試

沒有作日誌記錄的線上系統，絕對是給系統運維人員留下的坑。尤爲是先後端分離的項目，後端的接口日誌能夠解決對接、測試和運維時的不少問題。以前項目上發佈的接口都是經過Oracle Service Bus(OSB)來作統一編排，在編排時加上日誌記錄，並將接口日誌存儲到數據庫中。最後基於接口日誌數據開發日誌平臺，來統一的接口日誌分析。
但咱們總不能爲了記錄日誌而使用OSB，這樣很不自由。今年咱們有不少後臺接口使用Spring來開發，後臺程序的部署環境也不侷限於Oracle中間件的環境。當某些場景時，脫離了OSB，咱們該如何記錄接口日誌，這是本文要解決的問題。html

在我寫的Spring系列的文章中，有嘗試過使用Spring的AOP來記錄日誌。在每一個項目的代碼中，定義一個記錄日誌的切面，該切面會對該項目下的全部接口作日誌記錄。
對於一個週期很長、規模很大的一個獨立項目來講，這個方案是可行的。由於項目週期很長，花個兩天作日誌記錄的AOP開發沒啥問題，並且這個日誌更契合該系統的業務特徵。
但咱們團隊所面對的開發，基本上都是數量多、週期短的一些小項目。一個項目的開發週期可能只有十天，就算每一個項目在日誌記錄上只用一天的工做量，所佔的比重也有十分之一。若是咱們每一個項目都要獨立的記錄日誌，累積的工做量也挺大的，並且重複這樣的工做很枯燥。
就像面向切面編程（AOP），在一個項目的全部接口上設置「切面」統一編程。若是咱們的能在全部的項目上設置「切面」統一編程，就能解決咱們如今的問題。這個「切面」就是網關。前端

二、方案設計

這個方案是公司內的兩位技術大佬討論出來的，這樣驚奇的想法，讓以前困擾的一切迷霧都豁然開朗了起來。我花了兩天作了個Demo，驗證方案的確行得通，下文會附上本次Demo中實戰操做的代碼。
簡單來講，全部項目接口都經過Nginx的網關，而咱們不須要在代碼層面上收集日誌，而是在Nginx上獲取想要的日誌信息，配合ELKF（Elasticsearch、Logstash、Kibana、Filebeat）的解決方案，實現統一的日誌平臺搭建：java

Nginx+Lua編程，按照咱們定義的格式，全部經過網關的接口都會留下日誌信息，寫入log文件。
Filebeat收集數據，Filebeat實時監測目標log文件，收集數據推送給Logstash。
Logstash過濾處理數據，Logstash過濾處理數據後，會將數據同時推送給Elasticsearch和Kafka。
Elasticsearch+Kibana，Elasticsearch做爲數據的搜索引擎，並且利用Kibana的可視化界面，將日誌數據以報表的形式顯示出來。
Kafka消息隊列中間件，日誌的數據被推送到Kafka上以後發佈消息，而全部訂閱者就能從隊列中讀數據。本次就是寫程序實時的讀取隊列中的數據，存入數據庫。

三、系統環境

在本次Demo中，因爲資源限制，全部的產品服務都將部署在一臺服務器上，服務器上的相關環境以下：nginx

配置項	環境配置信息
服務器	阿里雲服務器ECS（公網：47.96.238.21 ，私網：172.16.187.25）
服務器配置	2 vCPU + 4 GB內存
JDK版本	JDK 1.8.0_181
操做系統	CentOS 7.4 64位
OpenResty	1.13.6.2
Filebeat	6.2.4
Elasticsearch	6.2.4
Logstash	6.2.4
Kibana	6.2.4
Kafka	2.10-0.10.2.1

基於OpenResty的日誌記錄

OpenResty® 是一個基於 Nginx 與 Lua 的高性能 Web 平臺，其內部集成了大量精良的 Lua 庫、第三方模塊以及大多數的依賴項。用於方便地搭建可以處理超高併發、擴展性極高的動態 Web 應用、Web 服務和動態網關。
咱們選擇OpenResty的目的有兩個：（1）使用Lua編程，能夠在Nginx上更好的拿到想要的日誌信息；（2）系統其它功能模塊的集成，例如Jwt的集成，可參考同事寫的文章《Nginx實現JWT驗證-基於OpenResty實現》。web

一、OpenResty安裝

在安裝OpenResty以前須要先安裝好依賴庫，OpenResty 依賴庫有： perl 5.6.1+, libreadline, libpcre, libssl。咱們是CentOS系統，能夠直接yum來安裝。spring

[root@Kerry ~]# yum install readline-devel pcre-devel openssl-devel perl

接下來咱們在當前CentOS系統上使用新的官方 yum 源sql

[root@Kerry ~]# yum install yum-utils
[root@Kerry ~]# yum-config-manager --add-repo https://openresty.org/package/centos/openresty.repo

這時咱們就能夠直接安裝OpenResty數據庫

[root@Kerry ~]# yum install openresty
[root@Kerry ~]# yum install openresty-resty

這樣OpenResty就安裝完成了，默認狀況下程序會被安裝到 /usr/local/openresty 目錄apache

# 可查看安裝成功
[root@Kerry ~]# cd /usr/local/openresty/bin/
[root@Kerry bin]# ./openresty -v
nginx version: openresty/1.13.6.2

# 設置環境變量
[root@Kerry sbin]# vi /etc/profile
# 在文件最後面加上 export PATH=${PATH}:/usr/local/openresty/nginx/sbin
[root@Kerry sbin]# source /etc/profile

二、記錄Nginx日誌

OpenResty 安裝以後就有配置文件及相關的目錄的，爲了工做目錄與安裝目錄互不干擾，咱們單獨建一個工做目錄。我在根目錄下新建了 /openrestyTest/v1/ 的文件夾，並在該目錄下建立 logs 和 conf 子目錄分別用於存放日誌和配置文件。編程

[root@Kerry ~]# mkdir /openrestyTest /openrestyTest/v1 /openrestyTest/v1/conf /openrestyTest/v1/logs
[root@Kerry ~]# cd /openrestyTest/v1/conf/
# 建立並編輯 nginx.conf
[root@Kerry conf]# vi nginx.conf

在nginx.conf中複製如下文本做爲測試

worker_processes  1;        #nginx worker 數量
error_log logs/error.log;   #指定錯誤日誌文件路徑
events {
    worker_connections 1024;
}

http {
    server {
        #監聽端口，若你的6699端口已經被佔用，則須要修改
        listen 6699;
        location / {
            default_type text/html;

            content_by_lua_block {
                ngx.say("HelloWorld")
            }
        }
    }
}

該語法是基於Lua，監聽6699端口，輸出HelloWorld。咱們如今啓動Openresty中的Nginx。

[root@Kerry ~]# /usr/local/openresty/nginx/sbin/nginx -p '/openrestyTest/v1/' -c conf/nginx.conf
# 因爲配置或環境變量，也能夠直接使用
[root@Kerry ~]# nginx -p '/openrestyTest/v1/' -c conf/nginx.conf
[root@Kerry conf]# curl http://localhost:6699
HelloWorld

訪問該端口地址，成功的顯示HelloWorld。我提早在本服務器的Tomcat上部署了一個接口，端口是8080。個人想法是將8080反向代理成9000，將全部經過8080端口的服務的日誌信息獲取到，並輸出到本地的log文件中。
我暫時須要記錄的日誌內容包括：接口地址，請求內容，請求時間，響應內容，響應時間等。代碼寫好了，直接替換 /openrestyTest/v1/conf/nginx.conf 的文件內容。

worker_processes  1;
error_log logs/error.log;

events {
    worker_connections 1024;
}

http {
log_format myformat '{"status":"$status","requestTime":"$requestTime","responseTime":"$responseTime","requestURL":"$requestURL","method":"$method","requestContent":"$request_body","responseContent":"$responseContent"}';
access_log logs/test.log myformat;

upstream tomcatTest {
    server 47.96.238.21:8080;
}

server {
        server_name 47.96.238.21;
        listen 9000;
        # 默認讀取 body
        lua_need_request_body on;

        location / {
                log_escape_non_ascii off;
                proxy_pass  http://tomcatTest;
                set $requestURL '';
                set $method '';
                set $requestTime '';
                set $responseTime '';
                set $responseContent '';

                body_filter_by_lua '
                        ngx.var.requestTime=os.date("%Y-%m-%d %H:%M:%S")

                        ngx.var.requestURL=ngx.var.scheme.."://"..ngx.var.server_name..":"..ngx.var.server_port..ngx.var.request_uri
                        ngx.var.method=ngx.var.request_uri

                        local resp_body = string.sub(ngx.arg[1], 1, 1000)
                        ngx.ctx.buffered = (ngx.ctx.buffered or"") .. resp_body
                        if ngx.arg[2] then
                                ngx.var.responseContent = ngx.ctx.buffered
                        end

                        ngx.var.responseTime=os.date("%Y-%m-%d %H:%M:%S")
                  ';

        }

    }
}

從新啓動Nginx，而後進行驗證

[root@Kerry conf]# nginx -p '/openrestyTest/v1/' -c conf/nginx.conf -s reload

我準備好的接口地址爲：http://47.96.238.21:8080/springboot-demo/hello ，該接口返回的結果都是「Hello!Spring boot」。
如今用POST方式調用接口http://47.96.238.21:9000/springboot-demo/hello，Request中使用application/json方式輸入內容：「segmentFault《日誌平臺（網關層） - 基於Openresty+ELKF+Kafka》」。而後查看logs文件夾，發現多了個 test.log 文件，咱們查看該文件。就能夠發現，當咱們每調用一次接口，就會同步的輸出接口日誌到該文件中。

[root@Kerry conf]#  tail -500f /openrestyTest/v1/logs/test.log
{"status":"200","requestTime":"2018-10-11 18:09:02","responseTime":"2018-10-11 18:09:02","requestURL":"http://47.96.238.21:9000/springboot-demo/hello","method":"/springboot-demo/hello","requestContent":"segmentFault《日誌平臺（網關層） - 基於Openresty+ELKF+Kafka》","responseContent":"Hello!Spring boot!"}

到此爲止，提取通過Nginx網關的接口信息，並將其寫入日誌文件就完成了，全部的接口日誌都寫入了 test.log 文件中。

E+L+K+F=日誌數據處理

ELKF是 Elastic + Logstash + Kibana + FileBeat 四個組件的組合，可能ELK對於你們來講更熟悉，ELKF只不過多了Filebeat，它們都是Elastic公司推出的開源產品。恰好這幾天Elastic公司成功上市，掀起了一波ELKF產品討論的熱潮。
原ELK架構中，Logstash負責收集日誌信息並上報，但後來Elastic公司又推出了Filebeat，你們發現Filebeat在日誌文件收集上效果更好，就只讓Logstash負責日誌的處理和上報了。在這個系統中，Elastic充當一個搜索引擎，Logstash爲日誌分析上報系統，FileBeat爲日誌文件收集系統，Kibana爲此係統提供可視化的Web界面。

一、Filebeat安裝配置

Filebeat：輕量型日誌採集器，負責採集文件形式的日誌，並將採集來的日誌推送給logstash進行處理。

[root@Kerry ~]# cd /u01/install/
[root@Kerry install]# wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.2.4-x86_64.rpm
[root@Kerry install]# yum localinstall -y filebeat-6.2.4-x86_64.rpm

安裝完成後，咱們開始配置Filebeat來採集日誌，並推送給Logstash。

[root@Kerry install]# cd /etc/filebeat/
[root@Kerry filebeat]# vi filebeat.yml

該filebeat.yml是filebeat的配置文件，裏面大部分的模塊都被註釋了，本次配置放開的代碼有；

filebeat.prospectors:
- type: log
  enabled: true
  paths:
    - /openrestyTest/v1/logs/*.log
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
setup.template.settings:
  index.number_of_shards: 3
output.logstash:
  hosts: ["47.96.238.21:5044"]

監聽 /openrestyTest/v1/logs/ 目錄下的log文件，採集的日誌信息輸出到logstash，該hosts等咱們安裝啓動了Logstash再說，先啓動Filebeat。

[root@Kerry filebeat]# cd /usr/share/filebeat/bin/
[root@Kerry bin]# touch admin.out
[root@Kerry bin]# nohup ./filebeat -e -c /etc/filebeat/filebeat.yml > admin.out &
# 查看admin.out 日誌，是否啓動成功

二、Logstash安裝配置

Logstash：日誌處理工具，負責日誌收集、轉換、解析等，並將解析後的日誌推送給ElasticSearch進行檢索。

[root@Kerry ~]# cd /u01/install/
[root@Kerry install]# wget https://artifacts.elastic.co/downloads/logstash/logstash-6.2.4.rpm
[root@Kerry install]# yum localinstall -y logstash-6.2.4.rpm
#Logstash不建議用root啓動
[root@Kerry install]# group add logstash
[root@Kerry install]# useradd -g logstash logstash
[root@Kerry install]# passwd logstash
# 設置密碼
[root@Kerry install]# su logstash
[root@Kerry install]# mkdir -pv /data/logstash/{data,logs}
[root@Kerry install]# chown -R logstash.logstash /data/logstash/
[root@Kerry install]# vi /etc/logstash/conf.d/logstash.conf

建立並編輯/etc/logstash/conf.d/logstash.conf 文件，配置以下：

input {
  beats {
    port => 5044
    codec => plain {
          charset => "UTF-8"
    }
  }
}

output {
  elasticsearch {
    hosts => "47.96.238.21:9200"
    manage_template => false
    index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
    document_type => "%{[@metadata][type]}"
  }
}

一、input：是指Logstash的數據來源，啓動後使用5044來監聽，是否很熟悉，就是上節Filebeat推送日誌的hosts。
二、output；是Logstash輸出數據的位置，咱們這裏定義爲elasticsearch，下文中會說到，用於ELK架構中的日誌分析

接下來咱們修改/etc/logstash/logstash.yml

#vim /etc/logstash/logstash.yml
path.data: /data/logstash/data
path.logs: /data/logstash/logs

如今能夠啓動Logstash了

[root@Kerry install]# su logstash
[logstash@Kerry root]$ cd /usr/share/logstash/bin/
[logstash@Kerry bin]$ touch admin.out
[logstash@Kerry bin]$ nohup ./logstash -f /etc/logstash/conf.d/logstash.conf >admin.out &

三、Elasticsearch安裝配置

ElasticSearch：是一個分佈式的RESTful風格的搜索和數據分析引擎，同時還提供了集中存儲功能，它主要負責將logstash抓取來的日誌數據進行檢索、查詢、分析等。

[root@Kerry ~]# cd /u01/install/
[root@Kerry install]# wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.2.4.rpm
[root@Kerry install]# yum localinstall -y elasticsearch-6.2.4.rpm
#Elasticsearch不建議用root啓動
[root@Kerry install]# group add elsearch
[root@Kerry install]# useradd -g elsearch elsearch
[root@Kerry install]# passwd elsearch
# 設置密碼
[root@Kerry install]# su elsearch
[elsearch@Kerry bin]$  mkdir -pv /data/elasticsearch/{data,logs}
[elsearch@Kerry bin]$  chown -R elsearch.elsearch /data/elasticsearch/
[elsearch@Kerry bin]$  vi /etc/elasticsearch/elasticsearch.yml
path.data: /data/elasticsearch/data
path.logs: /data/elasticsearch/logs
network.host: 0.0.0.0
http.port: 9200

若是想要外網能訪問，host就必需要設成0.0.0.0。Elasticsearch的啓動以下

[root@Kerry install]# su elsearch
[elsearch@Kerry bin]$ cd /usr/share/elasticsearch/bin/
[elsearch@Kerry bin]$ ./elasticsearch -d
# -d 保證後臺啓動

四、Kibana安裝配置

Kibana：Web前端，能夠將ElasticSearch檢索後的日誌轉化爲各類圖表，爲用戶提供數據可視化支持。

[root@Kerry ~]# cd /u01/install/
[root@Kerry install]# wget https://artifacts.elastic.co/downloads/kibana/kibana-6.2.4-x86_64.rpm
[root@Kerry install]# yum localinstall -y kibana-6.2.4-x86_64.rpm
[root@Kerry install]# vi /etc/kibana/kibana.yml
server.port: 5601
server.host: "0.0.0.0"
elasticsearch.url: "http://47.96.238.21:9200"

一樣的，host爲0.0.0.0，保證外網能訪問。Kibana只做爲前端展現，日誌數據的獲取仍是藉助於elasticsearch，因此這裏配置了elasticsearch.url。接着啓動Kibana，就能經過頁面看到日誌的報表。

[root@Kerry ~]# cd /usr/share/kibana/bin/
[root@Kerry bin]# touch admin.out
[root@Kerry bin]# nohup ./kibana >admin.out &

咱們在瀏覽器上訪問 http://47.96.238.21:5601/ ，正常來講就能訪問Kibana的頁面。若是 ELKF一整套配置沒問題，就能在Kibana的頁面上實時的看到全部日誌信息。

從Kafka到數據庫

在拿到日誌的數據後，經過Elasticsearch和Kibana，已經完成了一個日誌查看的平臺。但咱們本身項目內部也已經開發了日誌平臺，但願把這些日誌接入到以前的日誌平臺中；或者咱們但願定製化一個更符合實際使用的日誌平臺，這些都須要把拿到的日誌數據存儲到數據庫裏。
但全部日誌的記錄，很明顯處於高併發環境，很容易因爲來不及同步處理，致使請求發生堵塞。好比說，大量的insert，update之類的請求同時到達數據庫，直接致使無數的行鎖表鎖，甚至最後請求會堆積過多，從而觸發too many connections錯誤。經過使用消息隊列，咱們能夠異步處理請求，從而緩解系統的壓力。在比對市場上開源的消息中間件後，我選擇了Kafka。
Apache Kafka是一個分佈式的發佈-訂閱消息系統，可以支撐海量數據的數據傳遞。在離線和實時的消息處理業務系統中，Kafka都有普遍的應用。Kafka將消息持久化到磁盤中，並對消息建立了備份保證了數據的安全。Kafka主要特色是基於Pull的模式來處理消息消費，追求高吞吐量，一開始的目的就是用於日誌收集和傳輸。0.8版本開始支持複製，不支持事務，對消息的重複、丟失、錯誤沒有嚴格要求，適合產生大量數據的互聯網服務的數據收集業務。

Broker：Kafka的broker是無狀態的，broker使用Zookeeper維護集羣的狀態。Leader的選舉也由Zookeeper負責。
Zookeeper：Zookeeper負責維護和協調broker。當Kafka系統中新增了broker或者某個broker發生故障失效時，由ZooKeeper通知生產者和消費者。生產者和消費者依據Zookeeper的broker狀態信息與broker協調數據的發佈和訂閱任務。
Producer：生產者將數據推送到broker上，當集羣中出現新的broker時，全部的生產者將會搜尋到這個新的broker，並自動將數據發送到這個broker上。
Consumer：由於Kafka的broker是無狀態的，因此consumer必須使用partition
offset來記錄消費了多少數據。若是一個consumer指定了一個topic的offset，意味着該consumer已經消費了該offset以前的全部數據。consumer能夠經過指定offset，從topic的指定位置開始消費數據。consumer的offset存儲在Zookeeper中。

一、Kafka安裝與配置

咱們開始Kafka的安裝和啓動

# 安裝
[root@Kerry ~]# cd /u01/install/
[root@Kerry install]# wget http://apache.fayea.com/kafka/0.10.2.1/kafka_2.10-0.10.2.1.tgz
[root@Kerry install]# tar -zvxf kafka_2.10-0.10.2.1.tgz -C /usr/local/
[root@Kerry install]# cd /usr/local/
[root@Kerry local]# mv kafka_2.10-0.10.2.1 kafka
# 啓動
[root@Kerry local]# cd /usr/local/kafka/bin/
[root@Kerry bin]# ./zookeeper-server-start.sh -daemon ../config/zookeeper.properties
[root@Kerry bin]# touch admin.out
[root@Kerry bin]# nohup ./kafka-server-start.sh ../config/server.properties >admin.out &

建立一個topic，命名爲 kerry

[root@Kerry bin]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kerry
# topic建立成功，下面查看一下
[root@Kerry bin]# ./kafka-topics.sh --list --zookeeper localhost:2181
kerry

咱們往這個topic中發送信息

[root@Kerry bin]# ./kafka-console-producer.sh --broker-list localhost:9092 --topic kerry
Hello Kerry!this is the message for test

咱們再開一個窗口，從topic中接受消息

[root@Kerry bin]# ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic kerry --from-beginning
Hello Kerry!this is the message for test
# 能成功接收到

二、生產者:Logstash

Kafka已經安裝好了，也建好了topic，而我但願往topic中發送消息的對象（生產者）是Logstash。即Logstash從Filebeat中獲取數據後，除了輸出給Elasticsearch之外，還輸出給Logstash,Logstash做爲Kafka的生產者。
這裏須要修改一下Logstash的配置文件，在output中再加上kafka的信息

vi /etc/logstash/conf.d/logstash.conf
input {
  beats {
    port => 5044
    codec => plain {
          charset => "UTF-8"
    }
  }
}

output {
  elasticsearch {
    hosts => "47.96.238.21:9200"
    manage_template => false
    index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
    document_type => "%{[@metadata][type]}"
  }
  kafka {
    bootstrap_servers => "localhost:9092"    #生產者
    topic_id => "kerry"    #設置寫入kafka的topic
    compression_type => "snappy"
    codec => plain {
            format => "%{message}"
        }
  }
}

重啓Logstash

[root@Kerry bin]# cd /usr/share/logstash/bin
[root@Kerry bin]# ps -ef|grep logstash 
# kill 進程
[root@Kerry bin]# nohup ./logstash -f /etc/logstash/conf.d/logstash.conf >admin.out &

咱們再用POST方式調用以前的測試接口http://47.96.238.21:9000/springboot-demo/hello，請求request爲：「這是對kafka的測試」。而後再查看從topic中接受消息

[root@Kerry bin]#./kafka-console-consumer.sh --zookeeper localhost:2181 --topic kerry --from-beginning
{"status":"200","requestTime":"2018-10-12 09:40:02","responseTime":"2018-10-12 09:40:02","requestURL":"http://47.96.238.21:9000/springboot-demo/hello","method":"/springboot-demo/hello","requestContent":"這是對kafka的測試","responseContent":"Hello!Spring boot!"}

能夠成功的接收到推送過來的日誌消息

三、消費者:Springboot編程

日誌已經能夠保證可以持續不斷的推送到Kafka中，那麼就須要有消費者訂閱這些消息，寫入到數據庫。我用Spring boot寫了個程序，用來訂閱Kafka的日誌，重要代碼以下：
一、application.yml

spring:
  # kafka
  kafka:
    # kafka服務器地址(能夠多個)
    bootstrap-servers: 47.96.238.21:9092
    consumer:
      # 指定一個默認的組名
      group-id: kafka1
      # earliest:當各分區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，從頭開始消費
      # latest:當各分區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，消費新產生的該分區下的數據
      # none:topic各分區都存在已提交的offset時，從offset後開始消費；只要有一個分區不存在已提交的offset，則拋出異常
      auto-offset-reset: earliest
      # key/value的反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      # key/value的序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 批量抓取
      batch-size: 65536
      # 緩存容量
      buffer-memory: 524288
      # 服務器地址
      bootstrap-servers: 47.96.238.21:9092

二、POM.xml

<dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>1.0.6.RELEASE</version>
        </dependency>

三、KafkaController.java

package df.log.kafka.nginxlog.controller;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.naming.InitialContext;
import javax.sql.DataSource;
import java.sql.Connection;


@RestController
@EnableAutoConfiguration
public class KafkaController {

    @RequestMapping("/hello")
    public String hello(){
        return "Hello!Kerry. This is NginxLog program";
    }
    /**
     * 監聽信息
     */
    @KafkaListener(topics = "kerry" )
    public void receive(ConsumerRecord<?, ?> consumer) {
        // kafkaLog 就是獲取到的日誌信息
        String kafkaLog = (String) consumer.value();
        System.out.println("收到一條消息："+kafkaLog);
        // 存入數據庫的代碼省略
    }

}

當程序部署以後，@KafkaListener(topics = "kerry") 會持續監聽topics 爲kerry的消息。咱們再調用以前的測試接口，會發現新的接口日誌會被持續監聽到，在控制檯上打印出來，並存入數據庫。

尾聲

本次操做文檔是記錄Demo的過程，不少地方並不成熟，例如：如何在 Nginx+Lua 時獲取更加全面的日誌信息；在Logstash上對日誌進行再加工；寫出漂亮的Spring boot 代碼，使得可以很平緩的作寫入數據庫，用好Kibana的圖表等等。咱們下一步就是在項目的生產環境上正式的搭建日誌平臺，咱們已經有了rancher環境，這套架構計劃用微服務的方式實現。後續的搭建文檔會持續更新。

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。