Zabbix-(四)郵件、釘釘告警通知

一.前言

在之前的文章里，通過Zabbix對主機的磁盤、CPU以及內存進行了監控，並在首頁Dashboard里創建了監控圖形，但是只有當我們登錄到Zabbix后才能看到監控到的問題(Problem)，因此在本篇文章里，將利用觸發器(Trigger)，以及媒介(Media)等配置項，實現當觸發器觸發時，通過不同媒介，如：郵件、釘釘，發送動作(Action)，實現實時通知告警功能。

準備

• Zabbix Server (Zabbix 4.4)
• 在Zabbix中已配置一些監控項和觸發器(這些配置可以參考我的)

二.安裝相關環境

由於使用到腳本告警媒介，本文中通過調用Python腳本觸發告警，因此需要在Zabbix Server主機上安裝pip以及相關模塊。(這裏Python使用Centos7自帶的Python2.7.5)

1. 安裝pip

   # yum install -y epel-release
   
   # yum install -y python-pip

2. 安裝requests模塊

   # pip install requests

三.配置告警媒介類型

Zabbix默認自帶了2種報警媒介類型(Media Type)，电子郵件以及短信，我們將修改电子郵件類型配置，並新建腳本類型和Webhook類型。希望通過腳本、Webhook告警媒介發送釘釘消息。

注：Webhook告警媒介是Zabbix 4.4的新特性

1. 修改电子郵件告警媒介

   點擊【管理】-【報警媒介類型】-【Email】

   修改Email配置，我這裏用的是Outlook郵箱，具體SMTP服務器可以參考。使用其他郵箱也可以去對應官網查詢SMTP配置。

   測試發送郵箱，點擊【測試】

   輸入收件人郵箱

   收到郵件

2. 新增腳本告警媒介

   新建Python腳本告警媒介，用戶釘釘告警

   點擊【創建媒體類型】

   進行配置

   配置項	值
   * 名稱	Python腳本
   類型	腳本
   * 腳本名稱	pythonScript.py
   腳本參數(參數1)	{ALERT.MESSAGE}
   腳本參數(參數2)	{ALERT.SENDTO}
   腳本參數(參數3)	{ALERT.SUBJECT}

   接下來新建Python腳本，Zabbix Server配置文件中可以配置告警腳本路徑，默認為 /usr/lib/zabbix/alertscripts

   # 查看告警腳本路徑
   # cat zabbix_server.conf | grep AlertScriptsPath

   編寫告警腳本

   # cd /usr/lib/zabbix/alertscripts
   # vim pythonScript.py

   腳本內容

   #!/usr/bin/env python
   #coding:utf-8
   
   import requests,json,sys,os,datetime
   
   # 釘釘機器人地址
   webhook="https://oapi.dingtalk.com/robot/send?access_token=your_dingding_robot_access_token"
   
   # 對應{ALERT.SENDTO}, Zabbix告警媒介配置界面第2個參數
   user=sys.argv[2]
   
   # 對應{ALERT.MESSAGE}, Zabbix告警媒介配置界面第1個參數
   text=sys.argv[1]
   data={
       "msgtype": "text",
       "text": {
           "content": text
       },
       "at": {
           "atMobiles": [
               user
           ],
           "isAtAll": False
       }
   }
   headers = {'Content-Type': 'application/json'}
   x=requests.post(url=webhook,data=json.dumps(data),headers=headers)

   給腳本可執行權限

   # chmod uo+x /usr/lib/zabbix/alertscripts/pythonScript.py

   測試腳本

   釘釘收到消息

3. 新增Webhook告警媒介

   配置項	值
   * 名稱	Webhook
   類型	Webhook
   參數: （名稱）	值
   user	{ALERT.SENDTO}
   subject	{ALERT.SUBJECT}
   message	{ALERT.MESSAGE}

   腳本:

   try {
       Zabbix.Log(4, 'params= '+value);
   
       params = JSON.parse(value);
       req = new CurlHttpRequest();
       data = {};
       result = {};
   
       req.AddHeader('Content-Type: application/json');
   
       data.msgtype = "text";
       //   對應 message參數
       data.text = {"content" : params.message};
       //   對應 user參數
       data.at = {"atMobiles": [params.user], "isAtAll": "false"};
   
       //   釘釘機器人
       resp = req.Post('https://oapi.dingtalk.com/robot/send?access_token=your_access_token',
           JSON.stringify(data)
       );
   } catch (error) {
       result = {};
   }
   
   return JSON.stringify(result);

   測試Webhook

​

四.為用戶添加告警媒介

需要將新增的告警媒介添加給用戶

點擊【用戶】-【告警媒介】

將上述步驟添加的告警媒介(Python腳本、Webhoob、Email)，進行添加(收件人根據告警媒介類型填寫郵箱或手機號)，嚴重性也根據需要勾選。

五.配置動作

完成上述配置完成后，需要創建動作(Action)，將觸發器(Trigger)和告警媒介(Media Type)進行關聯，一旦觸發器觸發，那麼Zabbix會執行動作，再去執行告警媒介。

1. 添加動作

   點擊【配置】-【動作】-【創建動作】

2. 配置【動作】相關信息

   配置項	值
   * 名稱	告警動作
   新的觸發條件	【觸發器】【等於】【Template Disk Free Size: 磁盤剩餘空間觸發器】

   操作步驟如下圖：

   群組選擇 ->Linux servers

   主機選擇 -> Template Disk Free Size 模板()

   勾選觸發器 -> 磁盤剩餘空間觸發器 ()

   勾選後點擊【選擇】

3. 配置【操作】相關信息

   點擊【操作】

   先配置以下信息

   配置項	值
   * 默認操作步驟持續時間	1h（保持默認）
   默認標題	告警: {EVENT.NAME}
   消息內容	【磁盤空間不足告警】 告警事件: {EVENT.DATE} {EVENT.TIME} 告警問題: {EVENT.NAME} 告警主機: {HOST.IP} {HOST.NAME} 告警級別: {EVENT.SEVERITY} 磁盤剩餘:{ITEM.VALUE}

   上述配置表格【默認標題】和【消息內容】值中形如{EVENT.NAME}的內容是Zabbix中的宏(Marco)，宏是一個變量，例如 {HOST.IP} 表示告警主機的IP地址，Zabbix自帶的宏可以參考

   繼續配置操作

   點擊【新的】

   【操作類型】選擇發送消息，【發送到用戶】添加Admin

   【僅送到】根據需要選擇之前配置的，本文選擇Email和Python腳本(這裏只能單選或全選，所以需要先選擇一個，因此需要多次)

   添加完成後點擊【添加】

六.測試

向被監控主機拷貝或下載大文件，使其磁盤剩餘空間低於觸發器監控閾值，等待觸發器觸發問題，查看儀錶盤、郵件等。

儀錶盤

釘釘

郵件

七.參考文檔

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理
【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包”嚨底家”

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※小三通海運與一般國際貿易有何不同?

