filebeat采集日志到kafka配置及使用

Filebeat是elastic公司beats系列工具中的一個，主要用于收集本地日志。

創新互聯是一家集網站建設,麻山企業網站建設,麻山品牌網站建設,網站定制,麻山網站建設報價,網絡營銷,網絡優化,麻山網站推廣為一體的創新建站企業，幫助傳統企業提升企業形象加強企業競爭力。可充分滿足這一群體相比中小企業更為豐富、高端、多元的互聯網需求。同時我們時刻保持專業、時尚、前沿，時刻以成就客戶成長自我，堅持不斷學習、思考、沉淀、凈化自己，讓我們為更多的企業打造出實用型網站。

在服務器上安裝后，filebeat會監控日志目錄或者指定的日志文件，追蹤讀取這些文件（追蹤文件的變化，不停的讀），并且轉發這些信息到配置文件中指定的輸出端（例如：elasticsearch,logstarsh或kafka）。

Filebeat使用go語言開發，使用時沒有其他依賴，比logstash-forworder輕量，不會占用部署服務器太多的資源。

filebeat的工作流程：當你開啟filebeat程序的時候，它會啟動一個或多個探測器（prospectors）去檢測你指定的日志目錄或文件，對于探測器找出的每一個日志文件，filebeat啟動收割進程（harvester），每一個收割進程讀取一個日志文件的新內容，并發送這些新的日志數據到處理程序（spooler），處理程序會集合這些事件，最后filebeat會發送集合的數據到你指定的地點。

2.配置filebeat

配置filebeat需要編輯filebeat的配置文件，不同安裝方式，配置文件的存放路徑有一些不同， 對于 rpm 和 deb的方式, 配置文件路徑的是 /etc/filebeat/filebeat.yml，對于壓縮包的方式，配置文件存在在解壓目錄下（例如：我是在home目錄下進行的解壓，那么配置文件的路徑就應該是~/filebeat-6.2.4-linux-x86_64/filebeat.yml）。

由于我的預期目標是將filebeat收集的日志發送到kafka,所以配置output就選擇了kafka。讀者可根據自己的使用場景，配置output。

例子中的配置將對/var/log目錄下所有以.log結尾的文件進行采集。

3.啟動

本文中只是為滿足需求對filebeat進行了最基本的配置。filebeat的很多重要的配置和特性并沒有體現（例如：模塊，多行消息），讀者如果需要更深入的了解請參考： 。

歡迎大家在評論區討論使用過程的心得和疑惑。

golang使用Nsq

1. 介紹

最近在研究一些消息中間件，常用的MQ如RabbitMQ,ActiveMQ,Kafka等。NSQ是一個基于Go語言的分布式實時消息平臺，它基于MIT開源協議發布，由bitly公司開源出來的一款簡單易用的消息中間件。

官方和第三方還為NSQ開發了眾多客戶端功能庫，如官方提供的基于HTTP的nsqd、Go客戶端go-nsq、Python客戶端pynsq、基于Node.js的JavaScript客戶端nsqjs、異步C客戶端libnsq、Java客戶端nsq-java以及基于各種語言的眾多第三方客戶端功能庫。

1.1 Features

1). Distributed

NSQ提供了分布式的，去中心化，且沒有單點故障的拓撲結構，穩定的消息傳輸發布保障，能夠具有高容錯和HA（高可用）特性。

2). Scalable易于擴展

NSQ支持水平擴展，沒有中心化的brokers。內置的發現服務簡化了在集群中增加節點。同時支持pub-sub和load-balanced 的消息分發。

3). Ops Friendly

NSQ非常容易配置和部署，生來就綁定了一個管理界面。二進制包沒有運行時依賴。官方有Docker image。

4.Integrated高度集成

官方的 Go 和 Python庫都有提供。而且為大多數語言提供了庫。

1.2 組件

1.3 拓撲結構

NSQ推薦通過他們相應的nsqd實例使用協同定位發布者，這意味著即使面對網絡分區，消息也會被保存在本地，直到它們被一個消費者讀取。更重要的是，發布者不必去發現其他的nsqd節點，他們總是可以向本地實例發布消息。

