一、Kafka簡(jiǎn)介

Kafka是一種高吞吐量、分布式、基于發(fā)布/訂閱的消息系統(tǒng)，最初由LinkedIn公司開發(fā)，使用Scala語言編寫，目前是Apache的開源項(xiàng)目。

跟RabbitMQ、RocketMQ等目前流行的開源消息中間件相比，Kakfa具有高吞吐、低延遲等特點(diǎn)，在大數(shù)據(jù)、日志收集等應(yīng)用場(chǎng)景下被廣泛使用。

本文主要簡(jiǎn)單介紹Kafka的設(shè)計(jì)原理。

二、Kafka架構(gòu)

基本概念：

• broker：Kafka服務(wù)器，負(fù)責(zé)消息存儲(chǔ)和轉(zhuǎn)發(fā)
• topic：消息類別，Kafka按照topic來分類消息
• partition：topic的分區(qū)，一個(gè)topic可以包含多個(gè)partition，topic消息保存在各個(gè)partition上
• offset：消息在日志中的位置，可以理解是消息在partition上的偏移量，也是代表該消息的唯一序號(hào)
• Producer：消息生產(chǎn)者
• Consumer：消息消費(fèi)者
• Consumer Group：消費(fèi)者分組，每個(gè)Consumer必須屬于一個(gè)group
• Zookeeper：保存著集群broker、topic、partition等meta數(shù)據(jù);另外，還負(fù)責(zé)broker故障發(fā)現(xiàn)，partition leader選舉，負(fù)載均衡等功能

三、Kafka設(shè)計(jì)原理

3.1 數(shù)據(jù)存儲(chǔ)設(shè)計(jì)

partition以文件形式存儲(chǔ)在文件系統(tǒng)，目錄命名規(guī)則：<topic_name>-<partition_id>，例如，名為test的topic，其有3個(gè)partition，則Kafka數(shù)據(jù)目錄中有3個(gè)目錄：test-0, test-1, test-2，分別存儲(chǔ)相應(yīng)partition的數(shù)據(jù)。

partition的數(shù)據(jù)文件

partition中的每條Message包含了以下三個(gè)屬性：

• offset
• MessageSize
• data

其中offset表示Message在這個(gè)partition中的偏移量，offset不是該Message在partition數(shù)據(jù)文件中的實(shí)際存儲(chǔ)位置，而是邏輯上一個(gè)值，它唯一確定了partition中的一條Message，可以認(rèn)為offset是partition中Message的id;MessageSize表示消息內(nèi)容data的大小;data為Message的具體內(nèi)容。

partition的數(shù)據(jù)文件由以上格式的Message組成，按offset由小到大排列在一起。

如果一個(gè)partition只有一個(gè)數(shù)據(jù)文件：

1. 新數(shù)據(jù)是添加在文件末尾，不論文件數(shù)據(jù)文件有多大，這個(gè)操作永遠(yuǎn)都是O(1)的。
2. 查找某個(gè)offset的Message是順序查找的。因此，如果數(shù)據(jù)文件很大的話，查找的效率就低。

Kafka通過分段和索引來提高查找效率。

數(shù)據(jù)文件分段segment

partition物理上由多個(gè)segment文件組成，每個(gè)segment大小相等，順序讀寫。每個(gè)segment數(shù)據(jù)文件以該段中最小的offset命名，文件擴(kuò)展名為.log。這樣在查找指定offset的Message的時(shí)候，用二分查找就可以定位到該Message在哪個(gè)segment數(shù)據(jù)文件中。

數(shù)據(jù)文件索引

數(shù)據(jù)文件分段使得可以在一個(gè)較小的數(shù)據(jù)文件中查找對(duì)應(yīng)offset的Message了，但是這依然需要順序掃描才能找到對(duì)應(yīng)offset的Message。為了進(jìn)一步提高查找的效率，Kafka為每個(gè)分段后的數(shù)據(jù)文件建立了索引文件，文件名與數(shù)據(jù)文件的名字是一樣的，只是文件擴(kuò)展名為.index。

索引文件中包含若干個(gè)索引條目，每個(gè)條目表示數(shù)據(jù)文件中一條Message的索引。索引包含兩個(gè)部分，分別為相對(duì)offset和position。

