绐ユ帰婧愮爜锛岃鎴戞洿鍔犱紭闆呯殑浣跨敤Kafka鐢熶骇鑰咃紒

Posted 2021-05-02 鐭虫潐鐨勬灦鏋勭瑪璁?/a>

 鍛ㄤ竴鑷冲懆浜旀棭8鐐瑰崐锛佺簿鍝佹妧鏈枃绔犲噯鏃堕€佷笂锛?/span>

鍓嶈█

鍦ㄤ笂鏂?涓彁鍒版秷鎭祦杞噰鐢ㄧ殑鏄?Kafka 浣滀负涓棿浠躲€?/p>

鍏朵腑鏈夋湅鍙嬪挩璇㈠湪澶ч噺娑堟伅鐨勬儏鍐典笅 Kakfa 鏄浣曚繚璇佹秷鎭殑楂樻晥鍙婁竴鑷存€у憿锛?/p>

姝ｅソ浠ヨ繖涓棶棰樼粨鍚?Kakfa 鐨勬簮鐮佽璁轰笅濡備綍姝ｇ‘銆侀珮鏁堢殑鍙戦€佹秷鎭€?/p>

鍐呭杈冨锛屽婧愮爜鎰熷叴瓒ｇ殑鏈嬪弸璇风郴濂藉畨鍏ㄥ甫馃槒(婧愮爜鍩轰簬 v0.10.0.0 鐗堟湰鍒嗘瀽)銆傚悓鏃舵渶濂芥槸鏈変竴瀹氱殑 Kafka 浣跨敤缁忛獙锛岀煡鏅撳熀鏈殑鐢ㄦ硶銆?/p>

绠€鍗曠殑娑堟伅鍙戦€?/h1>

鍦ㄥ垎鏋愪箣鍓嶅厛鐪嬩竴涓畝鍗曠殑娑堟伅鍙戦€佹槸鎬庝箞鏍风殑銆?/p>

浠ヤ笅浠ｇ爜鍩轰簬 SpringBoot 鏋勫缓銆?/p>

棣栧厛鍒涘缓涓€涓?org.apache.kafka.clients.producer.Producer 鐨?bean銆?/p>

鍏朵綑鍑犱釜鍙傛暟鏆傛椂涓嶅仛璁ㄨ锛屽悗鏂囦細鏈夎缁嗕粙缁嶃€?/p>

鎺ョ潃娉ㄥ叆杩欎釜 bean 鍗冲彲璋冪敤瀹冪殑鍙戦€佸嚱鏁板彂閫佹秷鎭€?/p>

杩欓噷鎴戠粰鏌愪竴涓?Topic 鍙戦€佷簡 10W 鏉℃暟鎹紝杩愯绋嬪簭娑堟伅姝ｅ父鍙戦€併€?/p>

浣嗚繖浠呬粎鍙槸鍋氬埌浜嗘秷鎭彂閫侊紝瀵规秷鎭槸鍚︽垚鍔熼€佽揪瀹屽叏娌＄锛岀瓑浜庢槸绾?寮傛鐨勬柟寮忋€?/p>

鍚屾

閭ｄ箞鎴戞兂鐭ラ亾娑堟伅鍒板簳鍙戦€佹垚鍔熸病鏈夎鎬庝箞鍔炲憿锛?/p>

鍏跺疄 Producer 鐨?API 宸茬粡甯垜浠€冭檻鍒颁簡锛屽彂閫佷箣鍚庡彧闇€瑕佽皟鐢ㄥ畠鐨?get() 鏂规硶鍗冲彲鍚屾鑾峰彇鍙戦€佺粨鏋溿€?/p>

鍙戦€佺粨鏋滐細

杩欐牱鐨勫彂閫佹晥鐜囧叾瀹炴槸姣旇緝浣庝笅鐨勶紝鍥犱负姣忔閮介渶瑕佸悓姝ョ瓑寰呮秷鎭彂閫佺殑缁撴灉銆?/p>

寮傛