NSQ

首先，一個發布者向它的本地nsqd發送消息，要做到這點，首先要先打開一個連接，然后發送一個包含topic和消息主體的發布命令，在這種情況下，我們將消息發布到事件topic上以分散到我們不同的worker中。

事件topic會復制這些消息并且在每一個連接topic的channel上進行排隊，在我們的案例中，有三個channel，它們其中之一作為檔案channel。消費者會獲取這些消息并且上傳到S3。

nsqd

每個channel的消息都會進行排隊，直到一個worker把他們消費，如果此隊列超出了內存限制，消息將會被寫入到磁盤中。Nsqd節點首先會向nsqlookup廣播他們的位置信息，一旦它們注冊成功，worker將會從nsqlookup服務器節點上發現所有包含事件topic的nsqd節點。

nsqlookupd

2. Internals

2.1 消息傳遞擔保

1）客戶表示已經準備好接收消息

2）NSQ 發送一條消息，并暫時將數據存儲在本地（在 re-queue 或 timeout）

3）客戶端回復 FIN（結束）或 REQ（重新排隊）分別指示成功或失敗。如果客戶端沒有回復, NSQ 會在設定的時間超時，自動重新排隊消息

這確保了消息丟失唯一可能的情況是不正常結束 nsqd 進程。在這種情況下，這是在內存中的任何信息（或任何緩沖未刷新到磁盤）都將丟失。

如何防止消息丟失是最重要的，即使是這個意外情況可以得到緩解。一種解決方案是構成冗余 nsqd對（在不同的主機上）接收消息的相同部分的副本。因為你實現的消費者是冪等的，以兩倍時間處理這些消息不會對下游造成影響，并使得系統能夠承受任何單一節點故障而不會丟失信息。

2.2 簡化配置和管理

單個 nsqd 實例被設計成可以同時處理多個數據流。流被稱為“話題”和話題有 1 個或多個“通道”。每個通道都接收到一個話題中所有消息的拷貝。在實踐中，一個通道映射到下行服務消費一個話題。

在更底的層面，每個 nsqd 有一個與 nsqlookupd 的長期 TCP 連接，定期推動其狀態。這個數據被 nsqlookupd 用于給消費者通知 nsqd 地址。對于消費者來說，一個暴露的 HTTP /lookup 接口用于輪詢。為話題引入一個新的消費者，只需啟動一個配置了 nsqlookup 實例地址的 NSQ 客戶端。無需為添加任何新的消費者或生產者更改配置，大大降低了開銷和復雜性。

2.3 消除單點故障

NSQ被設計以分布的方式被使用。nsqd 客戶端（通過 TCP ）連接到指定話題的所有生產者實例。沒有中間人，沒有消息代理，也沒有單點故障。

這種拓撲結構消除單鏈，聚合，反饋。相反，你的消費者直接訪問所有生產者。從技術上講，哪個客戶端連接到哪個 NSQ 不重要，只要有足夠的消費者連接到所有生產者，以滿足大量的消息，保證所有東西最終將被處理。對于 nsqlookupd，高可用性是通過運行多個實例來實現。他們不直接相互通信和數據被認為是最終一致。消費者輪詢所有的配置的 nsqlookupd 實例和合并 response。失敗的，無法訪問的，或以其他方式故障的節點不會讓系統陷于停頓。

2.4 效率

對于數據的協議，通過推送數據到客戶端最大限度地提高性能和吞吐量的，而不是等待客戶端拉數據。這個概念，稱之為 RDY 狀態，基本上是客戶端流量控制的一種形式。

efficiency

2.5 心跳和超時

組合應用級別的心跳和 RDY 狀態，避免頭阻塞現象，也可能使心跳無用（即，如果消費者是在后面的處理消息流的接收緩沖區中，操作系統將被填滿，堵心跳）為了保證進度，所有的網絡 IO 時間上限勢必與配置的心跳間隔相關聯。這意味著，你可以從字面上拔掉之間的網絡連接 nsqd 和消費者，它會檢測并正確處理錯誤。當檢測到一個致命錯誤，客戶端連接被強制關閉。在傳輸中的消息會超時而重新排隊等待傳遞到另一個消費者。最后，錯誤會被記錄并累計到各種內部指標。