※小三通快遞通關作業有哪些?

寫在前面的

最近看了Kafka Summit上的這個分享，覺得名字很霸氣，標題直接沿用了。這個分享源於社區的，大體的意思今後Apache Kafka不再需要ZooKeeper。整個分享大約40幾分鐘。完整看下來感覺乾貨很多，這裏特意總結出來。如果你把這個分享看做是《三國志》的話，那麼姑且就把我的這篇看做是裴松之注吧：）

客戶端演進

首先，社區committer給出了Kafka Java客戶端移除ZooKeeper依賴的演進過程。下面兩張圖總結了0.8.x版本和0.11.x版本（是否真的是從0.11版本開始的變化並不重要）及以後的功能變遷：在Kafka 0.8時代，Kafka有3個客戶端，分別是Producer、Consumer和Admin Tool。其中Producer負責向Kafka寫消息，Consumer負責從Kafka讀消息，而Admin Tool執行各種運維任務，比如創建或刪除主題等。其中Consumer的位移數據保存在ZooKeeper上，因此Consumer端的位移提交和位移獲取操作都需要訪問ZooKeeper。另外Admin Tool執行運維操作也要訪問ZooKeeper，比如在對應的ZooKeeper znode上創建一個臨時節點，然後由預定義的Watch觸發相應的處理邏輯。

後面隨着Kafka的演進，社區引入了__consumer_offsets位移主題，同時定義了OffsetFetch和OffsetCommit等新的RPC協議，這樣Consumer的位移提交和位移獲取操作全部轉移到與位移主題進行交互，避免了對ZooKeeper的訪問。同時社區引入了新的運維工具AdminClient以及相應的CreateTopics、DeleteTopics、AlterConfigs等RPC協議，替換了原先的Admin Tool，這樣創建和刪除主題這樣的運維操作也完全移動Kafka這一端來做，就像下面右邊這張圖展示的：

至此， Kafka的3個客戶端基本上都不需要和ZooKeeper交互了。應該說移除ZooKeeper的工作完成了大部分，但依然還有一部分工作要在ZooKeeper的幫助下完成，即Consumer的Rebalance操作。在0.8時代，Consumer Group的管理是交由ZooKeeper完成的，包括組成員的管理和訂閱分區的分配。這個設計在新版Consumer中也得到了修正。全部的Group管理操作交由Kafka Broker端新引入的Coordinator組件來完成。要完成這些工作，Broker端新增了很多RPC協議，比如JoinGroup、SyncGroup、Heartbeat、LeaveGroup等。

  

此時，Kafka的Java客戶端除了AdminClient還有一點要依賴ZooKeeper之外，所有其他的組件全部擺脫了對ZooKeeper的依賴。

之後，社區引入了Kafka安全層，實現了對用戶的認證和授權。這個額外的安全層也是不需要訪問ZooKeeper的，因此之前依賴ZooKeeper的客戶端是無法“享用”這個安全層。一旦啟用，新版Clients都需要首先接入這一層並通過審核之後才能訪問到Broker，如下圖所示：

這麼做的好處在於統一了Clients訪問Broker的模式，即定義RPC協議，比如我們熟知的PRODUCE協議、FETCH協議、METADATA協議、CreateTopics協議等。如果後面需要實現更多的功能，社區只需要定義新的RPC協議即可。同時新引入的安全層負責對這套RPC協議進行安全校驗，統一了訪問模式。另外這些協議都是版本化的（versioned），因此能夠獨立地進行演進，同時也兼顧了兼容性方面的考量。

Broker間交互

說完了Clients端，我們說下Broker端的現狀。目前，應該說Kafka Broker端對ZooKeeper是重度依賴的，主要表現在以下幾個方面：

• Broker註冊管理
• ACL安全層配置管理
• 動態參數管理
• 副本ISR管理
• Controller選舉

我們拿一張圖來說明，圖中有4個Broker節點和一個ZooKeeper，左上角的Broker充當Controller的角色。當前，所有的Broker啟動后都必須維持與ZooKeeper的會話。Kafka依賴於這個會話實現Broker端的註冊，而且Kafka集群中的所有配置信息、副本信息、主題信息也都保存在ZooKeeper上。最後Controller與集群中每個Broker都維持了一個TCP長連接用於向這些Broker發送RPC請求。當前的Controller RPC類型主要有3大類：

• LeaderAndIsr：主要用於向集群廣播主題分區Leader和ISR的變更情況，比如對應的Broker應該是特定分區的Leader還是Follower
• StopReplica：向集群廣播執行停止副本的命令
• UpdateMetadata：向集群廣播執行變更元數據信息的命令

圖中還新增了一個AlterISR RPC，這是KIP-497要實現的新RPC協議。現階段Kafka各個主題的ISR信息全部保存在ZooKeeper中。如果後續要捨棄ZooKeeper，必須要將這些信息從ZooKeeper中移出來，放在了Controller一端來做。同時還要在程序層面支持對ISR的管理。因此社區計劃在KIP-497上增加AlterISR協議。對了，還要提一句，當前Controller的選舉也是依靠ZooKeeper完成的。

所以後面Broker端的演進可能和Clients端的路線差不多：首先是把Broker與ZooKeeper的交互全部幹掉，只讓Controller與ZooKeeper進行交互，而其他所有Broker都只與Controller交互，如下圖所示：

 

看上去這種演進路線社區已經走得輕車熟路了，但實際上還有遺留了一些問題需要解決。

Broker Liveness

首先就是Broker的liveness問題，即Kafka如何判斷一個Broker到底是否存活？在目前的設計中，Broker的生存性監測完全依賴於與ZooKeeper之間的會話。一旦會話超時或斷開Controller自動觸發ZooKeeper端的Watch來移除該Broker，並對其上的分區做善後處理。如果移除了ZooKeeper，Kafka應該採用什麼機制來判斷Broker的生存性是一個問題。

Network Partition

如何防範網絡分區也是一個需要討論的話題。當前可能出現的Network Partition有4種：1、單個Broker完全與集群隔離；2、Broker間無法通訊；3、Broker與ZooKeeper無法通訊；4、Broker與Controller無法通訊。下面4張圖分別展示了這4種情況：

 