涓烘鎴戜滑搴斿綋閲囧彇寮傛鐨勬柟寮忓彂閫侊紝鍏跺疄 send() 鏂规硶榛樿鍒欐槸寮傛鐨勶紝鍙涓嶆墜鍔ㄨ皟鐢?get() 鏂规硶銆?/p>

浣嗚繖鏍峰氨娌℃硶鑾风煡鍙戦€佺粨鏋溿€?/p>

鎵€浠ユ煡鐪?send() 鐨?API 鍙互鍙戠幇杩樻湁涓€涓弬鏁般€?/p>

     
       
       
     

      
        
        
      
Future<RecordMetadata> send(ProducerRecord<K, V> producer, Callback callback);
     
       
       
     

Callback 鏄竴涓洖璋冩帴鍙ｏ紝鍦ㄦ秷鎭彂閫佸畬鎴愪箣鍚庡彲浠ュ洖璋冩垜浠嚜瀹氫箟鐨勫疄鐜般€?/p>

鎵ц涔嬪悗鐨勭粨鏋滐細

鍚屾牱鐨勪篃鑳借幏鍙栫粨鏋滐紝鍚屾椂鍙戠幇鍥炶皟鐨勭嚎绋嬪苟涓嶆槸涓婃枃鍚屾鏃剁殑 涓荤嚎绋?/span>锛岃繖鏍蜂篃鑳借瘉鏄庢槸寮傛鍥炶皟鐨勩€?/p>

鍚屾椂鍥炶皟鐨勬椂鍊欎細浼犻€掍袱涓弬鏁帮細

• RecordMetadata 鍜屼笂鏂囦竴鑷寸殑娑堟伅鍙戦€佹垚鍔熷悗鐨勫厓鏁版嵁銆?/span>

• Exception 娑堟伅鍙戦€佽繃绋嬩腑鐨勫紓甯镐俊鎭€?/span>

浣嗘槸杩欎袱涓弬鏁板苟涓嶄細鍚屾椂閮芥湁鏁版嵁锛屽彧鏈夊彂閫佸け璐ユ墠浼氭湁寮傚父淇℃伅锛屽悓鏃跺彂閫佸厓鏁版嵁涓虹┖銆?/p>

鎵€浠ユ纭殑鍐欐硶搴斿綋鏄細

鑷充簬涓轰粈涔堜細鍙湁鍙傛暟涓€涓湁鍊硷紝鍦ㄤ笅鏂囩殑婧愮爜鍒嗘瀽涓細涓€涓€瑙ｉ噴銆?/p>

婧愮爜鍒嗘瀽

鐜板湪鍙帉鎻′簡鍩烘湰鐨勬秷鎭彂閫侊紝鎯宠娣卞埢鐨勭悊瑙ｅ彂閫佷腑鐨勪竴浜涘弬鏁伴厤缃繕鏄緱婧愮爜璇翠簡绠椼€?/p>

棣栧厛杩樻槸鏉ヨ皥璋堟秷鎭彂閫佹椂鐨勬暣涓祦绋嬫槸鎬庝箞鏍风殑锛?Kafka 骞朵笉鏄畝鍗曠殑鎶婃秷鎭€氳繃缃戠粶鍙戦€佸埌浜?broker涓紝鍦?Java 鍐呴儴杩樻槸缁忚繃浜嗚澶氫紭鍖栧拰璁捐銆?/p>

鍙戦€佹祦绋?/h2>

涓轰簡鐩磋鐨勪簡瑙ｅ彂閫佺殑娴佺▼锛岀畝鍗曠殑鐢讳簡鍑犱釜鍦ㄥ彂閫佽繃绋嬩腑鍏抽敭鐨勬楠ゃ€?/p>

浠庝笂鑷充笅渚濇鏄細

• 鍒濆鍖栦互鍙婄湡姝ｅ彂閫佹秷鎭殑 kafka-producer-network-thread IO 绾跨▼銆?/span>

• 灏嗘秷鎭簭鍒楀寲銆?/span>

• 寰楀埌闇€瑕佸彂閫佺殑鍒嗗尯銆?/span>