• 相對(duì)offset：因?yàn)閿?shù)據(jù)文件分段以后，每個(gè)數(shù)據(jù)文件的起始o(jì)ffset不為0，相對(duì)offset表示這條Message相對(duì)于其所屬數(shù)據(jù)文件中最小的offset的大小。舉例，分段后的一個(gè)數(shù)據(jù)文件的offset是從20開始，那么offset為25的Message在index文件中的相對(duì)offset就是25-20 = 5。存儲(chǔ)相對(duì)offset可以減小索引文件占用的空間。
• position，表示該條Message在數(shù)據(jù)文件中的絕對(duì)位置。只要打開文件并移動(dòng)文件指針到這個(gè)position就可以讀取對(duì)應(yīng)的Message了。

index文件中并沒有為數(shù)據(jù)文件中的每條Message建立索引，而是采用了稀疏存儲(chǔ)的方式，每隔一定字節(jié)的數(shù)據(jù)建立一條索引。這樣避免了索引文件占用過多的空間，從而可以將索引文件保留在內(nèi)存中。但缺點(diǎn)是沒有建立索引的Message也不能一次定位到其在數(shù)據(jù)文件的位置，從而需要做一次順序掃描，但是這次順序掃描的范圍就很小了。

總結(jié)

查找某個(gè)offset的消息，先二分法找出消息所在的segment文件(因?yàn)槊總€(gè)segment的命名都是以該文件中消息offset最小的值命名);然后，加載對(duì)應(yīng)的.index索引文件到內(nèi)存，同樣二分法找出小于等于給定offset的***的那個(gè)offset記錄(相對(duì)offset，position);***，根據(jù)position到.log文件中，順序查找出offset等于給定offset值的消息。

由于消息在partition的segment數(shù)據(jù)文件中是順序讀寫的，且消息消費(fèi)后不會(huì)刪除(刪除策略是針對(duì)過期的segment文件)，這種順序磁盤IO存儲(chǔ)設(shè)計(jì)是Kafka高性能很重要的原因。

3.2 生產(chǎn)者設(shè)計(jì)

• 負(fù)載均衡：由于消息topic由多個(gè)partition組成，且partition會(huì)均衡分布到不同broker上，因此，為了有效利用broker集群的性能，提高消息的吞吐量，producer可以通過隨機(jī)或者h(yuǎn)ash等方式，將消息平均發(fā)送到多個(gè)partition上，以實(shí)現(xiàn)負(fù)載均衡。
• 批量發(fā)送：是提高消息吞吐量重要的方式，Producer端可以在內(nèi)存中合并多條消息后，以一次請(qǐng)求的方式發(fā)送了批量的消息給broker，從而大大減少broker存儲(chǔ)消息的IO操作次數(shù)。但也一定程度上影響了消息的實(shí)時(shí)性，相當(dāng)于以時(shí)延代價(jià)，換取更好的吞吐量。

3.3 消費(fèi)者設(shè)計(jì)

• 任何Consumer必須屬于一個(gè)Consumer Group
• 同一Consumer Group中的多個(gè)Consumer實(shí)例，不同時(shí)消費(fèi)同一個(gè)partition，等效于隊(duì)列模式。如圖，Consumer Group 1的三個(gè)Consumer實(shí)例分別消費(fèi)不同的partition的消息，即，TopicA-part0、TopicA-part1、TopicA-part2。
• 不同Consumer Group的Consumer實(shí)例可以同時(shí)消費(fèi)同一個(gè)partition，等效于發(fā)布訂閱模式。如圖，Consumer Group 1的Consumer1和Consumer Group 2的Consumer4，同時(shí)消費(fèi)TopicA-part0的消息。
• partition內(nèi)消息是有序的，Consumer通過pull方式消費(fèi)消息。
• Kafka不刪除已消費(fèi)的消息

隊(duì)列模式

隊(duì)列模式，指每條消息只會(huì)有一個(gè)Consumer消費(fèi)到。Kafka保證同一Consumer Group中只有一個(gè)Consumer會(huì)消費(fèi)某條消息。

