Kafka 鏍稿績缁勪欢涔嬪崗璋冨櫒

Posted 2021-03-04

鏍囩锛?a href='http://www.mamicode.com/so/1/time' title='time'>time   绾跨▼   閫氫俊   request璇锋眰   浣嶇疆   閫夋嫨   閬垮厤   sync   

1銆佹秷璐硅€呬笌娑堣垂鑰呯粍

鍋囪鏌?topic 鏈?涓垎鍖猴紝娑堣垂鑰呯粍涓彧鏈変竴涓秷璐硅€咃紝閭ｄ箞杩欎釜娑堣垂鑰呭皢娑堣垂鍏ㄩ儴 partition 涓殑鏁版嵁銆?/p>

濡傛灉娑堣垂鑰呯粍涓湁涓や釜娑堣垂鑰咃紝閭ｄ箞姣忎釜娑堣垂鑰呮秷璐逛袱涓?partition銆?/p>

濡傛灉娑堣垂鑰呯粍涓湁4涓秷璐硅€咃紝閭ｄ箞姣忎釜娑堣垂鑰呮秷璐逛竴涓猵artition銆?/p>

濡傛灉娑堣垂鑰呯粍涓湁5涓秷璐硅€咃紝閭ｄ箞鏈変竴涓秷璐硅€呭氨鏄┖闂茬殑銆?/p>

娉ㄦ剰锛氬湪鍚屼竴涓秷璐硅€呯粍涓紝涓嶈璁╂秷璐硅€呯殑鏁伴噺澶т簬鍒嗗尯鐨勬暟閲?/strong>

鍦?kafka-0.10 鐗堟湰锛孠afka 鍦?strong>鏈嶅姟绔紩鍏ヤ簡缁勫崗璋冨櫒(GroupCoordinator)锛屾瘡涓?Kafka Server 鍚姩鏃堕兘浼氬垱寤轰竴涓?GroupCoordinator 瀹炰緥锛?strong>鐢ㄤ簬绠＄悊閮ㄥ垎娑堣垂鑰呯粍鍜岃娑堣垂鑰呯粍涓嬬殑姣忎釜娑堣垂鑰呯殑娑堣垂鍋忕Щ閲?/strong>銆?/p>

鍦ㄥ鎴风寮曞叆浜嗘秷璐硅€呭崗璋冨櫒(ConsumerCoordinator)锛屽疄渚嬪寲涓€涓秷璐硅€呭氨浼氬疄渚嬪寲涓€涓?ConsumerCoordinator 瀵硅薄锛孋onsumerCoordinator 璐熻矗鍚屼竴涓秷璐硅€呯粍涓嬪悇娑堣垂鑰呬笌鏈嶅姟绔殑 GroupCoordinator 杩涜閫氫俊銆?/p>

2.1 娑堣垂鑰呭崗璋冨櫒(ConsumerCoordinator)

ConsumerCoordinator 瀹氫箟鐨勪綅缃細

public class KafkaConsumer<K, V> implements Consumer<K, V> {
    
    private final ConsumerCoordinator coordinator;
    
}

ConsumerCoordinator 鏄?KafkaConsumer 鐨勪竴涓鏈夌殑鎴愬憳鍙橀噺锛屽洜姝?ConsumerCoordinator 涓瓨鍌ㄧ殑淇℃伅涔熷彧鏈変笌涔嬪搴旂殑娑堣垂鑰呭彲瑙侊紝涓嶅悓娑堣垂鑰呬箣闂存槸鐪嬩笉鍒板郊姝ょ殑 ConsumerCoordinator 涓殑淇℃伅鐨?/p>

ConsumerCoordinator 鐨勪綔鐢細