我們分別討論下。首先是第一種情況，單Broker與集群其他Broker隔離，這其實並不算太嚴重的問題。當前的設計已然能夠保證很好地應對此種情況。一旦Broker被隔離，Controller會將其從集群中摘除，雖然可用性降低了，但是整個集群的一致性依然能夠得到保證。第二種情況是Broker間無法通訊，可能的後果是消息的備份機制無法執行，Kafka要收縮ISR，依然是可用性上的降低，但是一致性狀態並沒有被破壞。情況三是Broker無法與ZooKeeper通訊。Broker能正常運轉，它只是無法與ZooKeeper進行通訊。此時我們說該Broker處於殭屍狀態，即所謂的Zoobie狀態。因Zoobie狀態引入的一致性bug社區jira中一直沒有斷過，社區這幾年也一直在修正這方面的問題，主要對抗的機制就是fencing。比如leader epoch等。最後一類情況是Broker無法與Controller通訊，那麼所有的元數據更新通道被堵死，即使這個Broker依然是healthy的，但是它保存的元數據信息可能是非常過期的。這樣連接該Broker的客戶端可能會看到各種非常古怪的問題。之前在知乎上回答過類似的問題：4。目前，社區對這種情況並沒有太好的解決辦法，主要的原因是Broker的liveness完全交由ZooKeeper來做的。一旦Broker與ZooKeeper之間的交互沒有問題，其他原因導致的liveness問題就無法徹底規避。

第四類Network Partition引入了一個經典的場景：元數據不一致。目前每個Broker都緩存了一份集群的元數據信息，這份數據是異步更新的。當第四類Partition發生時，Broker端緩存的元數據信息必然與Controller的不同步，從而造成各種各樣的問題。

下面簡要介紹一下元數據更新的過程。主要的流程就是Controller啟動時會同步地從ZooKeeper上拉取集群全量的元數據信息，之後再以異步的方式同步給其他Broker。其他Broker與Controller之間的同步往往有一個時間差，也就是說可能Clients訪問的元數據並不是最新的。我個人認為現在社區很多flaky test failure都是因為這個原因導致的。 事實上，實際使用過程中有很多場景是Broker端的元數據與Controller端永遠不同步。通常情況下如果我們不重啟Broker的話，那麼這個Broker上的元數據將永遠“錯誤”下去。好在社區還給出了一個最後的“大招”： 登錄到ZooKeeper SHELL，手動執行rmr /controller，強迫Controller重選舉，然後重新加載元數據，並給所有Broker重刷一份。不過在實際生產環境，我懷疑是否有人真的要這麼干，畢竟代價不小，而且最關鍵的是這麼做依然可能存在兩個問題：1. 我們如何確保Controller和Broker的數據是一致的？2. 加載元數據的過程通常很慢。

這裏詳細說說第二點，即加載元數據的性能問題。總體來說，加載元數據是一個O(N)時間複雜度的過程，這裏的N就是你集群中總的分區數。考慮到Controller從ZooKeeper加載之後還要推給其他的Broker，那麼做這件事的總的時間複雜度就是O(N * M)，其中M是集群中Broker的數量。可以想見，當M和N都很大時，在集群中廣播元數據不是一個很快的過程。

Metadata as an Event Log

Okay，鑒於以上所提到的所有問題，當Kafka拋棄了ZooKeeper之後，社區應該如何解決它們呢？總體的思路就是Metadata as an Event Log + Controller quorum。我們先說metadata as an event log。如果你讀過Jay Kreps的《I ️Logs》，你應該有感觸，整個Kafka的架構其實都是構建在Log上的。每個topic的分區本質上就是一個Commit Log，但元數據信息的保存卻不是Log形式。在現有的架構設計中你基本上可以認為元數據的數據結構是KV形式的。這一次，社區採用了與消息相同的數據保存方式，即將元數據作為Log的方式保存起來，如下錶所示：

 

這樣做的好處在於每次元數據的變更都被當做是一條消息保存在Log中，而這個Log可以被視作是一個普通的Kafka主題被備份到多台Broker上。Log的一個好處在於它有清晰的前後順序關係，即每個事件發生的時間是可以排序的，配合以恰當的處理邏輯，我們就能保證對元數據變更的處理是按照變更發生時間順序處理，不出現亂序的情形。另外Log機制還有一個好處是，在Broker間同步元數據時，我們可以選擇同步增量數據（delta），而非全量狀態。現在Kafka Broker間同步元數據都是全量狀態同步的。前面說過了，當集群分區數很大時，這個開銷是很可觀的。如果我們能夠只同步增量狀態，勢必能極大地降低同步成本。最後一個好處是，我們可以很容易地量化元數據同步的進度，因為對Log的消費有位移數據，因此通過監控Log Lag就能算出當前同步的進度或是落後的進度。

採用Log機制后，其他Broker像是一個普通的Consumer，從Controller拉取元數據變更消息或事件。由於每個Broker都是一個Consumer，所以它們會維護自己的消費位移，就像下面這張圖一樣：

 這種設計下，Controller所在的Broker必須要承擔起所有元數據topic的管理工作，包括創建topic、管理topic分區的leader以及為每個元數據變更創建相應的事件等。既然社區選擇和__consumer_offsets類似的處理方式，一個很自然的問題在於這個元數據topic的管理是否能夠復用Kafka現有的副本機制？答案是：不可行。理由是現有的副本機制依賴於Controller，因此Kafka沒法依靠現有的副本機制來實現Controller——按照我們的俗語來說，這有點雞生蛋、蛋生雞的問題，屬於典型的循環依賴。為了實現這個，Kafka需要一套leader選舉協議，而這套協議或算法是不依賴於Controller的，即它是一個自管理的集群quorum（抱歉，在分佈式領域內，特別是分佈式共識算法領域中，針對quorum的恰當翻譯我目前還未找到，因此直接使用quorum原詞了）。最終社區決定採用Raft來實現這組quorum。這就是上面我們提到的第二個解決思路：Controller quorum。

Controller Quorum

與藉助Controller幫忙選擇Leader不同，Raft是讓自己的節點自行選擇Leader並最終令所有節點達成共識——對選擇Controller而言，這是一個很好的特性。其實Kafka現有的備份機制與Raft已經很接近了，下錶羅列了一下它們的異同：

 一眼掃過去，其實Kafka的備份機制和Raft很類似，比如Kafka中的offset其實就是Raft中的index，epoch對應於term。當然Raft中採用的半數機制來確保消息被提交以及Leader選舉，而Kafka設計了ISR機制來實現這兩點。總體來說，社區認為只需要對備份機製做一些小改動就應該可以很容易地切換到Raft-based算法。

下面這張圖展示Controller quorum可能更加直觀：

整個controller quorum類似於一個小的集群。和ZooKeeper類似，這個quorum通常是3台或5台機器，不需要讓Kafka中的每個Broker都自動稱為這個quorum中的一個節點。該quorum裏面有一個Leader負責處理客戶端發來的讀寫請求，這個Leader就是Kafka中的active controller。根據ZooKeeper的Zab協議，leader處理所有的寫請求，而follower是可以處理讀請求的。當寫請求發送給follower后，follower會將該請求轉發給leader處理。不過我猜Kafka應該不會這樣實現，它應該只會讓leader（即active controller）處理所有的讀寫請求，而客戶端（也就是其他Broker）壓根就不會發送讀寫請求給follower。在這一點上，這種設計和現有的Kafka請求處理機制是一致的。

