Ինչպես Կաֆկան դարձավ իրականություն

Հե՜յ Հաբր։

Ես աշխատում եմ Tinkoff թիմի վրա, որը մշակում է իր սեփական ծանուցման կենտրոնը: Ես հիմնականում զարգանում եմ Java-ում` օգտագործելով Spring boot-ը և լուծում եմ տարբեր տեխնիկական խնդիրներ, որոնք առաջանում են նախագծում:

Մեր միկրոծառայությունների մեծ մասը միմյանց հետ շփվում է ասինխրոն կերպով՝ հաղորդագրության բրոքերի միջոցով: Նախկինում որպես բրոքեր օգտագործում էինք IBM MQ-ն, որն այլեւս չէր կարողանում հաղթահարել ծանրաբեռնվածությունը, բայց միևնույն ժամանակ ուներ առաքման բարձր երաշխիքներ։

Որպես փոխարինող, մեզ առաջարկեցին Apache Kafka-ն, որն ունի մեծ մասշտաբային ներուժ, բայց, ցավոք, պահանջում է գրեթե անհատական ​​մոտեցում տարբեր սցենարների համար կազմաձևման համար: Բացի այդ, առնվազն մեկ անգամ առաքման մեխանիզմը, որն աշխատում է Կաֆկայում լռելյայնորեն, թույլ չի տվել պահպանել համապատասխանության անհրաժեշտ մակարդակը տուփից դուրս: Հաջորդը, ես կկիսվեմ մեր փորձով Kafka-ի կոնֆիգուրացիայի վերաբերյալ, մասնավորապես, ես ձեզ կասեմ, թե ինչպես կարելի է կարգավորել և ապրել ուղիղ մեկ անգամ առաքմամբ:

Երաշխավորված առաքում և ավելին

Ստորև քննարկված կարգավորումները կօգնեն կանխել կանխադրված կապի կարգավորումների հետ կապված մի շարք խնդիրներ: Բայց նախ կցանկանայի ուշադրություն դարձնել մեկ պարամետրի վրա, որը կհեշտացնի հնարավոր կարգաբերումը.

Սա կօգնի client.id արտադրողի և սպառողի համար: Առաջին հայացքից դուք կարող եք օգտագործել հավելվածի անունը որպես արժեք, և շատ դեպքերում դա կաշխատի: Թեև այն իրավիճակը, երբ հավելվածն օգտագործում է մի քանի սպառող, և դուք նրանց տալիս եք նույն client.id-ը, հանգեցնում է հետևյալ նախազգուշացմանը.

org.apache.kafka.common.utils.AppInfoParser — Error registering AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0

Եթե ​​ցանկանում եք օգտագործել JMX-ը Կաֆկայի հետ հավելվածում, ապա դա կարող է խնդիր լինել: Այս դեպքում լավագույնն է օգտագործել հավելվածի անվան և, օրինակ, թեմայի անվան համակցությունը որպես client.id արժեք: Մեր կազմաձևման արդյունքը կարելի է տեսնել հրամանի ելքում կաֆկա-սպառողական-խմբեր կոմունալ ծառայություններից Confluent-ից.

Հիմա եկեք նայենք հաղորդագրությունների երաշխավորված առաքման սցենարին: Kafka Producer-ն ունի պարամետր ակսեր, որը թույլ է տալիս կարգավորել, թե քանի ընդունումից հետո կլաստերի ղեկավարին անհրաժեշտ է հաղորդագրությունը հաջողությամբ գրված համարելու համար: Այս պարամետրը կարող է վերցնել հետևյալ արժեքները.

• 0 — ընդունումը չի դիտարկվի:
• 1-ը լռելյայն պարամետրն է, դրա հաստատման համար պահանջվում է ընդամենը 1 կրկնօրինակ:
• −1 — բոլոր համաժամացված կրկնօրինակներից պահանջվում է ճանաչում (կլաստերի կարգավորում min.insync.replicas).

Թվարկված արժեքներից պարզ է դառնում, որ −1-ի հավասարությունը տալիս է ամենաուժեղ երաշխիքը, որ հաղորդագրությունը չի կորչի:

Ինչպես բոլորս գիտենք, բաշխված համակարգերն անվստահելի են: Անցումային անսարքություններից պաշտպանվելու համար Kafka Producer-ը տրամադրում է տարբերակը կրկնում է, որը թույլ է տալիս սահմանել ներսում կրկին ուղարկելու փորձերի քանակը delivery.timeout.ms. Քանի որ կրկնվող պարամետրն ունի Integer.MAX_VALUE (2147483647) լռելյայն արժեքը, հաղորդագրությունների կրկնությունների քանակը կարող է ճշգրտվել՝ փոխելով միայն delivery.timeout.ms:

Մենք գնում ենք դեպի ուղիղ մեկ առաքում

Թվարկված կարգավորումները թույլ են տալիս մեր Արտադրողին հաղորդագրություններ ուղարկել բարձր երաշխիքով: Եկեք հիմա խոսենք այն մասին, թե ինչպես ապահովել, որ հաղորդագրության միայն մեկ օրինակ գրվի Կաֆկայի թեմայում: Ամենապարզ դեպքում դա անելու համար պետք է պարամետրը սահմանել Producer-ի վրա միացնել.idempotence ճշմարիտ. Idempotency-ն երաշխավորում է, որ միայն մեկ հաղորդագրություն է գրված մեկ թեմայի կոնկրետ բաժնի վրա: Դեմպոտենցիայի հնարավորություն տալու նախապայմանը արժեքներն են acks = բոլորը, նորից փորձել > 0, max.in.flight.requests.per.connection ≤ 5. Եթե ​​մշակողի կողմից նշված պարամետրերը նշված չեն, վերը նշված արժեքները ավտոմատ կերպով կսահմանվեն:

Երբ idempotency-ն կարգավորվում է, անհրաժեշտ է ապահովել, որ ամեն անգամ նույն հաղորդագրությունները հայտնվում են նույն բաժանմունքներում: Դա կարելի է անել՝ partitioner.class ստեղնը և պարամետրը Producer-ի վրա դնելով: Սկսենք բանալիից: Այն պետք է լինի նույնը յուրաքանչյուր ներկայացման համար: Դրան կարելի է հեշտությամբ հասնել՝ օգտագործելով սկզբնական գրառման բիզնեսի ID-ներից որևէ մեկը: Partitioner.class պարամետրն ունի լռելյայն արժեք − Default Partitioner. Այս բաժանման ռազմավարությամբ մենք լռելյայն գործում ենք այսպես.

• Եթե ​​հաղորդագրությունն ուղարկելիս բաժանումը հստակորեն նշված է, ապա մենք օգտագործում ենք այն:
• Եթե ​​բաժանումը նշված չէ, բայց ստեղնը նշված է, ընտրեք բաժանումը ստեղնի հեշով:
• Եթե ​​միջնորմը և բանալին նշված չեն, ապա մեկ առ մեկ ընտրեք միջնորմները (շրջանաձեւ):

Նաև՝ օգտագործելով բանալի և պարամետրով անիմաստ ուղարկում max.in.flight.requests.per.connection = 1 Ձեզ տրամադրում է հաղորդագրությունների պարզեցված մշակում Սպառողի վրա: Հարկ է նաև հիշել, որ եթե մուտքի հսկողությունը կազմաձևված է ձեր կլաստերի վրա, ապա ձեզ անհրաժեշտ կլինեն իրավունքներ՝ թեմային անզուսպ գրելու համար:

Եթե ​​հանկարծ ձեզ պակասեն բանալիով անիմաստ ուղարկելու հնարավորությունները կամ Արտադրողի կողմից տրամաբանությունը պահանջում է պահպանել տվյալների համապատասխանությունը տարբեր բաժինների միջև, ապա գործարքները կգան օգնության: Բացի այդ, օգտագործելով շղթայական գործարքը, դուք կարող եք պայմանականորեն համաժամեցնել Կաֆկայի գրառումը, օրինակ, տվյալների բազայի գրառումների հետ: Գործարքներով Արտադրողին ուղարկելը հնարավոր դարձնելու համար այն պետք է լինի անիմաստ և լրացուցիչ սահմանված գործարքային.id. Եթե ​​ձեր Kafka կլաստերին մուտքի կառավարումը կազմաձևված է, ապա գործարքային գրառումը, ինչպես անիմաստ գրառումը, գրելու թույլտվությունների կարիք կունենա, որոնք կարող են տրամադրվել դիմակի միջոցով՝ օգտագործելով transanctional.id-ում պահվող արժեքը:

Ֆորմալ կերպով, ցանկացած տող, ինչպիսին է հավելվածի անունը, կարող է օգտագործվել որպես գործարքի նույնացուցիչ: Բայց եթե գործարկեք միևնույն հավելվածի մի քանի օրինակ նույն transaksional.id-ով, ապա առաջին գործարկված օրինակը կդադարեցվի սխալմամբ, քանի որ Կաֆկան դա կհամարի զոմբիացման գործընթաց:

org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker.

Այս խնդիրը լուծելու համար հավելվածի անվան վրա մենք ավելացնում ենք վերջածանց՝ հոսթանունի տեսքով, որը ստանում ենք շրջակա միջավայրի փոփոխականներից։

Արտադրողը կազմաձևված է, բայց Կաֆկայի վրա կատարված գործարքները վերահսկում են միայն հաղորդագրության շրջանակը: Անկախ գործարքի կարգավիճակից, հաղորդագրությունն անմիջապես անցնում է թեմա, բայց ունի լրացուցիչ համակարգի ատրիբուտներ:

Նման հաղորդագրությունները Սպառողի կողմից ժամանակից շուտ կարդալուց խուսափելու համար անհրաժեշտ է սահմանել պարամետրը մեկուսացում.մակարդակ read_committed արժեքը: Նման Սպառողը կկարողանա կարդալ ոչ գործարքային հաղորդագրությունները նախկինի պես, իսկ գործարքային հաղորդագրությունները միայն հանձնառությունից հետո:
Եթե ​​դուք սահմանել եք ավելի վաղ թվարկված բոլոր կարգավորումները, ապա դուք կարգավորել եք առաքումը հենց մեկ անգամ: Շնորհավորում եմ:

Բայց կա ևս մեկ նրբերանգ. Transactional.id-ը, որը մենք կազմաձևեցինք վերևում, իրականում գործարքի նախածանցն է: Գործարքների կառավարչի վրա դրան ավելացվում է հերթական համարը: Ստացված նույնացուցիչը տրվում է transaksional.id.expiration.ms, որը կազմաձևված է Kafka կլաստերի վրա և ունի «7 օր» լռելյայն արժեք: Եթե ​​այս ընթացքում հավելվածը որևէ հաղորդագրություն չի ստացել, ապա երբ փորձեք հաջորդ գործարքային ուղարկումը, դուք կստանաք InvalidPidMappingException. Գործարքի համակարգողն այնուհետև հաջորդ գործարքի համար կտրամադրի նոր հաջորդական համար: Այնուամենայնիվ, հաղորդագրությունը կարող է կորչել, եթե InvalidPidMappingException-ը ճիշտ չմշակվի:

Ընդհանուր դրույթների փոխարեն

Ինչպես տեսնում եք, Կաֆկային պարզապես հաղորդագրություններ ուղարկելը բավարար չէ։ Դուք պետք է ընտրեք պարամետրերի համադրություն և պատրաստ լինեք արագ փոփոխություններ կատարելու: Այս հոդվածում ես փորձեցի մանրամասն ցույց տալ հենց մեկ անգամ առաքման կարգավորումը և նկարագրեցի մի շարք խնդիրներ՝ կապված client.id և transaksional.id կոնֆիգուրացիաների հետ, որոնք մենք հանդիպեցինք: Ստորև բերված է Արտադրողի և Սպառողի կարգավորումների ամփոփագիրը:

Պրոդյուսեր

1. ակս = բոլորը
2. կրկնել > 0
3. enable.idempotence = ճշմարիտ
4. max.in.flight.requests.per.connection ≤ 5 (1 կանոնավոր ուղարկման համար)
5. transactional.id = ${application-name}-${hostname}

Սպառող

1. isolation.level = read_committed

Ապագա հավելվածներում սխալները նվազագույնի հասցնելու համար մենք պատրաստեցինք մեր սեփական փաթաթան գարնանային կոնֆիգուրացիայի վրա, որտեղ թվարկված որոշ պարամետրերի արժեքներն արդեն սահմանված են:

Ահա մի քանի նյութեր ինքնուրույն ուսումնասիրության համար.

• KIP-98 - Հենց մեկ անգամ առաքում և գործարքային հաղորդագրություններ
• Պարամետրերի նկարագրությունը

Source: www.habr.com