2.6 分布式

因為NSQ沒有在守護程序之間共享信息，所以它從一開始就是為了分布式操作而生。個別的機器可以隨便宕機隨便啟動而不會影響到系統的其余部分，消息發布者可以在本地發布，即使面對網絡分區。

這種“分布式優先”的設計理念意味著NSQ基本上可以永遠不斷地擴展，需要更高的吞吐量？那就添加更多的nsqd吧。唯一的共享狀態就是保存在lookup節點上，甚至它們不需要全局視圖，配置某些nsqd注冊到某些lookup節點上這是很簡單的配置，唯一關鍵的地方就是消費者可以通過lookup節點獲取所有完整的節點集。清晰的故障事件——NSQ在組件內建立了一套明確關于可能導致故障的的故障權衡機制，這對消息傳遞和恢復都有意義。雖然它們可能不像Kafka系統那樣提供嚴格的保證級別，但NSQ簡單的操作使故障情況非常明顯。

2.7 no replication

不像其他的隊列組件，NSQ并沒有提供任何形式的復制和集群，也正是這點讓它能夠如此簡單地運行，但它確實對于一些高保證性高可靠性的消息發布沒有足夠的保證。我們可以通過降低文件同步的時間來部分避免，只需通過一個標志配置，通過EBS支持我們的隊列。但是這樣仍然存在一個消息被發布后馬上死亡，丟失了有效的寫入的情況。

2.8 沒有嚴格的順序

雖然Kafka由一個有序的日志構成，但NSQ不是。消息可以在任何時間以任何順序進入隊列。在我們使用的案例中，這通常沒有關系，因為所有的數據都被加上了時間戳，但它并不適合需要嚴格順序的情況。

2.9 無數據重復刪除功能

NSQ對于超時系統，它使用了心跳檢測機制去測試消費者是否存活還是死亡。很多原因會導致我們的consumer無法完成心跳檢測，所以在consumer中必須有一個單獨的步驟確保冪等性。

3. 實踐安裝過程

本文將nsq集群具體的安裝過程略去，大家可以自行參考官網，比較簡單。這部分介紹下筆者實驗的拓撲，以及nsqadmin的相關信息。

3.1 拓撲結構

topology

實驗采用3臺NSQD服務，2臺LOOKUPD服務。

采用官方推薦的拓撲，消息發布的服務和NSQD在一臺主機。一共5臺機器。

NSQ基本沒有配置文件，配置通過命令行指定參數。

主要命令如下:

LOOKUPD命令

NSQD命令

工具類，消費后存儲到本地文件。

發布一條消息

3.2 nsqadmin

對Streams的詳細信息進行查看，包括NSQD節點，具體的channel，隊列中的消息數，連接數等信息。

nsqadmin

channel

列出所有的NSQD節點:

nodes

消息的統計:

msgs

lookup主機的列表:

hosts

4. 總結

NSQ基本核心就是簡單性，是一個簡單的隊列，這意味著它很容易進行故障推理和很容易發現bug。消費者可以自行處理故障事件而不會影響系統剩下的其余部分。

事實上，簡單性是我們決定使用NSQ的首要因素，這方便與我們的許多其他軟件一起維護，通過引入隊列使我們得到了堪稱完美的表現，通過隊列甚至讓我們增加了幾個數量級的吞吐量。越來越多的consumer需要一套嚴格可靠性和順序性保障，這已經超過了NSQ提供的簡單功能。

結合我們的業務系統來看，對于我們所需要傳輸的發票消息，相對比較敏感，無法容忍某個nsqd宕機，或者磁盤無法使用的情況，該節點堆積的消息無法找回。這是我們沒有選擇該消息中間件的主要原因。簡單性和可靠性似乎并不能完全滿足。相比Kafka，ops肩負起更多負責的運營。另一方面，它擁有一個可復制的、有序的日志可以提供給我們更好的服務。但對于其他適合NSQ的consumer，它為我們服務的相當好，我們期待著繼續鞏固它的堅實的基礎。