現在還需要解決一個問題，即Leader是怎麼被選出來的？既然是Raft-based，那麼採用的也是Raft算法中的Leader選舉策略。讓Raft選出的Leader稱為active controller。網上有很多關於Raft選主的文章，這裏就不在贅述了，有興趣的可以讀一讀Raft的論文：《In Search of an Understandable Consensus Algorithm(Extended Version)》。

這套Raft quorum的一個好處在於它天然提供了低延時的failover，因此leader的切換會非常的迅速和及時，因為理論上不再有元數據加載的過程了，所有的元數據現在都同步保存follower節點的內存中，它已經有其他Broker需要拉取的所有元數據信息了！更酷的是，它避免了現在機制中一旦Controller切換要全量拉取元數據的低效行為，Broker無需重新拉取之前已經“消費”的元數據變更消息，它只需要從新Leader繼續“消費”即可。

另一個好處在於：採用了這套機制后，Kafka可以做元數據的緩存了（metadata caching）：即Broker能夠把元數據保存在磁盤上，同時就像剛才說的，Broker只需讀取它關心的那部分數據即可。還有，和現在snapshot機制類似，如果一個Broker保存的元數據落後Controller太多或者是一個全新的Broker，Kafka甚至可以像Raft那樣直接發送一個snapshot文件，快速令其追上進度。當然大多數情況下，Broker只需要拉取delta增量數據即可。

Post KIP-500 Broker註冊

當前Broker啟動之後會向ZooKeeper註冊自己的信息，比如自己的主機名、端口、監聽協議等數據。移除ZooKeeper之後，Broker的註冊機制也要發生變化：Broker需要向active controller發送心跳來進行註冊。Controller收集心跳中包含的Broker數據構建整個Kafka集群信息，如下圖所示：

 同時Controller也會對心跳進行響應，顯式地告知Broker它們是否被允許加入集群——如果不允許，則可能需要被隔離（fenced）。當然controller自己也可以對自己進行隔離。我們針對前面提到的隔離場景討論下KIP-500是怎麼應對的。

Fencing

首先是普通Broker與集群完全隔離的場景，比如該Broker無法與controller和其他Broker進行通信，但它依然可以和客戶端程序交互。此時，fencing機制就很簡單了，直接讓controller令其下線即可。這和現在依靠ZooKeeper會話機制維持Broker判活的機制是一模一樣的，沒有太大改進。

第二種情況是Broker間的通訊中斷。此時消息無法在leader、follower間進行備份。但是對於元數據而言，我們不會看到數據不一致的情形，因為Broker依然可以和controller通訊，因此也不會有什麼問題。

第三種情況是Broker與Controller的隔離。現有機制下這是個問題，但KIP-500之後，Controller僅僅將該Broker“踢出場”即可，不會造成元數據的不一致。

最後一種情況是Broker與ZooKeeper的隔離， 既然ZooKeeper要被移除了，自然這也不是問題了。

部署

終於聊到KIP-500之後的Kafka運維了。下錶總結了KIP-500前後的部署情況對比：

很簡單，現在任何時候部署和運維Kafka都要考慮對ZooKeeper的運維管理。在KIP-500之後我們只需要關心Kafka即可。

Controller quorum共享模式

如前所述，controller改成Raft quorum機制后，可能使用3或5台機器構成一個小的quorum。那麼一個很自然的問題是，這些Broker機器還能否用作他用，是唯一用作controller quorum還是和其他Broker一樣正常處理。社區對此也做了解釋：兩種都支持！

如果你的Kafka集群資源很緊張，你可以使用共享controller模式（Shared Controller Mode），即充當controller quorum的Broker機器也能處理普通的客戶端請求；相反地，如果你的Kafka資源很充足，專屬controller模式（Separate Controller Mode）可能是更適合的，即在controller quorum中的Broker機器排它地用作Controller的選舉之用，不再對客戶端提供讀寫服務。這樣可以實現更好的資源隔離，適用於大集群。

Roadmap

最後說一下KIP-500的計劃。社區計劃分三步走：

第一步是移除客戶端對ZooKeeper的依賴——這一步基本上已經完成了，除了目前AdminClient還有少量的API依賴ZooKeeper之外，其他客戶端應該說都不需要訪問ZooKeeper了；第二步是移除Broker端的ZooKeeper依賴：這主要包括移除Broker端需要訪問ZooKeeper的代碼，以及增加新的Broker端API，如前面所說的AlterISR等，最後是將對ZooKeeper的訪問全部集中在controller端；最後一步就是實現controller quorum，實現Raft-based的quorum負責controller的選舉。

至於Kafka升級，如果從現有的Kafka直接升級到KIP-500之後的Kafka會比較困難，因此社區打算引入一個名為Bridge Release的中間過渡版本，如下圖所示：

這個Bridge版本的特點在於所有對ZooKeeper的訪問都集中到了controller端，Broker訪問ZooKeeper的其他代碼都被移除了。 

總結

KIP-500應該說是最近幾年社區提出的最重磅的KIP改進了。它幾乎是顛覆了Kafka已有的使用模式，摒棄了之前重度依賴的Apache ZooKeeper。就我個人而言，我是很期待這個KIP，後續有最新消息我也會在一併同步出來。讓我們靜觀其變吧~~~

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站，提高曝光率!!

※網頁設計一頭霧水??該從何著手呢? 找到專業技術的網頁設計公司，幫您輕鬆架站!

※想知道最厲害的台北網頁設計公司推薦、台中網頁設計公司推薦專業設計師”嚨底家”!!

※大陸寄台灣空運注意事項

※大陸海運台灣交貨時間多久?

※避免吃悶虧無故遭抬價!台中搬家公司免費估價,有契約讓您安心有保障!

Redis 提供了兩種持久化方式，一種是基於快照形式的 RDB，另一種是基於日誌形式的 AOF，每種方式都有自己的優缺點，本文將介紹 Redis 這兩種持久化方式，希望閱讀本文後你對 Redis 的這兩種方式有更加全面、清晰的認識。

RDB 快照方式持久化

先從 RDB 快照方式聊起，RDB 是 Redis 默認開啟的持久化方式，並不需要我們單獨開啟，先來看看跟 RDB 相關的配置信息：

################################ SNAPSHOTTING  ################################
#
# Save the DB on disk:
#
#   save <seconds> <changes>
#
#   Will save the DB if both the given number of seconds and the given
#   number of write operations against the DB occurred.
#
#   In the example below the behaviour will be to save:
#   after 900 sec (15 min) if at least 1 key changed
#   after 300 sec (5 min) if at least 10 keys changed
#   after 60 sec if at least 10000 keys changed
#   save ""
# 自動生成快照的觸發機制 中間的是時間，單位秒，後面的是變更數據 60 秒變更 10000 條數據則自動生成快照
save 900 1
save 300 10
save 60 10000

# 生成快照失敗時，主線程是否停止寫入
stop-writes-on-bgsave-error yes

# 是否採用壓縮算法存儲
rdbcompression yes

# 數據恢復時是否檢測 RDB文件有效性
rdbchecksum yes