• 澶勭悊鏇存柊娑堣垂鑰呯紦瀛樼殑 Metadata 璇锋眰
• 鍚戠粍鍗忚皟鍣ㄥ彂璧峰姞鍏ユ秷璐硅€呯粍鐨勮姹?/li>
• 瀵规湰娑堣垂鑰呭姞鍏ユ秷璐硅€呭墠鍚庣殑鐩稿簲澶勭悊
• 璇锋眰绂诲紑娑堣垂鑰呯粍(渚嬪褰撴秷璐硅€呭彇娑堣闃呮椂)
• 鍚戠粍鍗忚皟鍣ㄥ彂閫佹彁浜ゅ亸绉婚噺鐨勮姹?/li>
• 閫氳繃涓€涓畾鏃剁殑蹇冭烦妫€娴嬩换鍔℃潵璁╃粍鍗忚皟鍣ㄦ劅鐭ヨ嚜宸辩殑杩愯鐘舵€?/li>
• Leader娑堣垂鑰呯殑 ConsumerCoordinator 杩樿礋璐ｆ墽琛屽垎鍖虹殑鍒嗛厤锛屼竴涓秷璐硅€呯粍涓秷璐硅€?leader 鐢辩粍鍗忚皟鍣ㄩ€夊嚭锛宭eader 娑堣垂鑰呯殑 ConsumerCoordinator 璐熻矗娑堣垂鑰呬笌鍒嗗尯鐨勫垎閰嶏紝鐒跺悗鎶婂垎閰嶇粨鏋滃彂閫佺粰缁勫崗璋冨櫒锛岀劧鍚庣粍鍗忚皟鍣ㄥ啀鎶婂垎閰嶇粨鏋滆繑鍥炵粰鍏朵粬娑堣垂鑰呯殑娑堣垂鑰呭崗璋冨櫒锛岃繖鏍峰噺杞讳簡鏈嶅姟绔殑璐熸媴

ConsumerCoordinator 瀹炵幇涓婅堪鍔熻兘鐨勭粍浠舵槸 ConsumerCoordinator 绫荤殑绉佹湁鎴愬憳鎴栬€呮槸鍏剁埗绫荤殑绉佹湁鎴愬憳锛?/p>

 1 public final class ConsumerCoordinator extends AbstractCoordinator {
 2     private final List<PartitionAssignor> assignors;
 3     private final OffsetCommitCallback defaultOffsetCommitCallback;
 4     private final SubscriptionState subscriptions;
 5     private final ConsumerInterceptors<?, ?> interceptors;
 6     private boolean isLeader = false;
 7     private MetadataSnapshot metadataSnapshot;
 8     private MetadataSnapshot assignmentSnapshot;
 9     
10     鐪佺暐浜嗛儴鍒嗕唬鐮?...
11 }
12 
13 
14 public abstract class AbstractCoordinator implements Closeable {
15     private enum MemberState {
16         UNJOINED,    // the client is not part of a group
17         REBALANCING, // the client has begun rebalancing
18         STABLE,      // the client has joined and is sending heartbeats
19     }
20 
21     private final Heartbeat heartbeat;
22     protected final ConsumerNetworkClient client;
23     private HeartbeatThread heartbeatThread = null;
24     private MemberState state = MemberState.UNJOINED;
25     private RequestFuture<ByteBuffer> joinFuture = null;
26     
27     鐪佺暐浜嗛儴鍒嗕唬鐮?...
28 }

 

2.2 缁勫崗璋冨櫒锛圙roupCoordinator锛?/h3>

GroupCoordinator 鐨勪綔鐢細

• 璐熻矗瀵瑰叾绠＄悊鐨勭粍鍛?娑堣垂鑰?鎻愪氦鐨勭浉鍏宠姹傝繘琛屽鐞?/li>
• 涓庢秷璐硅€呬箣闂村缓绔嬭繛鎺ワ紝骞朵粠涓庝箣杩炴帴鐨勬秷璐硅€呬箣闂撮€夊嚭涓€涓?leader
• 褰?leader 鍒嗛厤濂芥秷璐硅€呬笌鍒嗗尯鐨勮闃呭叧绯诲悗锛屼細鎶婄粨鏋滃彂閫佺粰缁勫崗璋冨櫒锛岀粍鍗忚皟鍣ㄥ啀鎶婄粨鏋滆繑鍥炵粰鍚勪釜娑堣垂鑰?/li>
• 绠＄悊涓庝箣杩炴帴鐨勬秷璐硅€呯殑娑堣垂鍋忕Щ閲忕殑鎻愪氦锛屽皢姣忎釜娑堣垂鑰呯殑娑堣垂鍋忕Щ閲忎繚瀛樺埌kafka鐨勫唴閮ㄤ富棰樹腑
• 閫氳繃蹇冭烦妫€娴嬫秷璐硅€呬笌鑷繁鐨勮繛鎺ョ姸鎬?/li>
• 鍚姩缁勫崗璋冨櫒鐨勬椂鍊欏垱寤轰竴涓畾鏃朵换鍔★紝鐢ㄤ簬娓呯悊杩囨湡鐨勬秷璐圭粍鍏冩暟鎹互鍙婅繃鍘荤殑娑堣垂鍋忕Щ閲忎俊鎭?/li>