• 鍐欏叆鍐呴儴鐨勪竴涓紦瀛樺尯涓€?/span>

• 鍒濆鍖栫殑 IO 绾跨▼涓嶆柇鐨勬秷璐硅繖涓紦瀛樻潵鍙戦€佹秷鎭€?/span>

姝ラ瑙ｆ瀽

鎺ヤ笅鏉ヨ瑙ｆ瘡涓楠ゃ€?/p>

鍒濆鍖?/h3>

璋冪敤璇ユ瀯閫犳柟娉曡繘琛屽垵濮嬪寲鏃讹紝涓嶆鏄畝鍗曠殑灏嗗熀鏈弬鏁板啓鍏?KafkaProducer銆傛瘮杈冮夯鐑︾殑鏄垵濮嬪寲 Sender 绾跨▼杩涜缂撳啿鍖烘秷璐广€?/p>

鍒濆鍖?IO 绾跨▼澶勶細

鍙互鐪嬪埌 Sender 绾跨▼鏈夐渶瑕佹垚鍛樺彉閲忥紝姣斿锛?/p>

     
       
       
     

      
        
        
      
acks,retries,requestTimeout
     
       
       
     

绛夛紝杩欎簺鍙傛暟浼氬湪鍚庢枃鍒嗘瀽銆?/p>

搴忓垪鍖栨秷鎭?/h3>

鍦ㄨ皟鐢?send() 鍑芥暟鍚庡叾瀹炵涓€姝ュ氨鏄簭鍒楀寲锛屾瘯绔熸垜浠殑娑堟伅闇€瑕侀€氳繃缃戠粶鎵嶈兘鍙戦€佸埌 Kafka銆?/p>

鍏朵腑鐨?valueSerializer.serialize(record.topic(),record.value()); 鏄竴涓帴鍙ｏ紝鎴戜滑闇€瑕佸湪鍒濆鍖栨椂鍊欐寚瀹氬簭鍒楀寲瀹炵幇绫汇€?/p>

鎴戜滑涔熷彲浠ヨ嚜宸卞疄鐜板簭鍒楀寲锛屽彧闇€瑕佸疄鐜?org.apache.kafka.common.serialization.Serializer 鎺ュ彛鍗冲彲銆?/p>

璺敱鍒嗗尯

鎺ヤ笅鏉ュ氨鏄矾鐢卞垎鍖猴紝閫氬父鎴戜滑浣跨敤鐨?Topic 涓轰簡瀹炵幇鎵╁睍鎬т互鍙婇珮鎬ц兘閮戒細鍒涘缓澶氫釜鍒嗗尯銆?/p>

濡傛灉鏄竴涓垎鍖哄ソ璇达紝鎵€鏈夋秷鎭兘寰€閲岄潰鍐欏叆鍗冲彲銆?/p>

浣嗗涓垎鍖哄氨涓嶅彲閬垮厤闇€瑕佺煡閬撳啓鍏ュ摢涓垎鍖恒€?/p>

閫氬父鏈変笁绉嶆柟寮忋€?/p>

鎸囧畾鍒嗗尯

鍙互鍦ㄦ瀯寤?ProducerRecord 涓烘瘡鏉℃秷鎭寚瀹氬垎鍖恒€?/p>

杩欐牱鍦ㄨ矾鐢辨椂浼氬垽鏂槸鍚︽湁鎸囧畾锛屾湁灏辩洿鎺ヤ娇鐢ㄨ鍒嗗尯銆?/p>

杩欑涓€鑸湪鐗规畩鍦烘櫙涓嬩細浣跨敤銆?/p>

鑷畾涔夎矾鐢辩瓥鐣?/h4>

濡傛灉娌℃湁鎸囧畾鍒嗗尯锛屽垯浼氳皟鐢?partitioner.partition 鎺ュ彛鎵ц鑷畾涔夊垎鍖虹瓥鐣ャ€?/p>