# The filename where to dump the DB
# RDB 快照生成的文件名稱
dbfilename dump.rdb

# 快照生成的路徑 AOF 也是存放在這個路徑下面
dir .

關於 RDB 相關配置信息不多，需要我們調整的就更少了，我們只需要根據自己的業務量修改生成快照的機制和文件存放路徑即可。

RDB 有兩種持久化方式：手動觸發 和 自動觸發，手動觸發使用以下兩個命令：

• save：會阻塞當前 Redis 服務器響應其他命令，直到 RDB 快照生成完成為止，對於內存 比較大的實例會造成長時間阻塞，所以線上環境不建議使用

• bgsave：Redis 主進程會 fork 一個子進程，RDB 快照生成有子進程來負責，完成之後，子進程自動結束，bgsave 只會在 fork 子進程的時候短暫的阻塞，這個過程是非常短的，所以推薦使用該命令來手動觸發

除了執行命令手動觸發之外，Redis 內部還存在自動觸發 RDB 的持久化機制，在以下幾種情況下 Redis 會自動觸發 RDB 持久化：

• 在配置中配置了 save 相關配置信息，如我們上面配置文件中的 save 60 10000 ，也可以把它歸類為“save m n”格式的配置，表示 m 秒內數據集存在 n 次修改時，會自動觸發 bgsave。

• 在主從情況下，如果從節點執行全量複製操作，主節點自動執行 bgsave 生成 RDB 文件併發送給從節點

• 執行 debug reload 命令重新加載 Redis 時，也會自動觸發 save 操作

• 默認情況下執行 shutdown 命令時，如果沒有開啟 AOF 持久化功能則自動執行 bgsave

上面就是 RDB 持久化的方式，可以看出 save 命令使用的比較少，大多數情況下使用的都是 bgsave 命令，所以這個 bgsave 命令還是有一些東西，那接下來我們就一起看看 bgsave 背後的原理，先從流程圖開始入手：

bgsave 命令大概有以下幾個步驟：

• 1、執行 bgsave 命令，Redis 主進程判斷當前是否存在正在執行的 RDB/AOF 子進程，如果存在， bgsave 命令直接返回不在往下執行。
• 2、父進程執行 fork 操作創建子進程，fork 操作過程中父進程會阻塞，fork 完成後父進程將不在阻塞可以接受其他命令。
• 3、子進程創建新的 RDB 文件，基於父進程當前內存數據生成臨時快照文件，完成後用新的 RDB 文件替換原有的 RDB 文件，並且給父進程發送 RDB 快照生成完畢通知

上面就是 bgsave 命令背後的一些內容，RDB 的內容就差不多了，我們一起來總結 RDB 持久化的優缺點，RDB 方式的優點：

• RDB 快照是某一時刻 Redis 節點內存數據，非常適合做備份，上傳到遠程服務器或者文件系統中，用於容災備份
• 數據恢復時 RDB 要遠遠快於 AOF

有優點同樣存在缺點，RDB 的缺點有：

• RDB 持久化方式數據沒辦法做到實時持久化/秒級持久化。我們已經知道了 bgsave 命令每次運行都要執行 fork 操作創建子進程，屬於重量級操作，頻繁執行成本過高。
• RDB 文件使用特定二進制格式保存，Redis 版本演進過程中有多個格式 的 RDB 版本，存在老版本 Redis 服務無法兼容新版 RDB 格式的問題

如果我們對數據要求比較高，每一秒的數據都不能丟，RDB 持久化方式肯定是不能夠滿足要求的，那 Redis 有沒有辦法滿足呢，答案是有的，那就是接下來的 AOF 持久化方式

AOF 持久化方式

Redis 默認並沒有開啟 AOF 持久化方式，需要我們自行開啟，在 redis.conf 配置文件中將 appendonly no 調整為 appendonly yes，這樣就開啟了 AOF 持久化，與 RDB 不同的是 AOF 是以記錄操作命令的形式來持久化數據的，我們可以查看以下 AOF 的持久化文件 appendonly.aof

*2
$6
SELECT
$1
0
*3
$3
set
$6
mykey1
$6
你好
*3
$3
set
$4
key2
$5
hello
*1
$8

大概就是長這樣的，具體的你可以查看你 Redis 服務器上的 appendonly.aof 配置文件，這也意味着我們可以在 appendonly.aof 文件中國修改值，等 Redis 重啟時將會加載修改之後的值。看似一些簡單的操作命令，其實從命令到 appendonly.aof 這個過程中非常有學問的，下面時 AOF 持久化流程圖：

在 AOF 持久化過程中有兩個非常重要的操作：一個是將操作命令追加到 AOF_BUF 緩存區，另一個是 AOF_buf 緩存區數據同步到 AOF 文件，接下來我們詳細聊一聊這兩個操作：

1、為什麼要將命令寫入到 aof_buf 緩存區而不是直接寫入到 aof 文件？

我們知道 Redis 是單線程響應，如果每次寫入 AOF 命令都直接追加到磁盤上的 AOF 文件中，這樣頻繁的 IO 開銷，Redis 的性能就完成取決於你的機器硬件了，為了提升 Redis 的響應效率就添加了一層 aof_buf 緩存層， 利用的是操作系統的 cache 技術，這樣就提升了 Redis 的性能，雖然這樣性能是解決了，但是同時也引入了一個問題，aof_buf 緩存區數據如何同步到 AOF 文件呢？由誰同步呢？這就是我們接下來要聊的一個操作：fsync 操作

2、aof_buf 緩存區數據如何同步到 aof 文件中？

aof_buf 緩存區數據寫入到 aof 文件是有 linux 系統去完成的，由於 Linux 系統調度機制周期比較長，如果系統故障宕機了，意味着一個周期內的數據將全部丟失，這不是我們想要的，所以 Linux 提供了一個 fsync 命令，fsync 是針對單個文件操作（比如這裏的 AOF 文件），做強制硬盤同步，fsync 將阻塞直到寫入硬盤完成后返回，保證了數據持久化，正是由於有這個命令，所以 redis 提供了配置項讓我們自行決定何時進行磁盤同步，redis 在 redis.conf 中提供了appendfsync 配置項，有如下三個選項：

# appendfsync always
appendfsync everysec
# appendfsync no

• always：每次有寫入命令都進行緩存區與磁盤數據同步，這樣保證不會有數據丟失，但是這樣會導致 redis 的吞吐量大大下降，下降到每秒只能支持幾百的 TPS ，這違背了 redis 的設計，所以不推薦使用這種方式
• everysec：這是 redis 默認的同步機制，雖然每秒同步一次數據，看上去時間也很快的，但是它對 redis 的吞吐量沒有任何影響，每秒同步一次的話意味着最壞的情況下我們只會丟失 1 秒的數據， 推薦使用這種同步機制，兼顧性能和數據安全
• no：不做任何處理，緩存區與 aof 文件同步交給系統去調度，操作系統同步調度的周期不固定，最長會有 30 秒的間隔，這樣出故障了就會丟失比較多的數據。