• 在Consumer Group穩(wěn)定狀態(tài)下，每一個(gè)Consumer實(shí)例只會(huì)消費(fèi)某一個(gè)或多個(gè)特定partition的數(shù)據(jù)，而某個(gè)partition的數(shù)據(jù)只會(huì)被某一個(gè)特定的Consumer實(shí)例所消費(fèi)，也就是說Kafka對(duì)消息的分配是以partition為單位分配的，而非以每一條消息作為分配單元;
• 同一Consumer Group中，如果Consumer實(shí)例數(shù)量少于partition數(shù)量，則至少有一個(gè)Consumer會(huì)消費(fèi)多個(gè)partition的數(shù)據(jù);如果Consumer的數(shù)量與partition數(shù)量相同，則正好一個(gè)Consumer消費(fèi)一個(gè)partition的數(shù)據(jù);而如果Consumer的數(shù)量多于partition的數(shù)量時(shí)，會(huì)有部分Consumer無法消費(fèi)該Topic下任何一條消息;
• 設(shè)計(jì)的優(yōu)勢(shì)是：每個(gè)Consumer不用都跟大量的broker通信，減少通信開銷，同時(shí)也降低了分配難度，實(shí)現(xiàn)也更簡(jiǎn)單;可以保證每個(gè)partition里的數(shù)據(jù)可以被Consumer有序消費(fèi)。
• 設(shè)計(jì)的劣勢(shì)是：無法保證同一個(gè)Consumer Group里的Consumer均勻消費(fèi)數(shù)據(jù)，且在Consumer實(shí)例多于partition個(gè)數(shù)時(shí)導(dǎo)致有些Consumer會(huì)餓死。

如果有partition或者Consumer的增減，為了保證均衡消費(fèi)，需要實(shí)現(xiàn)Consumer Rebalance，分配算法如下：

broker對(duì)Consumer設(shè)計(jì)原理：

• 對(duì)于每個(gè)Consumer Group，選舉出一個(gè)Broker作為Coordinator(0.9版本以上)，由它Watch Zookeeper，從而監(jiān)控判斷是否有partition或者Consumer的增減，然后生成Rebalance命令，按照以上算法重新分配。
• 當(dāng)Consumer Group***次被初始化時(shí)，Consumer通常會(huì)讀取每個(gè)partition的最早或最近的offset(Zookeeper記錄)，然后順序地讀取每個(gè)partition log的消息，在Consumer讀取過程中，它會(huì)提交已經(jīng)成功處理的消息的offsets(由Zookeeper記錄)。
• 當(dāng)一個(gè)partition被重新分配給Consumer Group中的其他Consumer，新的Consumer消費(fèi)的初始位置會(huì)設(shè)置為(原來Consumer)最近提交的offset。

如圖，Last Commited Offset指Consumer最近一次提交的消費(fèi)記錄offset，Current Position是當(dāng)前消費(fèi)的位置，High Watermark是成功拷貝到log的所有副本節(jié)點(diǎn)(partition的所有ISR節(jié)點(diǎn)，下文介紹)的最近消息的offset，Log End Offset是寫入log中***一條消息的offset+1。

從Consumer的角度來看，最多只能讀取到High watermark的位置，后面的消息對(duì)消費(fèi)者不可見，因?yàn)槲赐耆珡?fù)制的數(shù)據(jù)還沒可靠存儲(chǔ)，有丟失可能。

發(fā)布訂閱模式

發(fā)布訂閱模式，又指廣播模式，Kafka保證topic的每條消息會(huì)被所有Consumer Group消費(fèi)到，而對(duì)于同一個(gè)Consumer Group，還是保證只有一個(gè)Consumer實(shí)例消費(fèi)到這條消息。

3.4 Replication設(shè)計(jì)

作為消息中間件，數(shù)據(jù)的可靠性以及系統(tǒng)的可用性，必然依賴數(shù)據(jù)副本的設(shè)計(jì)。

Kafka的replica副本單元是topic的partition，一個(gè)partition的replica數(shù)量不能超過broker的數(shù)量，因?yàn)橐粋€(gè)broker最多只會(huì)存儲(chǔ)這個(gè)partition的一個(gè)副本。所有消息生產(chǎn)、消費(fèi)請(qǐng)求都是由partition的leader replica來處理，其他follower replica負(fù)責(zé)從leader復(fù)制數(shù)據(jù)進(jìn)行備份。

Replica均勻分布到整個(gè)集群，Replica的算法如下：

• 將所有Broker(假設(shè)共n個(gè)Broker)和待分配的Partition排序
• 將第i個(gè)Partition分配到第(i mod n)個(gè)Broker上
• 將第i個(gè)Partition的第j個(gè)Replica分配到第((i + j) mode n)個(gè)Broker上