鑰屾垜浠篃鍙渶瑕佽嚜瀹氫箟涓€涓被瀹炵幇 org.apache.kafka.clients.producer.Partitioner鎺ュ彛锛屽悓鏃跺湪鍒涘缓 KafkaProducer 瀹炰緥鏃堕厤缃?partitioner.class 鍙傛暟銆?/p>

閫氬父闇€瑕佽嚜瀹氫箟鍒嗗尯涓€鑸槸鍦ㄦ兂灏介噺鐨勪繚璇佹秷鎭殑椤哄簭鎬с€?/p>

鎴栬€呮槸鍐欏叆鏌愪簺鐗规湁鐨勫垎鍖猴紝鐢辩壒鍒殑娑堣垂鑰呮潵杩涜澶勭悊绛夈€?/p>

榛樿绛栫暐

鏈€鍚庝竴绉嶅垯鏄粯璁ょ殑璺敱绛栫暐锛屽鏋滄垜浠暐閮芥病鍋氬氨浼氭墽琛岃绛栫暐銆?/p>

璇ョ瓥鐣ヤ篃浼氫娇寰楁秷鎭垎閰嶇殑姣旇緝鍧囧寑銆?/p>

鏉ョ湅鐪嬪畠鐨勫疄鐜帮細

绠€鍗曠殑鏉ヨ鍒嗕负浠ヤ笅鍑犳锛?/p>

• 鑾峰彇 Topic 鍒嗗尯鏁般€?/span>

• 灏嗗唴閮ㄧ淮鎶ょ殑涓€涓嚎绋嬪畨鍏ㄨ鏁板櫒 +1銆?/span>

• 涓庡垎鍖烘暟鍙栨ā寰楀埌鍒嗗尯缂栧彿銆?/span>

鍏跺疄杩欏氨鏄緢鍏稿瀷鐨勮疆璇㈢畻娉曪紝鎵€浠ュ彧瑕佸垎鍖烘暟涓嶉绻佸彉鍔ㄨ繖绉嶆柟寮忎篃浼氭瘮杈冨潎鍖€銆?/p>

鍐欏叆鍐呴儴缂撳瓨

鍦?send() 鏂规硶鎷垮埌鍒嗗尯鍚庝細璋冪敤涓€涓?append() 鍑芥暟锛?/p>

璇ュ嚱鏁颁腑浼氳皟鐢ㄤ竴涓?getOrCreateDeque() 鍐欏叆鍒颁竴涓唴閮ㄧ紦瀛樹腑 batches銆?/p>

娑堣垂缂撳瓨

鍦ㄦ渶寮€濮嬪垵濮嬪寲鐨?IO 绾跨▼鍏跺疄鏄竴涓畧鎶ょ嚎绋嬶紝瀹冧細涓€鐩存秷璐硅繖浜涙暟鎹€?/p>

閫氳繃鍥句腑鐨勫嚑涓嚱鏁颁細鑾峰彇鍒颁箣鍓嶅啓鍏ョ殑鏁版嵁銆傝繖鍧楀唴瀹瑰彲浠ヤ笉蹇呮繁绌讹紝浣嗗叾涓湁涓?completeBatch 鏂规硶鍗撮潪甯稿叧閿€?/p>

璋冪敤璇ユ柟娉曟椂鍊欒偗瀹氬凡缁忔槸娑堟伅鍙戦€佸畬姣曚簡锛屾墍浠ヤ細璋冪敤 batch.done() 鏉ュ畬鎴愪箣鍓嶆垜浠湪 send() 鏂规硶涓畾涔夌殑鍥炶皟鎺ュ彛銆?/p>

浠庤繖閲屼篃鍙互鐪嬪嚭涓轰粈涔堜箣鍓嶈鍙戦€佸畬鎴愬悗鍏冩暟鎹拰寮傚父淇℃伅鍙細鍑虹幇涓€涓€?/p>

Producer 鍙傛暟瑙ｆ瀽

鍙戦€佹祦绋嬭瀹屼簡鍐嶆潵鐪嬬湅 Producer 涓瘮杈冮噸瑕佺殑鍑犱釜鍙傛暟銆?/p>