Golang kafka簡述和操作（sarama同步異步和消費組）

一、Kafka簡述

1. 為什么需要用到消息隊列

異步：對比以前的串行同步方式來說，可以在同一時間做更多的事情，提高效率；

解耦：在耦合太高的場景，多個任務要對同一個數據進行操作消費的時候，會導致一個任務的處理因為另一個任務對數據的操作變得及其復雜。

緩沖：當遇到突發大流量的時候，消息隊列可以先把所有消息有序保存起來，避免直接作用于系統主體，系統主題始終以一個平穩的速率去消費這些消息。

2.為什么選擇kafka呢？

這沒有絕對的好壞，看個人需求來選擇，我這里就抄了一段他人總結的的優缺點，可見原文

kafka的優點：

1.支持多個生產者和消費者2.支持broker的橫向拓展3.副本集機制，實現數據冗余，保證數據不丟失4.通過topic將數據進行分類5.通過分批發送壓縮數據的方式，減少數據傳輸開銷，提高吞高量6.支持多種模式的消息7.基于磁盤實現數據的持久化8.高性能的處理信息，在大數據的情況下，可以保證亞秒級的消息延遲9.一個消費者可以支持多種topic的消息10.對CPU和內存的消耗比較小11.對網絡開銷也比較小12.支持跨數據中心的數據復制13.支持鏡像集群

kafka的缺點：

1.由于是批量發送，所以數據達不到真正的實時2.對于mqtt協議不支持3.不支持物聯網傳感數據直接接入4.只能支持統一分區內消息有序，無法實現全局消息有序5.監控不完善，需要安裝插件6.需要配合zookeeper進行元數據管理7.會丟失數據，并且不支持事務8.可能會重復消費數據，消息會亂序，可用保證一個固定的partition內部的消息是有序的，但是一個topic有多個partition的話，就不能保證有序了，需要zookeeper的支持，topic一般需要人工創建，部署和維護一般都比mq高

3. Golang 操作kafka

3.1. kafka的環境

網上有很多搭建kafka環境教程，這里就不再搭建，就展示一下kafka的環境，在kubernetes上進行的搭建，有需要的私我，可以發yaml文件

3.2. 第三方庫

github.com/Shopify/sarama // kafka主要的庫*github.com/bsm/sarama-cluster // kafka消費組

3.3. 消費者

單個消費者