如圖，TopicA有三個(gè)partition：part0、part1、part2，每個(gè)partition的replica數(shù)等于2(一個(gè)是leader，另一個(gè)是follower)，按照以上算法會(huì)均勻落到三個(gè)broker上。

broker對(duì)replica管理：

選舉出一個(gè)broker作為controller，由它Watch Zookeeper，負(fù)責(zé)partition的replica的集群分配，以及l(fā)eader切換選舉等流程。

In-Sync-Replica(ISR)

分布式系統(tǒng)在處理節(jié)點(diǎn)故障時(shí)，需要預(yù)先明確節(jié)點(diǎn)的”failure”和”alive”的定義。對(duì)于Kafka節(jié)點(diǎn)，判斷是”alive”有以下兩個(gè)條件：

• 節(jié)點(diǎn)必須和Zookeeper保持心跳連接
• 如果節(jié)點(diǎn)是follower，必須從leader節(jié)點(diǎn)上復(fù)制數(shù)據(jù)來備份，而且備份的數(shù)據(jù)相比leader而言，不能落后太多。

Kafka將滿足以上條件的replica節(jié)點(diǎn)認(rèn)為是”in sync”(同步中)，稱為In-Sync-Replica(ISR)。

Kafka的Zookeeper維護(hù)了每個(gè)partition的ISR信息，理想情況下，ISR包含了partition的所有replica所在的broker節(jié)點(diǎn)信息，而當(dāng)某些節(jié)點(diǎn)不滿足以上條件時(shí)，ISR可能只包含部分replica。例如，上圖中的TopicA-part0的ISR列表可能是[broker1,broker2,broker3]，也可能是[broker1,broker3]和[broker1]。

數(shù)據(jù)可靠性

Kafka如何保證數(shù)據(jù)可靠性?首先看下，Producer生產(chǎn)一條消息，該消息被認(rèn)為是”committed”(即broker認(rèn)為消息已經(jīng)可靠存儲(chǔ))的過程：

• 消息所在partition的ISR replicas會(huì)定時(shí)異步從leader上批量復(fù)制數(shù)據(jù)log
• 當(dāng)所有ISR replica都返回ack，告訴leader該消息已經(jīng)寫log成功后，leader認(rèn)為該消息committed，并告訴Producer生產(chǎn)成功。這里和以上”alive”條件的第二點(diǎn)是不矛盾的，因?yàn)閘eader有超時(shí)機(jī)制，leader等ISR的follower復(fù)制數(shù)據(jù)，如果一定時(shí)間不返回ack(可能數(shù)據(jù)復(fù)制進(jìn)度落后太多)，則leader將該follower replica從ISR中剔除。
• 消息committed之后，Consumer才能消費(fèi)到。

ISR機(jī)制下的數(shù)據(jù)復(fù)制，既不是完全的同步復(fù)制，也不是單純的異步復(fù)制，這是Kafka高吞吐很重要的機(jī)制。同步復(fù)制要求所有能工作的follower都復(fù)制完，這條消息才會(huì)被認(rèn)為committed，這種復(fù)制方式極大的影響了吞吐量。而異步復(fù)制方式下，follower異步的從leader復(fù)制數(shù)據(jù)，數(shù)據(jù)只要被leader寫入log就被認(rèn)為已經(jīng)committed，這種情況下如果follower都復(fù)制完都落后于leader，而如果leader突然宕機(jī)，則會(huì)丟失數(shù)據(jù)。而Kafka的這種使用ISR的方式則很好的均衡了確保數(shù)據(jù)不丟失以及吞吐量，follower可以批量的從leader復(fù)制數(shù)據(jù)，數(shù)據(jù)復(fù)制到內(nèi)存即返回ack，這樣極大的提高復(fù)制性能，當(dāng)然數(shù)據(jù)仍然是有丟失風(fēng)險(xiǎn)的。

Kafka本身定位于高性能的MQ，更多注重消息吞吐量，在此基礎(chǔ)上結(jié)合ISR的機(jī)制去盡量保證消息的可靠性，但不是絕對(duì)可靠的。

服務(wù)可用性