這就是三種磁盤同步策略，但是你有沒有注意到一個問題，AOF 文件都是追加的，隨着服務器的運行 AOF 文件會越來越大，體積過大的 AOF 文件對 redis 服務器甚至是主機都會有影響，而且在 Redis 重啟時加載過大的 AOF 文件需要過多的時間，這些都是不友好的，那 Redis 是如何解決這個問題的呢？Redis 引入了重寫機制來解決 AOF 文件過大的問題。

3、Redis 是如何進行 AOF 文件重寫的？

Redis AOF 文件重寫是把 Redis 進程內的數據轉化為寫命令同步到新 AOF 文件的過程，重寫之後的 AOF 文件會比舊的 AOF 文件占更小的體積，這是由以下幾個原因導致的：

• 進程內已經超時的數據不再寫入文件
• 舊的 AOF 文件含有無效命令，如 del key1、hdel key2、srem keys、set a111、set a222等。重寫使用進程內數據直接生成，這樣新的AOF文件只保 留最終數據的寫入命令
• 多條寫命令可以合併為一個，如：lpush list a、lpush list b、lpush list c可以轉化為：lpush list a b c。為了防止單條命令過大造成客戶端緩衝區溢 出，對於 list、set、hash、zset 等類型操作，以 64 個元素為界拆分為多條。

重寫之後的 AOF 文件體積更小了，不但能夠節約磁盤空間，更重要的是在 Redis 數據恢復時，更小體積的 AOF 文件加載時間更短。AOF 文件重寫跟 RDB 持久化一樣分為手動觸發和自動觸發，手動觸發直接調用 bgrewriteaof 命令就好了，我們後面會詳細聊一聊這個命令，自動觸發就需要我們在 redis.conf 中修改以下幾個配置

auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb

• auto-aof-rewrite-percentage：代表當前 AOF文件空間 （aof_current_size）和上一次重寫后 AOF 文件空間（aof_base_size）的比值，默認是 100%，也就是一樣大的時候
• auto-aof-rewrite-min-size：表示運行 AOF 重寫時 AOF 文件最小體積，默認為 64MB，也就是說 AOF 文件最小為 64MB 才有可能觸發重寫

滿足了這兩個條件，Redis 就會自動觸發 AOF 文件重寫，AOF 文件重寫的細節跟 RDB 持久化生成快照有點類似，下面是 AOF 文件重寫流程圖：

AOF 文件重寫也是交給子進程來完成，跟 RDB 生成快照很像，AOF 文件重寫在重寫期間建立了一個 aof_rewrite_buf 緩存區來保存重寫期間主進程響應的命令，等新的 AOF 文件重寫完成后，將這部分文件同步到新的 AOF 文件中，最後用新的 AOF 文件替換掉舊的 AOF 文件。需要注意的是在重寫期間，舊的 AOF 文件依然會進行磁盤同步，這樣做的目的是防止重寫失敗導致數據丟失，

Redis 持久化數據恢復

我們知道 Redis 是基於內存的，所有的數據都存放在內存中，由於機器宕機或者其他因素重啟了就會導致我們的數據全部丟失，這也就是要做持久化的原因，當服務器重啟時，Redis 會從持久化文件中加載數據，這樣我們的數據就恢復到了重啟前的數據，在數據恢復這一塊Redis 是如何實現的？我們先來看看數據恢復的流程圖：

Redis 的數據恢複流程比較簡單，優先恢復的是 AOF 文件，如果 AOF 文件不存在時則嘗試加載 RDB 文件，為什麼 RDB 的恢復速度比 AOF 文件快，但是還是會優先加載 AOF 文件呢？我個人認為是 AOF 文件數據更全面並且 AOF 兼容性比 RDB 強，需要注意的是當存在 RDB/AOF 時，如果數據加載不成功，Redis 服務啟動會失敗。

最後

目前互聯網上很多大佬都有 Redis 系列教程，如有雷同，請多多包涵了。原創不易，碼字不易，還希望大家多多支持。若文中有所錯誤之處，還望提出，謝謝。

歡迎掃碼關注微信公眾號：「平頭哥的技術博文」，和平頭哥一起學習，一起進步。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理
【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包”嚨底家”

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※小三通海運與一般國際貿易有何不同?

※小三通快遞通關作業有哪些?

前面介紹過 Thrift 安裝和使用，介紹了Thrift服務的發布和客戶端調用，可以查看我之前的文章：

但是，之前介紹的都是單向的客戶端發送消息，服務端接收消息。而客戶端卻得不到服務器的響應。

那如果我們要實現雙向通信（即：客戶端發送請求，服務端處理返回，服務端發送消息，客戶端處理返回）的功能，該怎麼實現呢？

 

其實在不涉及語言平台的制約，WebService或是webapi 就可以實現這種客戶端發起請求，服務端的處理的單向流程。

然而，實際場景中，可能我們的某些業務需求，更需要服務端能夠響應請求並處理數據。下面我通過一個demo案例，介紹下Thrift 是如何實現雙向通信的。

 

一、安裝Thrift

這裏不再贅述，戳這裏查看我上篇文章的介紹：

 

二、編寫Thrift IDL文件 

編寫thrift腳本，命名為student.thrift  如下：

service HelloWorldService{
    void SayHello(1:string msg);
}

生成service 的方法，之前的文章有介紹，這裏就不介紹了。

 

三、編寫服務端代碼