GroupCoordinator 渚濊禆鐨勭粍浠跺強鍏朵綔鐢?/strong>

• KafkaConfig锛氬疄渚嬪寲 OffsetConfig 鍜?GroupConfig
• ZkUtils锛氬垎娑堣垂鑰呭垎閰嶇粍鍗忚皟鍣ㄦ椂浠嶼ookeeper鑾峰彇鍐呴儴涓婚鐨勫垎鍖哄厓鏁版嵁淇℃伅銆?/li>
• GroupMetadataManager锛氳礋璐ｇ鐞?GroupMetadata浠ュ強娑堣垂鍋忕Щ閲忕殑鎻愪氦锛屽苟鎻愪緵浜嗕竴绯诲垪鐨勭粍绠＄悊鐨勬柟娉曚緵缁勫崗璋冨櫒璋冪敤銆侴roupMetadataManager 涓嶄粎鎶?GroupMetadata 鍐欏埌kafka鍐呴儴涓婚涓紝鑰屼笖杩樺湪鍐呭瓨涓紦瀛樹簡涓€浠紾roupMetadata锛屽叾涓寘鎷簡缁勫憳(娑堣垂鑰?鐨勫厓鏁版嵁淇℃伅锛屼緥濡傛秷璐硅€呯殑 memberId銆乴eaderId銆佸垎鍖哄垎閰嶅叧绯伙紝鐘舵€佸厓鏁版嵁绛夈€傜姸鎬佸厓鏁版嵁鍙互鏄互涓嬩簲绉嶇姸鎬侊細
  • PreparingRebalance锛氭秷璐圭粍鍑嗗杩涜骞宠　鎿嶄綔
  • AwaitingSync锛氱瓑寰卨eader娑堣垂鑰呭皢鍒嗗尯鍒嗛厤鍏崇郴鍙戦€佺粰缁勫崗璋冨櫒
  • Stable锛氭秷璐硅€呮甯歌繍琛岀姸鎬侊紝蹇冭烦妫€娴嬫甯?/li>
  • Dead锛氬浜庤鐘舵€佺殑娑堣垂缁勬病鏈変换浣曟秷璐硅€呮垚鍛橈紝涓斿厓鏁版嵁淇℃伅涔熷凡缁忚鍒犻櫎
  • Empty锛氬浜庤鐘舵€佺殑娑堣垂缁勬病鏈変换浣曟秷璐硅€呮垚鍛橈紝浣嗗厓鏁版嵁淇℃伅涔熸病鏈夎鍒犻櫎锛岀煡閬撴墍鏈夋秷璐硅€呭搴旂殑娑堣垂鍋忕Щ閲忓厓鏁版嵁淇℃伅杩囨湡銆?/li>
• ReplicaManager锛欸roupMetadataManager闇€瑕佹妸娑堣垂缁勫厓鏁版嵁淇℃伅浠ュ強娑堣垂鑰呮彁浜ょ殑宸叉秷璐瑰亸绉婚噺淇℃伅鍐欏叆 Kafka 鍐呴儴涓婚涓紝瀵瑰唴閮ㄤ富棰樼殑鎿嶄綔涓庡鍏朵粬涓婚鐨勬搷浣滀竴鏍凤紝鍏堥€氳繃 ReplicaManager 灏嗘秷鎭啓鍏?leader 鍓湰锛孯eplicaManager 璐熻矗 leader 鍓湰涓庡叾浠栧壇鏈殑绠＄悊銆?/li>
• DelayedJoin锛氬欢杩熸搷浣滅被锛岀敤浜庣洃瑙嗗鐞嗘墍鏈夋秷璐圭粍鎴愬憳涓庣粍鍗忚皟鍣ㄤ箣闂寸殑蹇冭烦瓒呮椂
• GroupConfig锛氬畾涔変簡缁勬垚鍛樹笌缁勫崗璋冨櫒涔嬮棿session瓒呮椂鏃堕棿閰嶇疆

3銆佹秷璐硅€呭崗璋冨櫒鍜岀粍鍗忚皟鍣ㄧ殑浜や簰

3.1 蹇冭烦

娑堣垂鑰呭崗璋冨櫒閫氳繃鍜岀粍鍗忚皟鍣ㄥ彂閫佸績璺虫潵缁存寔瀹冧滑鍜岀兢缁勭殑浠庡睘鍏崇郴浠ュ強瀹冧滑瀵瑰垎鍖虹殑鎵€鏈夋潈鍏崇郴銆傚彧瑕佹秷璐硅€呬互姝ｅ父鐨勬椂闂撮棿闅斿彂閫佸績璺筹紝灏辫璁や负鏄椿璺冪殑锛岃鏄庡畠杩樺湪璇诲彇鍒嗗尯閲岀殑娑堟伅銆傛秷璐硅€呬細鍦ㄨ疆璇㈣幏鍙栨秷鎭垨鎻愪氦鍋忕Щ閲忔椂鍙戦€佸績璺炽€?/p>

濡傛灉娑堣垂鑰呭仠姝㈠彂閫佸績璺崇殑鏃堕棿瓒冲闀匡紝浼氳瘽灏变細杩囨湡锛岀粍鍗忚皟鍣ㄨ涓哄畠宸茬粡姝讳骸锛屽氨浼氳Е鍙戜竴娆″啀鍧囪　銆?/p>

鍦?0.10 鐗堟湰閲岋紝蹇冭烦浠诲姟鐢变竴涓嫭绔嬬殑蹇冭烦绾跨▼鏉ユ墽琛岋紝鍙互鍦ㄨ疆璇㈣幏鍙栨秷鎭殑绌烘。鍙戦€佸績璺炽€傝繖鏍蜂竴鏉ワ紝鍙戦€佸績璺崇殑棰戠巼锛堜篃灏辨槸缁勫崗璋冨櫒缇ゆ娴嬫秷璐硅€呰繍琛岀姸鎬佺殑鏃堕棿锛変笌娑堟伅杞鐨勯鐜囷紙鐢卞鐞嗘秷鎭墍鑺辫垂鐨勬椂闂存潵纭畾锛変箣闂村氨鏄浉浜掔嫭绔嬬殑銆傚湪0.10 鐗堟湰鐨?Kafka 閲岋紝鍙互鎸囧畾娑堣垂鑰呭湪绂诲紑缇ょ粍骞惰Е鍙戝啀鍧囪　涔嬪墠鍙互鏈夊闀挎椂闂翠笉杩涜娑堟伅杞锛岃繖鏍峰彲浠ラ伩鍏嶅嚭鐜版椿閿侊紙livelock锛夛紝姣斿鏈夋椂鍊欏簲鐢ㄧ▼搴忓苟娌℃湁宕╂簝锛屽彧鏄敱浜庢煇浜涘師鍥犲鑷存棤娉曟甯歌繍琛屻€傝繖涓厤缃笌 session.timeout.ms 鏄浉浜掔嫭绔嬬殑锛屽悗鑰呯敤浜庢帶鍒舵娴嬫秷璐硅€呭彂鐢熷穿婧冪殑鏃堕棿鍜屽仠姝㈠彂閫佸績璺崇殑鏃堕棿銆?/p>

3.2 鍒嗗尯鍐嶅钩琛?/h3>

鍙戠敓鍒嗗尯鍐嶅潎琛＄殑3绉嶆儏鍐碉細

• 涓€涓柊鐨勬秷璐硅€呭姞鍏ョ兢缁勬椂锛屽畠璇诲彇鐨勬槸鍘熸湰鐢卞叾浠栨秷璐硅€呰鍙栫殑娑堟伅銆?/li>
• 褰撲竴涓秷璐硅€呰鍏抽棴鎴栧彂鐢熷穿婧冩椂锛屽畠灏辩寮€缇ょ粍锛屽師鏈敱瀹冭鍙栫殑鍒嗗尯灏嗙敱缇ょ粍閲岀殑鍏朵粬娑堣垂鑰呮潵璇诲彇銆傚鏋滀竴涓秷璐硅€呬富鍔ㄧ寮€娑堣垂缁勶紝娑堣垂鑰呬細閫氱煡缁勫崗璋冨櫒瀹冨皢瑕佺寮€缇ょ粍锛岀粍鍗忚皟鍣ㄤ細绔嬪嵆瑙﹀彂涓€娆″啀鍧囪　锛屽敖閲忛檷浣庡鐞嗗仠椤裤€傚鏋滀竴涓秷璐硅€呮剰澶栧彂鐢熷穿婧冿紝娌℃湁閫氱煡缁勫崗璋冨櫒灏卞仠姝㈣鍙栨秷鎭紝缁勫崗璋冨櫒浼氱瓑寰呭嚑绉掗挓锛岀‘璁ゅ畠姝讳骸浜嗘墠浼氳Е鍙戝啀鍧囪　銆傚湪杩欏嚑绉掗挓鏃堕棿閲岋紝姝绘帀鐨勬秷璐硅€呬笉浼氳鍙栧垎鍖洪噷鐨勬秷鎭€?/li>
• 鍦ㄤ富棰樺彂鐢熷彉鍖栨椂锛屾瘮濡傜鐞嗗憳娣诲姞浜嗘柊鐨勫垎鍖猴紝浼氬彂鐢熷垎鍖洪噸鍒嗛厤銆?/li>

鍒嗗尯鐨勬墍鏈夋潈浠庝竴涓秷璐硅€呰浆绉诲埌鍙︿竴涓秷璐硅€咃紝杩欐牱鐨勮涓鸿绉颁负鍒嗗尯鍐嶅潎琛?/strong>銆?/p>

鍐嶅潎琛￠潪甯搁噸瑕侊紝瀹冧负娑堣垂鑰呯兢缁勫甫鏉ヤ簡楂樺彲鐢ㄦ€у拰浼哥缉鎬э紙鎴戜滑鍙互鏀惧績鍦版坊鍔犳垨绉婚櫎娑堣垂鑰咃級锛屼笉杩囧湪姝ｅ父鎯呭喌涓嬶紝鎴戜滑骞朵笉甯屾湜鍙戠敓杩欐牱鐨勮涓恒€傚湪鍐嶅潎琛℃湡闂达紝娑堣垂鑰呮棤娉曡鍙栨秷鎭紝閫犳垚鏁翠釜缇ょ粍涓€灏忔鏃堕棿鐨勪笉鍙敤銆傚彟澶栵紝褰撳垎鍖鸿閲嶆柊鍒嗛厤缁欏彟涓€涓秷璐硅€呮椂锛屾秷璐硅€呭綋鍓嶇殑璇诲彇鐘舵€佷細涓㈠け锛屽畠鏈夊彲鑳借繕闇€瑕佸幓鍒锋柊缂撳瓨锛屽湪瀹冮噸鏂版仮澶嶇姸鎬佷箣鍓嶄細鎷栨參搴旂敤绋嬪簭銆?/p>

3.3 leader 娑堣垂鑰呭垎閰嶅垎鍖虹殑绛栫暐

褰撴秷璐硅€呰鍔犲叆缇ょ粍鏃讹紝瀹冧細鍚戠兢缁勫崗璋冨櫒鍙戦€佷竴涓?JoinGroup 璇锋眰銆傜涓€涓姞鍏ョ兢缁勭殑娑堣垂鑰呭皢鎴愪负leader娑堣垂鑰呫€俵eader娑堣垂鑰呬粠缁勫崗璋冨櫒閭ｉ噷鑾峰緱缇ょ粍鐨勬垚鍛樺垪琛紙鍒楄〃涓寘鍚簡鎵€鏈夋渶杩戝彂閫佽繃蹇冭烦鐨勬秷璐硅€咃紝瀹冧滑琚涓烘槸娲昏穬鐨勶級锛屽苟璐熻矗缁欐瘡涓€涓秷璐硅€呭垎閰嶅垎鍖恒€?/p>

姣忎釜娑堣垂鑰呯殑娑堣垂鑰呭崗璋冨櫒鍦ㄥ悜缁勫崗璋冨櫒璇锋眰鍔犲叆缁勬椂锛岄兘浼氭妸鑷繁鏀寔鐨勫垎鍖哄垎閰嶇瓥鐣ユ姤鍛婄粰缁勫崗璋冨櫒(杞鎴栬€呮槸鎸夎法搴﹀垎閰嶆垨鑰呭叾浠?锛岀粍鍗忚皟鍣ㄩ€夊嚭璇ユ秷璐圭粍涓嬫墍鏈夋秷璐硅€呴兘鏀寔鐨勭殑鍒嗗尯鍒嗛厤绛栫暐鍙戦€佺粰leader娑堣垂鑰咃紝leader娑堣垂鑰呮牴鎹繖涓垎鍖哄垎閰嶇瓥鐣ヨ繘琛屽垎閰嶃€?/p>

瀹屾瘯涔嬪悗锛宭eader娑堣垂鑰呮妸鍒嗛厤鎯呭喌鍒楄〃鍙戦€佺粰缁勫崗璋冨櫒锛屾秷璐硅€呭崗璋冨櫒鍐嶆妸杩欎簺淇℃伅鍙戦€佺粰鎵€鏈夋秷璐硅€呫€傛瘡涓秷璐硅€呭彧鑳界湅鍒拌嚜宸辩殑鍒嗛厤淇℃伅锛屽彧鏈塴eader娑堣垂鑰呯煡閬撶兢缁勯噷鎵€鏈夋秷璐硅€呯殑鍒嗛厤淇℃伅銆傝繖涓繃绋嬩細鍦ㄦ瘡娆″啀鍧囪　鏃堕噸澶嶅彂鐢熴€?/p>

3.4 娑堣垂鑰呭叆缁勮繃绋?/h3>

• 娑堣垂鑰呭垱寤哄悗锛屾秷璐硅€呭崗璋冨櫒浼氶€夋嫨涓€涓礋杞借緝灏忕殑鑺傜偣锛屽悜璇ヨ妭鐐瑰彂閫佸鎵剧粍鍗忚皟鍣ㄧ殑璇锋眰
• KafkaApis 澶勭悊璇锋眰锛岃皟鐢ㄨ繑鍥炵粍鍗忚皟鍣ㄦ墍鍦ㄧ殑鑺傜偣锛岃繃绋嬪涓嬶細

• 鎵惧埌缁勫崗璋冨櫒鍚庯紝娑堣垂鑰呭崗璋冨櫒鐢宠鍔犲叆娑堣垂缁勶紝鍙戦€?JoinGroupRequest璇锋眰
• KafkaApis 璋冪敤 handleJoinGroup() 鏂规硶澶勭悊璇锋眰
  • 鎶婃秷璐硅€呮敞鍐屽埌娑堣垂缁勪腑
  • 鎶婃秷璐硅€呯殑clientId涓庝竴涓猆UID鍊肩敓鎴愪竴涓猰emberId鍒嗛厤缁欐秷璐硅€?/li>
  • 鏋勯€犲櫒璇ユ秷璐硅€呯殑MemberMetadata淇℃伅
  • 鎶婅娑堣垂鑰呯殑MemberMetadata淇℃伅娉ㄥ唽鍒癎roupMetadata涓?/li>
  • 绗竴涓姞鍏ョ粍鐨勬秷璐硅€呭皢鎴愪负leader
• 鎶婂鐞咼oinGroupRequest璇锋眰鐨勭粨鏋滆繑鍥炵粰娑堣垂鑰?/li>
• 鍔犲叆缁勬垚鍔熷悗锛岃繘琛屽垎鍖哄啀鍧囪　

以上是关于Kafka 鏍稿績缁勪欢涔嬪崗璋冨櫒的主要内容，如果未能解决你的问题，请参考以下文章

馃幆銆愭繁鍏ヨВ鏋愩€戣法绔鏋剁殑鏍稿績鎶€鏈埌搴曟槸浠€涔堬紵

Vue甯哥敤鐨刄I缁勪欢-vant锛堣交閲忋€佸彲闈犵殑绉诲姩绔疺ue缁勪欢搴擄級

Vue3 DOM Diff 鏍稿績绠楁硶瑙ｆ瀽

婊戝姩鍥剧墖鑷€傚簲image缁勪欢涓璦spectFill

涓篟eact Native寮€鍙戠殑杞挱灞曠ず鍔ㄧ敾缁勪欢

绠€鍗曡В鏋怱pring鏍稿績IOC瀹瑰櫒鍘熺悊