Kafka所有收發(fā)消息請(qǐng)求都由leader節(jié)點(diǎn)處理，由以上數(shù)據(jù)可靠性設(shè)計(jì)可知，當(dāng)ISR的follower replica故障后，leader會(huì)及時(shí)地從ISR列表中把它剔除掉，并不影響服務(wù)可用性，那么當(dāng)leader故障后會(huì)怎樣呢?如何選舉新的leader?

leader選舉

• Kafka在Zookeeper存儲(chǔ)partition的ISR信息，并且能動(dòng)態(tài)調(diào)整ISR列表的成員，只有ISR里的成員replica才會(huì)被選為leader，并且ISR所有的replica都有可能成為leader;
• leader節(jié)點(diǎn)宕機(jī)后，Zookeeper能監(jiān)控發(fā)現(xiàn)，并由broker的controller節(jié)點(diǎn)從ISR中選舉出新的leader，并通知ISR內(nèi)的所有broker節(jié)點(diǎn)。

因此，可以看出，只要ISR中至少有一個(gè)replica，Kafka就能保證服務(wù)的可用性(但不保證網(wǎng)絡(luò)分區(qū)下的可用性)。

容災(zāi)和數(shù)據(jù)一致性

分布式系統(tǒng)的容災(zāi)能力，跟其本身針對(duì)數(shù)據(jù)一致性考慮所選擇的算法有關(guān)，例如，Zookeeper的Zab算法，raft算法等。Kafka的ISR機(jī)制和這些Majority Vote算法對(duì)比如下：

ISR機(jī)制能容忍更多的節(jié)點(diǎn)失敗。假如replica節(jié)點(diǎn)有2f+1個(gè)，每個(gè)partition最多能容忍2f個(gè)失敗，且不丟失消息數(shù)據(jù);但相對(duì)Majority Vote選舉算法，只能最多容忍f個(gè)失敗。

在消息committed持久化上，ISR需要等2f個(gè)節(jié)點(diǎn)返回ack，但Majority Vote只需等f+1個(gè)節(jié)點(diǎn)返回ack，且不依賴處理最慢的follower節(jié)點(diǎn)，因此Majority Vote有優(yōu)勢(shì)

ISR機(jī)制能節(jié)省更多replica節(jié)點(diǎn)數(shù)。例如，要保證f個(gè)節(jié)點(diǎn)可用，ISR方式至少要f個(gè)節(jié)點(diǎn)，而Majority Vote至少需要2f+1個(gè)節(jié)點(diǎn)。

如果所有replica都宕機(jī)了，有兩種方式恢復(fù)服務(wù)：

1. 等ISR任一節(jié)點(diǎn)恢復(fù)，并選舉為leader;
2. 選擇***個(gè)恢復(fù)的節(jié)點(diǎn)(不一定是ISR中的節(jié)點(diǎn))為leader

***種方式消息不會(huì)丟失(只能說這種方式最有可能不丟而已)，第二種方式可能會(huì)丟消息，但能盡快恢復(fù)服務(wù)可用。這是可用性和一致性場(chǎng)景的兩種考慮，Kafka默認(rèn)選擇第二種，用戶也可以自主配置。

大部分考慮CP的分布式系統(tǒng)(假設(shè)2f+1個(gè)節(jié)點(diǎn))，為了保證數(shù)據(jù)一致性，最多只能容忍f個(gè)節(jié)點(diǎn)的失敗，而Kafka為了兼顧可用性，允許最多2f個(gè)節(jié)點(diǎn)失敗，因此是無法保證數(shù)據(jù)強(qiáng)一致的。

如圖所示，一開始ISR數(shù)量等于3，正常同步數(shù)據(jù)，紅色部分開始，leader發(fā)現(xiàn)其他兩個(gè)follower復(fù)制進(jìn)度太慢或者其他原因(網(wǎng)絡(luò)分區(qū)、節(jié)點(diǎn)故障等)，將其從ISR剔除后，leader單節(jié)點(diǎn)存儲(chǔ)數(shù)據(jù);然后，leader宕機(jī)，觸發(fā)重新選舉第二節(jié)點(diǎn)為leader，重新開始同步數(shù)據(jù)，但紅色部分的數(shù)據(jù)在新leader上是沒有的;***原leader節(jié)點(diǎn)恢復(fù)服務(wù)后，重新從新leader上復(fù)制數(shù)據(jù)，而紅色部分的數(shù)據(jù)已經(jīng)消費(fèi)不到了。