創建HelloThrift.Server 服務端工程，添加HelloWorldBidirectionServer類，HelloWorldBidirectionServer 實現了Iface接口用於接收客戶端消息，並有一個客戶端傳輸層對象集合用於記錄所有已連接的客戶端。

 public class HelloWorldBidirectionServer : HelloWorldBidirectionService.Iface
    {
        public void Run(int port)
        {
            try
            {
                TServerTransport transport = new TServerSocket(port);

                TTransportFactory transportFac = new TTransportFactory();

                TProtocolFactory inputProtocolFactory = new TBinaryProtocol.Factory();
                TThreadPoolServer server = new TThreadPoolServer(getProcessorFactory(), transport, transportFac, inputProtocolFactory);

                server.Serve();
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
            }
        }

        public static List<TTransport> TransportCollection = new List<TTransport>();

        public void SayHello(string msg)
        {
            Console.WriteLine(string.Format("{0:yyyy/MM/dd hh:mm:ss} 服務端接收到消息： {1}", DateTime.Now, msg));
        }

        public void SayToClient(string msg)
        {
            try
            {
                foreach (TTransport trans in TransportCollection)
                {
                    TBinaryProtocol protocol = new TBinaryProtocol(trans);
                    HelloWorldBidirectionService.Client client = new HelloWorldBidirectionService.Client(protocol);
                    //Thread.Sleep(1000);
                    client.SayHello(msg);
                    //Console.WriteLine("發給了客戶端喲");
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
            }
        }

        public TProcessorFactory getProcessorFactory()
        {
            return new HelloWorldBidirectionProcessor();
        }
    }

    public class HelloWorldBidirectionProcessor : TProcessorFactory
    {
        public TProcessor GetProcessor(TTransport trans, TServer server = null)
        {
            if (trans.IsOpen)
            {
                HelloWorldBidirectionServer.TransportCollection.Add(trans);
                Console.WriteLine("客戶端連上。");
            }

            HelloWorldBidirectionServer srv = new HelloWorldBidirectionServer();
            return new global::HelloWorldBidirectionService.Processor(srv);
        }
    }

 

四、編寫客戶端代碼

首先創建HelloThrift.Client客戶端項目，添加接收服務端消息的類HelloWorldBidirectionClient，裏面只有一個實現Iface接口的方法：

  public class HelloWorldBidirectionClient
    {
        static HelloWorldBidirectionService.Client client = null;
        public void ConnectAndListern(int port, string ip = "127.0.0.1")
        {
            //Tsocket: TCP/IP Socket接口
            TSocket tSocket = new TSocket(ip, port);
            //消息結構協議
            TProtocol protocol = new TBinaryProtocol(tSocket);
            try
            {
                if (client == null)
                {
                    client = new global::HelloWorldBidirectionService.Client(protocol);
                    tSocket.Open();//建立連接
                    StartListern(tSocket);//啟動監聽線程
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
            }
        }

        public void Say(string msg)
        {
            if (client != null)
                client.SayHello(msg);
        }

        void StartListern(TSocket tSocket)
        {
            Thread t = new Thread(new ParameterizedThreadStart(Run));
            t.Start(tSocket);
        }

        public void Run(object tSocket)
        {
            HelloWorldBidirectionService.Processor process = new HelloWorldBidirectionService.Processor(new HelloWorldBidirectionFace());

            try
            {
                while (process.Process(new TBinaryProtocol((TSocket)tSocket), new TBinaryProtocol((TSocket)tSocket)))
                {
                    Console.WriteLine("消息接收完成，等下一波，阻塞中......");
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("連接斷開..." + ex.Message);
            }
        }

    }
    class HelloWorldBidirectionFace : HelloWorldBidirectionService.Iface
    {
        public void SayHello(string msg)
        {
            Console.WriteLine(string.Format("{0:yyyy/MM/dd hh:mm:ss} 收到服務端響應消息 {1}", DateTime.Now, msg));

        }
    }

 實現客戶端，ConnectAndListern方法可以與服務端建立連接，並開啟客戶端端口監聽來自服務端的信息。Say方法可將消息發送至服務端。

 

五、測試

 測試效果如下：

 

 

 

六、最後

　　1. 關於使用Thrift 構建我們自己的rpc 的方法，這裏基本講完了。其他的方法本文就不再演示了，調用起來都是一樣。  

　　2. 後續會簡單討論一下Thrift 框架的通信原理。

　　3. 源代碼下載，

 

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理
【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包”嚨底家”

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※小三通海運與一般國際貿易有何不同?

※小三通快遞通關作業有哪些?

NioEventLoop的創建

NioEventLoop是netty及其重要的組成部件，它的首要職責就是為註冊在它上的channels服務，發現這些channels上發生的新連接、讀寫等I/O事件，然後將事件轉交 channel 流水線處理。使用netty時，我們首先要做的就是創建NioEventLoopGroup，這是一組NioEventLoop的集合，類似線程與線程池。通常，服務端會創建2個group,一個叫做bossGroup,一個叫做workerGroup。bossGroup負責監聽綁定的端口，接受請求並創建新連接，初始化后交由workerGroup處理後續IO事件。

NioEventLoop和NioEventLoopGroup的類圖

首先看看NioEventLoop和NioEventLoopGroup的類關係圖

類多但不亂，可以發現三個特點：

1. 兩者都繼承了ExecutorService，從而與線程池建立了聯繫
2. NioEventLoop繼承的都是SingleThread，NioEventLoop繼承的是MultiThread
3. NioEventLoop還繼承了AbstractScheduledEventExecutor，不難猜出這是個和定時任務調度有關的線程池

NioEventLoopGroup的創建

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

我們先看看bossGroup和workerGroup的構造方法。

public NioEventLoopGroup() {
    this(0);
}
public NioEventLoopGroup(int nThreads) {
    this(nThreads, (Executor) null);
}
除此之外，還有多達8種構造方法，這些構造方法可以指定5種參數：
1、最大線程數量。如果指定為0，那麼Netty會將線程數量設置為CPU邏輯處理器數量的2倍
2、線程工廠。要求線程工廠類必須實現java.util.concurrent.ThreadFactory接口。如果沒有指定線程工廠，那麼默認DefaultThreadFactory。
3、SelectorProvider。如果沒有指定SelectorProvider，那麼默認的SelectorProvider為SelectorProvider.provider()。
4、SelectStrategyFactory。如果沒有指定則默認為DefaultSelectStrategyFactory.INSTANCE
5、RejectedExecutionHandler。拒絕策略處理類，如果這個EventLoopGroup已被關閉，那麼之後提交的Runnable任務會默認調用RejectedExecutionHandler的reject方法進行處理。如果沒有指定，則默認調用拒絕策略。

最終，NioEventLoopGroup會重載到父類MultiThreadEventExecutorGroup的構造方法上,這裏省略了一些健壯性代碼。

protected MultithreadEventExecutorGroup(int nThreads, Executor executor,EventExecutorChooserFactory chooserFactory, Object... args) {
    // 步驟1
    if (executor == null) {
        executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    }
    
    // 步驟2
    children = new EventExecutor[nThreads];
    for (int i = 0; i < nThreads; i ++) {
        children[i] = newChild(executor, args);
    }    

    // 步驟3
    chooser = chooserFactory.newChooser(children);

    // 步驟4
    final FutureListener<Object> terminationListener = future -> {
        if (terminatedChildren.incrementAndGet() == children.length) {
            terminationFuture.setSuccess(null);
        }
    };
    for (EventExecutor e: children) {
        e.terminationFuture().addListener(terminationListener);
    }

    // 步驟5
    Set<EventExecutor> childrenSet = new LinkedHashSet<>(children.length);
    Collections.addAll(childrenSet, children);
    readonlyChildren = Collections.unmodifiableSet(childrenSet);
    }

這裏可以分解為5個步驟，下面一步步講解

步驟1

第一個步驟是創建線程池executor。從workerGroup構造方法可知，默認傳進來的executor為null，所以首先創建executor。newDefaultThreadFactory的作用是設置線程的前綴名和線程優先級，默認情況下，前綴名是nioEventLoopGroup-x-y這樣的命名規則,而線程優先級則是5，處於中間位置。
創建完newDefaultThreadFactory后，進入到ThreadPerTaskExecutor。它直接實現了juc包的線程池頂級接口，從構造方法可以看到它只是簡單的把factory賦值給自己的成員變量。而它實現的接口方法調用了threadFactory的newThread方法。從名字可以看出，它構造了一個thread，並立即啟動thread。

public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
    this.threadFactory = threadFactory;
}
@Override
public void execute(Runnable command) {
    threadFactory.newThread(command).start();
}    

那麼我們回過頭來看下DefaultThreadFactory的newThread方法,發現他創建了一個FastThreadLocalThread。這是netty自定義的一個線程類，顧名思義，netty認為它的性能更快。關於它的解析留待以後。這裏步驟1創建線程池就完成了。總的來說他與我們通常使用的線程池不太一樣，不設置線程池的線程數和任務隊列，而是來一個任務啟動一個線程。(問題：那任務一多豈不是直接線程爆炸？)

@Override
public Thread newThread(Runnable r) {
    Thread t = newThread(FastThreadLocalRunnable.wrap(r), prefix + nextId.incrementAndGet());        
    return t;
}
protected Thread newThread(Runnable r, String name) {
    return new FastThreadLocalThread(threadGroup, r, name);
}

步驟2

步驟2是創建workerGroup中的NioEventLoop。在示例代碼中，傳進來的線程數是0，顯然不可能真的只創建0個nioEventLoop線程。在調用父類MultithreadEventLoopGroup構造函數時，對線程數進行了判斷，若為0，則傳入默認線程數，該值默認為2倍CPU核心數

protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
    super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}
// 靜態代碼塊初始化DEFAULT_EVENT_LOOP_THREADS
static {
    DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads",     NettyRuntime.availableProcessors() * 2));
}    

接下來是通過newChild方法為每一個EventExecutor創建一個對應的NioEventLoop。這個方法傳入了一些args到NioEventLoop中，前三個是在NioEventLoopGroup創建時傳過來的參數。默認值見上文

1. SlectorProvider.provider, 用於創建 Java NIO Selector 對象;
2. SelectStrategyFactory, 選擇策略工廠;
3. RejectedExecutionHandlers, 拒絕執行處理器;
4. EventLoopTaskQueueFactory,任務隊列工廠,默認為null;

進入NioEventLoop的構造函數，如下：

NioEventLoop構造函數
NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
                 SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
                 EventLoopTaskQueueFactory queueFactory) {
        super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
                rejectedExecutionHandler);
        if (selectorProvider == null) {
            throw new NullPointerException("selectorProvider");
        }
        if (strategy == null) {
            throw new NullPointerException("selectStrategy");
        }
        provider = selectorProvider;
        final SelectorTuple selectorTuple = openSelector();
        selector = selectorTuple.selector;
        unwrappedSelector = selectorTuple.unwrappedSelector;
        selectStrategy = strategy;
    }