funcconsumer(){varwg sync.WaitGroup? consumer, err := sarama.NewConsumer([]string{"172.20.3.13:30901"},nil)iferr !=nil{? ? ? fmt.Println("Failed to start consumer: %s", err)return}? partitionList, err := consumer.Partitions("test0")//獲得該topic所有的分區iferr !=nil{? ? ? fmt.Println("Failed to get the list of partition:, ", err)return}forpartition :=rangepartitionList {? ? ? pc, err := consumer.ConsumePartition("test0",int32(partition), sarama.OffsetNewest)iferr !=nil{? ? ? ? fmt.Println("Failed to start consumer for partition %d: %s\n", partition, err)return}? ? ? wg.Add(1)gofunc(sarama.PartitionConsumer){//為每個分區開一個go協程去取值formsg :=rangepc.Messages() {//阻塞直到有值發送過來，然后再繼續等待fmt.Printf("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset,string(msg.Key),string(msg.Value))? ? ? ? }deferpc.AsyncClose()? ? ? ? wg.Done()? ? ? }(pc)? }? wg.Wait()}funcmain(){? consumer()}

消費組

funcconsumerCluster(){? groupID :="group-1"config := cluster.NewConfig()? config.Group.Return.Notifications =trueconfig.Consumer.Offsets.CommitInterval =1* time.Second? config.Consumer.Offsets.Initial = sarama.OffsetNewest//初始從最新的offset開始c, err := cluster.NewConsumer(strings.Split("172.20.3.13:30901",","),groupID, strings.Split("test0",","), config)iferr !=nil{? ? ? glog.Errorf("Failed open consumer: %v", err)return}deferc.Close()gofunc(c *cluster.Consumer){? ? ? errors := c.Errors()? ? ? noti := c.Notifications()for{select{caseerr := -errors:? ? ? ? ? ? glog.Errorln(err)case-noti:? ? ? ? }? ? ? }? }(c)formsg :=rangec.Messages() {? ? ? fmt.Printf("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset,string(msg.Key),string(msg.Value))? ? ? c.MarkOffset(msg,"")//MarkOffset 并不是實時寫入kafka，有可能在程序crash時丟掉未提交的offset}}funcmain(){goconsumerCluster()}

3.4. 生產者

同步生產者

packagemainimport("fmt""github.com/Shopify/sarama")funcmain(){? config := sarama.NewConfig()? config.Producer.RequiredAcks = sarama.WaitForAll//賦值為-1：這意味著producer在follower副本確認接收到數據后才算一次發送完成。config.Producer.Partitioner = sarama.NewRandomPartitioner//寫到隨機分區中，默認設置8個分區config.Producer.Return.Successes =truemsg := sarama.ProducerMessage{}? msg.Topic =`test0`msg.Value = sarama.StringEncoder("Hello World!")? client, err := sarama.NewSyncProducer([]string{"172.20.3.13:30901"}, config)iferr !=nil{? ? ? fmt.Println("producer close err, ", err)return}deferclient.Close()? pid, offset, err := client.SendMessage(msg)iferr !=nil{? ? ? fmt.Println("send message failed, ", err)return}? fmt.Printf("分區ID:%v, offset:%v \n", pid, offset)}

異步生產者

funcasyncProducer(){? config := sarama.NewConfig()? config.Producer.Return.Successes =true//必須有這個選項config.Producer.Timeout =5* time.Second? p, err := sarama.NewAsyncProducer(strings.Split("172.20.3.13:30901",","), config)deferp.Close()iferr !=nil{return}//這個部分一定要寫，不然通道會被堵塞gofunc(p sarama.AsyncProducer){? ? ? errors := p.Errors()? ? ? success := p.Successes()for{select{caseerr := -errors:iferr !=nil{? ? ? ? ? ? ? glog.Errorln(err)? ? ? ? ? ? }case-success:? ? ? ? }? ? ? }? }(p)for{? ? ? v :="async: "+ strconv.Itoa(rand.New(rand.NewSource(time.Now().UnixNano())).Intn(10000))? ? ? fmt.Fprintln(os.Stdout, v)? ? ? msg := sarama.ProducerMessage{? ? ? ? Topic: topics,? ? ? ? Value: sarama.ByteEncoder(v),? ? ? }? ? ? p.Input() - msg? ? ? time.Sleep(time.Second *1)? }}funcmain(){goasyncProducer()select{? ? ? }}

3.5. 結果展示-

同步生產打?。?/p>

分區ID:0,offset:90

消費打?。?/p>

Partition:0,Offset:90,key:,value:Hello World!

異步生產打印：

async:7272async:7616async:998

消費打印：

Partition:0,Offset:91,key:,value:async:7272Partition:0,Offset:92,key:,value:async:7616Partition:0,Offset:93,key:,value:async:998

聊聊golang的zap的ZapKafkaWriter

本文主要研究一下golang的zap的ZapKafkaWriter

WriteSyncer內嵌了io.Writer接口，定義了Sync方法；Sink接口內嵌了zapcore.WriteSyncer及io.Closer接口；ZapKafkaWriter實現Sink接口及zapcore.WriteSyncer接口，其Write方法直接將data通過kafka發送出去。

golang的回調和接口

最近寫了個kafka的接收消息的功能，需要使用回調處理收到的消息。

一個是基本的回調，一個是使用接口功能實現回調，對接口是個很好的學習。

1.正?；卣{

kafka的接收消息處。收到消息后，使用傳入的Onmessage進行處理。

調用kafka接收消息的單元，并在調用方寫好回調

在調用方實現回調需要執行的方法

感覺還是使用基本回調相對簡單點，接口就當學習了。

另外跨包的接口的方法要大寫！定位了好久發現個入門的問題。