因此，為了減少數(shù)據(jù)丟失的概率，可以設(shè)置Kafka的ISR最小replica數(shù)，低于該值后直接返回不可用，當(dāng)然是以犧牲一定可用性和吞吐量為前提了。

重復(fù)消息

消息傳輸有三種方式：

At most once：消息可能會(huì)丟失，但不會(huì)重復(fù)傳輸

At least once：消息不會(huì)丟失，但可能重復(fù)傳輸

Exactly once：消息保證會(huì)被傳輸一次且僅傳輸一次

Kafka實(shí)現(xiàn)了第二種方式，即，可能存在重復(fù)消息，需要業(yè)務(wù)自己保證消息冪等性處理。

3.5 高吞吐設(shè)計(jì)

1. 對(duì)于partition，順序讀寫磁盤數(shù)據(jù)，以時(shí)間復(fù)雜度O(1)方式提供消息持久化能力。
2. Producer批量向broker寫數(shù)據(jù)
3. Consumer批量從broker拉數(shù)據(jù)
4. 日志壓縮
5. Topic分多個(gè)partition，提高并發(fā)
6. broker零拷貝(Zero Copy)，使用sendfile系統(tǒng)調(diào)用，將數(shù)據(jù)直接從page cache發(fā)送到socket上
7. Producer可配置是否等待消息committed。如果Producer生產(chǎn)消息，每次都必須等ISR存儲(chǔ)后才返回，時(shí)延會(huì)很高，進(jìn)而影響整體消息的吞吐量。為了解決這個(gè)問題，一方面Producer可以配置減少partition的副本數(shù)，例如，ISR大小為1;另一方面，在不太關(guān)注消息可靠存儲(chǔ)的場(chǎng)景下，Producer可以通過配置選擇是否等待消息committed，如下：

這是用戶在消息吞吐量和持久化之間做的權(quán)衡選擇，持久化等級(jí)越高，生產(chǎn)消息吞吐量越小，反之，持久化等級(jí)越低，吞吐量越高。

3.6 HA基本原理

broker HA

broker集群信息由Zookeeper維護(hù)，并選舉出一個(gè)controller。所有partition的leader選舉都由controller決定，將leader的變更直接通過rpc方式通知需要為此做出響應(yīng)的brokers;controller也負(fù)責(zé)增刪topic以及partition replica的重新分配。

controller在Zookeeper上注冊(cè)watch，一旦有broker宕機(jī)，其對(duì)應(yīng)在Zookeeper的臨時(shí)節(jié)點(diǎn)自動(dòng)被刪除，controller對(duì)宕機(jī)broker上的所有partition重新分配新leader;如果controller宕機(jī)，其他broker通過Zookeeper選舉出新的controller，然后同樣對(duì)宕機(jī)broker上的所有partition重新分配新leader。

partition HA

partition leader所在的broker宕機(jī)，如上所述，broker controller根據(jù)動(dòng)態(tài)維護(hù)的ISR，會(huì)重新在剩下的broker機(jī)器中選出ISR里面的一個(gè)成員成為新的leader。如果ISR中至少有一個(gè)follower，則可以確保已經(jīng)committed的數(shù)據(jù)不丟失;否則選擇任意一個(gè)replica作為leader，該場(chǎng)景可能會(huì)有潛在的數(shù)據(jù)丟失;如果partition所有的replica都宕機(jī)了，就無法保證數(shù)據(jù)不丟失了，有兩種恢復(fù)方案，上文已介紹過。

四、推廣

騰訊云即將推出高性能的消息隊(duì)列服務(wù)Ckafka，完全兼容開源Kafka API(0.9版本)。Ckafka服務(wù)端完全托管在騰訊云上，用戶無需自己維護(hù)和搭建，使用開源Kafka API客戶端即可訪問實(shí)例，大大降低了用戶使用Kafka的門檻，歡迎體驗(yàn)：)

原文鏈接：https://cloud.tencent.com/community/article/369570

【本文是51CTO專欄作者“騰訊云技術(shù)社區(qū)”的原創(chuàng)稿件，轉(zhuǎn)載請(qǐng)通過51CTO聯(lián)系原作者獲取授權(quán)】

戳這里，看該作者更多好文