// 父類構造函數    
protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
                                        boolean addTaskWakesUp, Queue<Runnable> taskQueue,
                                        RejectedExecutionHandler rejectedHandler) {
    super(parent);
    this.addTaskWakesUp = addTaskWakesUp;
    this.maxPendingTasks = DEFAULT_MAX_PENDING_EXECUTOR_TASKS;
    this.executor = ThreadExecutorMap.apply(executor, this);
    this.taskQueue = ObjectUtil.checkNotNull(taskQueue, taskQueue");
    rejectedExecutionHandler = ObjectUtil.checkNotNullrejectedHandler, "rejectedHandler");
}    

首先調用一個newTaskQueue方法創建一個任務隊列。這是一個mpsc即多生產者單消費者的無鎖隊列。之後調用父類的構造函數，在父類的構造函數中，將NioEventLoopGroup設置為自己的parent，並通過匿名內部類創建了這樣一個Executor————通過ThreadPerTaskExecutor執行傳進來的任務，並且在執行時將當前線程與NioEventLoop綁定。其他屬性也一一設置。
在nioEventLoop構造函數中，我們發現創建了一個selector，不妨看一看netty對它的包裝。

unwrappedSelector = provider.openSelector();
if (DISABLE_KEY_SET_OPTIMIZATION) {
    return new SelectorTuple(unwrappedSelector);
}

首先看到netty定義了一個常量DISABLE_KEY_SET_OPTIMIZATION，如果這個常量設置為true，也即不對keyset進行優化，則直接返回未包裝的selector。那麼netty對selector進行了哪些優化？

final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();

final class SelectedSelectionKeySet extends AbstractSet<SelectionKey> {

    SelectionKey[] keys;
    int size;

    SelectedSelectionKeySet() {
        keys = new SelectionKey[1024];
    }
} 

往下，我們看到了一個叫做selectedSelectionKeySet的類，點進去可以看到，它繼承了AbstractSet，然而它的成員變量卻讓我們想到了ArrayList,再看看它定義的方法，除了不支持remove和contains外，活脫脫一個簡化版的ArrayList，甚至也支持擴容。
沒錯，netty確實通過反射的方式，將selectionKey從Set替換為了ArrayList。仔細一想，卻又覺得此番做法有些道理。眾所周知，雖然HashSet和ArrayList隨機查找的時間複雜度都是o(1)，但相比數組直接通過偏移量定位，HashSet由於需要Hash運算，時間消耗上又稍稍遜色了些。再加上使用場景上，都是獲取selectionKey集合然後遍歷,Set去重的特性完全用不上，也無怪乎追求性能的netty想要替換它了。

步驟3

創建完workerGroup的NioEventLoop后,如何挑選一個nioEventLoop進行工作是netty接下來想要做的事。一般來說輪詢是一個很容易想到的方案，為此需要創建一個類似負載均衡作用的線程選擇器。當然追求性能到喪心病狂的netty是不會輕易滿足的。我們看看netty在這樣常見的方案里又做了哪些操作。

public EventExecutorChooser newChooser(EventExecutor[] executors) {
    if (isPowerOfTwo(executors.length)) {
        return new PowerOfTwoEventExecutorChooser(executors);
    } else {
        return new GenericEventExecutorChooser(executors);
    }
}
// PowerOfTwo
public EventExecutor next() {
    return executors[idx.getAndIncrement() & executors.length - 1];
}
// Generic
public EventExecutor next() {
    return executors[Math.abs(idx.getAndIncrement() % executors.length)];
}

可以看到，netty根據workerGroup內線程的數量採取了2種不同的線程選擇器，當線程數x是2的冪次方時，可以通過&(x-1)來達到對x取模的效果，其他情況則需要直接取模。這與hashmap強制設置容量為2的冪次方有異曲同工之妙。

步驟4

步驟4就是添加一些保證健壯性而添加的監聽器了，這些監聽器會在EventLoop被關閉后得到通知。

步驟5

創建一個只讀的NioEventLoop線程組

到此NioEventLoopGroup及其包含的NioEventLoop組就創建完成了

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理
【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包”嚨底家”

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

※小三通海運與一般國際貿易有何不同?

※小三通快遞通關作業有哪些?