acks

acks 鏄竴涓奖鍝嶆秷鎭悶鍚愰噺鐨勪竴涓叧閿弬鏁般€?/p>

涓昏鏈?[all銆?1,0,1] 杩欏嚑涓€夐」锛岄粯璁や负 1銆?/p>

鐢变簬 Kafka 涓嶆槸閲囧彇鐨勪富澶囨ā寮忥紝鑰屾槸閲囩敤绫讳技浜?Zookeeper 鐨勪富澶囨ā寮忋€?/p>

鍓嶆彁鏄?Topic 閰嶇疆鍓湰鏁伴噺 replica>1銆?/p>

褰?acks=all/-1 鏃讹細

鎰忓懗鐫€浼氱‘淇濇墍鏈夌殑 follower 鍓湰閮藉畬鎴愭暟鎹殑鍐欏叆鎵嶄細杩斿洖銆?/p>

杩欐牱鍙互淇濊瘉娑堟伅涓嶄細涓㈠け锛?/p>

浣嗗悓鏃舵€ц兘鍜屽悶鍚愰噺鍗存槸鏈€浣庣殑銆?/p>

褰?acks=0 鏃讹細

producer 涓嶄細绛夊緟鍓湰鐨勪换浣曞搷搴旓紝杩欐牱鏈€瀹规槗涓㈠け娑堟伅浣嗗悓鏃舵€ц兘鍗存槸鏈€濂界殑锛?/p>

褰?acks=1 鏃讹細

杩欐槸涓€绉嶆姌涓殑鏂规锛屽畠浼氱瓑寰呭壇鏈?Leader 鍝嶅簲锛屼絾涓嶄細绛夊埌 follower 鐨勫搷搴斻€?/p>

涓€鏃?Leader 鎸傛帀娑堟伅灏变細涓㈠け銆備絾鎬ц兘鍜屾秷鎭畨鍏ㄦ€ч兘寰楀埌浜嗕竴瀹氱殑淇濊瘉銆?/p>

batch.size

杩欎釜鍙傛暟鐪嬪悕绉板氨鐭ラ亾鏄唴閮ㄧ紦瀛樺尯鐨勫ぇ灏忛檺鍒讹紝瀵逛粬閫傚綋鐨勮皟澶у彲浠ユ彁楂樺悶鍚愰噺銆?/p>

浣嗕篃涓嶈兘鏋佺锛岃皟澶ぇ浼氭氮璐瑰唴瀛樸€傚皬浜嗕篃鍙戞尌涓嶄簡浣滅敤锛屼篃鏄竴涓吀鍨嬬殑鏃堕棿鍜岀┖闂寸殑鏉冭　銆?/p>

涓婂浘鏄嚑涓娇鐢ㄧ殑浣撶幇銆?/p>

retries

retries 璇ュ弬鏁颁富瑕佹槸鏉ュ仛閲嶈瘯浣跨敤锛屽綋鍙戠敓涓€浜涚綉缁滄姈鍔ㄩ兘浼氶€犳垚閲嶈瘯銆?/p>

杩欎釜鍙傛暟涔熷氨鏄檺鍒堕噸璇曟鏁般€?/p>

浣嗕篃鏈変竴浜涘叾浠栭棶棰樸€?/p>

• 鍥犱负鏄噸鍙戞墍浠ユ秷鎭『搴忓彲鑳戒笉浼氫竴鑷达紝杩欎篃鏄笂鏂囨彁鍒板氨绠楁槸涓€涓垎鍖烘秷鎭篃涓嶄細鏄畬鍏ㄩ『搴忕殑鎯呭喌銆?/span>

• 杩樻槸鐢变簬缃戠粶闂锛屾湰鏉ユ秷鎭凡缁忔垚鍔熷啓鍏ヤ簡浣嗘槸娌℃湁鎴愬姛鍝嶅簲缁?producer锛岃繘琛岄噸璇曟椂灏卞彲鑳戒細鍑虹幇 娑堟伅閲嶅銆傝繖绉嶅彧鑳芥槸娑堣垂鑰呰繘琛屽箓绛夊鐞嗐€?/span>

楂樻晥鐨勫彂閫佹柟寮?/h1>

濡傛灉娑堟伅閲忕湡鐨勯潪甯稿ぇ锛屽悓鏃跺張闇€瑕佸敖蹇殑灏嗘秷鎭彂閫佸埌 Kafka銆備竴涓?producer 濮嬬粓浼氭敹鍒扮紦瀛樺ぇ灏忕瓑褰卞搷銆?/p>

閭ｆ槸鍚﹀彲浠ュ垱寤哄涓?producer 鏉ヨ繘琛屽彂閫佸憿锛?/p>

• 閰嶇疆涓€涓渶澶?producer 涓暟銆?/span>

• 鍙戦€佹秷鎭椂棣栧厛鑾峰彇涓€涓?nbsp;producer锛岃幏鍙栫殑鍚屾椂鍒ゆ柇鏄惁杈惧埌鏈€澶т笂闄愶紝娌℃湁灏辨柊寤轰竴涓悓鏃朵繚瀛樺埌鍐呴儴鐨?nbsp;List涓紝淇濆瓨鏃跺仛濂藉悓姝ュ鐞嗛槻姝㈠苟鍙戦棶棰樸€?/span>

• 鑾峰彇鍙戦€佽€呮椂鍙互鎸夌収榛樿鐨勫垎鍖虹瓥鐣ヤ娇鐢ㄨ疆璇㈢殑鏂瑰紡鑾峰彇锛堜繚璇佷娇鐢ㄥ潎鍖€锛夈€?/span>

杩欐牱鍦ㄥぇ閲忋€侀绻佺殑娑堟伅鍙戦€佸満鏅腑鍙互鎻愰珮鍙戦€佹晥鐜囧噺杞诲崟涓?producer 鐨勫帇鍔涖€?/p>

鍏抽棴 Producer

鏈€鍚庡垯鏄?Producer 鐨勫叧闂紝Producer 鍦ㄤ娇鐢ㄨ繃绋嬩腑娑堣€椾簡涓嶅皯璧勬簮锛堢嚎绋嬨€佸唴瀛樸€佺綉缁滅瓑锛夊洜姝ら渶瑕佹樉寮忕殑鍏抽棴浠庤€屽洖鏀惰繖浜涜祫婧愩€?/p>

榛樿鐨?close() 鏂规硶鍜屽甫鏈夎秴鏃舵椂闂寸殑鏂规硶閮芥槸鍦ㄤ竴瀹氱殑鏃堕棿鍚庡己鍒跺叧闂€?/p>

浣嗗湪杩囨湡涔嬪墠閮戒細澶勭悊瀹屽墿浣欑殑浠诲姟銆?/p>

鎵€浠ヤ娇鐢ㄥ摢涓€涓緱瑙嗘儏鍐佃€屽畾銆?/p>

鎬荤粨

鏈枃鍐呭杈冨锛屼粠瀹炰緥鍜屾簮鐮佺殑瑙掑害鍒嗘瀽浜?Kafka 鐢熶骇鑰呫€?/p>

甯屾湜鐪嬪畬鐨勬湅鍙嬭兘鏈夋敹鑾凤紝鍚屾椂涔熸杩庣暀瑷€璁ㄨ銆?/p>

涓嶅嚭鎰忓涓嬫湡浼氳璁?Kafka 娑堣垂鑰呫€?/p>

杩樻病璇诲锛?span class="mq-472">鏇村鍘熷垱绯诲垪鏂囩珷锛岃绉绘鑷?/span>

•  

•                                     

鎸佺画鍏虫敞锛屼竴澶ф尝鍘熷垱绯诲垪鏂囩珷姝ｅ湪璺笂

娆㈣繋鎵弿涓嬫柟浜岀淮鐮?/strong>

鐭虫潐鐨勬灦鏋勭瑪璁帮紙id:shishan100锛?/span>

BAT鏋舵瀯缁忛獙鍊惧泭鐩告巿