Përshëndetje, Habr!
Unë punoj në ekipin Tinkoff, i cili merret me zhvillimin e qendrës sonë të njoftimeve. Kryesisht, unë zhvilloj në Java duke përdorur Spring Boot dhe zgjidh probleme të ndryshme teknike që lindin në projekt.
Shumica e mikroshërbimeve tona komunikojnë asinkronisht me njëra-tjetrën përmes një brokeri mesazhesh. Më parë, ne kemi përdorur IBM MQ si broker, i cili nuk arriti të përballonte ngarkesën, por kishte garanci të larta të dorëzimit.
Si zëvendësim, na u propozua Apache Kafka, i cili ka një potencial të lartë të shkallëzimit, por, fatkeqësisht, kërkon një qasje pothuajse të veçantë për konfigurimin për skenarë të ndryshëm. Për më tepër, mekanizmi i dorëzimit të paktën një herë, i cili punon në Kafka si parazgjedhje, nuk lejonte mbajtjen e një niveli të nevojshëm të konsistencës nga kutia. Më poshtë do të ndaj përvojën tonë të konfigurimit të Kafka, sidomos do të flas për si të konfiguroni dhe të jetoni me dorëzimin e saktë një herë.
Dorëzimi i garantuar dhe jo vetëm
Parametrat që do të shqyrtohen më tej do të ndihmojnë në parandalimin e disa problemeve me konfigurimet e parazgjedhura të lidhjes. Por fillimisht dua t'i kushtoj vëmendje një parametri, i cili do të lehtësojë mundësitë e debugging.
Këtë do ta ndihmojë client.id për Prodhuese dhe Konsumatore. Në pamje të parë, si vlerë mund të përdoret emri i aplikacionit, dhe në shumicën e rasteve kjo do të funksiononte. Megjithatë, situata kur një aplikacion përdor disa Konsumatorë dhe ju u jepni atyre të njëjtën client.id, çon në paralajmërimin e mëposhtëm:
org.apache.kafka.common.utils.AppInfoParser — Gabim në regjistrimin e AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0Nëse dëshironi të përdorni JMX në një aplikacion me Kafka, kjo mund të jetë një problem. Për këtë rast, është më mirë të përdorni si vlerë për client.id një kombinim të emrit të aplikacionit dhe, për shembull, emrit të temës. Rezultati i konfigurimit tonë mund të shihet në daljen e komandës kafka-consumer-groups nga utilitat e Confluent:
Tani le të shqyrtojmë skenarin e dorëzimit të garantuar të mesazhit. Një Kafka Producer ka një parametër acks, i cili lejon konfigurimin e numrit të acknowledge-ve që lideri i klasterit duhet të marrë për të konsideruar mesazhin si të shkruar me sukses. Ky parametër mund të marrë vlerat e mëposhtme:
- 0 — acknowledge-t nuk do të merren parasysh.
- 1 — parametri parazgjedhor, nevojitet acknowledge vetëm nga 1 replikë.
- −1 — nevojiten acknowledge nga të gjitha replikat e sinkronizuara (konfigurimi i klasterit min.insync.replicas).
Nga vlerat e lartpërmendura, është e qartë se acks i barabartë me −1 ofron garanci më të forta që mesazhi nuk do të humbasë.
Siç e dimë të gjithë, sistemet e shpërndara nuk janë të besueshme. Për të mbrojtur veten nga defekte përkohësore, Kafka Producer ofron parametrin retries, i cili lejon përcaktimin e numrit të përpjekjeve për dërgimin gjatë delivery.timeout.ms. Duke qenë se parametri retries ka një vlerë parazgjedhore Integer.MAX_VALUE (2147483647), numri i përsëritjeve të dërgimit të mesazhit mund të rregullohet duke ndryshuar vetëm delivery.timeout.ms.
Të kalojmë tek dorëzimi i saktë një herë
Parametrat e përmendur lejojnë Prodhuest tonë të dërgojë mesazhe me një garanci të lartë. Tani le të flasim për si të garantoni regjistrimin vetëm të një kopje të mesazhit në temën e Kafka? Në rastin më të thjeshtë, për këtë në Prodhuese duhet të vendosni parametrin enable.idempotence në vlerën true. Idempotenca garanton regjistrimin e vetëm një mesazhi në një pjesë të një teme të caktuar. Parakushte për aktivizimin e idempotencës janë vlerat acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Nëse këto parametra nuk janë caktuar nga zhvilluesi, atëherë automatikisht do të caktohen vlerat e përmendura më sipër.
Kur idempotenca është e konfiguruar, është e nevojshme të sigurohet që mesazhe të njëjta të shkojnë gjithmonë në të njëjtat pjesë. Kjo mund të arrihet duke konfiguruar çelësin dhe parametrin partitioner.class në Prodhuese. Le të fillojmë me çelësin. Për çdo dërgesë ai duhet të jetë i njëjtë. Këtë është e lehtë të arrihet duke përdorur një identifikues biznesi nga mesazhi origjinal. Parametri partitioner.class ka vlerën parazgjedhore — . Me këtë strategji të particionimit përfitojmë kështu:
- Nëse pjesa është e dhënë qartë gjatë dërgimit të mesazhit, atëherë e përdorim atë.
- Nëse pjesa nuk është e dhënë, por është i caktuar çelësi — zgjedhim pjesën në bazë të hashes nga çelësi.
- Nëse as pjesa as çelësi nuk janë të dhëna — zgjedhim pjesët në radhë (round-robin).
Për më tepër, përdorimi i çelësit dhe dërgimi idempotent me parametrin max.in.flight.requests.per.connection = 1 ju jep një përpunim të rregullt të mesazheve në Consumer. Është e rëndësishme të mbani mend se, nëse në klasterin tuaj është vendosur menaxhimi i qasjes, do t'ju nevojiten të drejtat për shkruaj the idempotent në temë.
Nëse ndodhi që ju mungojnë mundësitë e dërgimit idempotent sipas çelësit ose logjika në anën e Producer kërkon ruajtjen e qëndrueshmërisë së të dhënave midis particioneve të ndryshme, atëherë do të ndihmojnë transaksionet. Për më tepër, me ndihmën e transaksionit të lidhur, mund të sinkronizoni në mënyrë të kushtëzuar shkrimin në Kafka, për shembull, me shkrimin në bazën e të dhënave. Për të aktivizuar dërgimin transaksional në Producer, është e nevojshme që ai të ketë idempotencë dhe gjithashtu të vendosë transactional.id. Nëse klasteri juaj Kafka ka menaxhim qasje, atëherë për shkrimin transaksional, si dhe për atë idempotent, do t'ju nevojiten të drejta për shkruaj, të cilat mund të jepen sipas njëmaske duke përdorur vlerën që ruhet në transactional.id.
Formalizisht, si identifikues transaksioni mund të përdoret çdo varg, për shembull emri i aplikacionit. Por nëse po nisni disa instance të njëjtit aplikacion me të njëjtin transactional.id, atëherë instanca e parë e nisur do të ndalojë me një gabim, pasi Kafka do ta konsiderojë atë një proces zombik.
org.apache.kafka.common.errors.ProducerFencedException: Producer u përpoq të realizonte një operacion me një epokë të vjetër. Ose ka një producent më të ri me të njëjtin transactionalId, ose transaksioni i producentit ka skaduar nga brokeri.Për të zgjidhur këtë problem, ne shtojmë një sufix në emrin e aplikacionit në formën e emrit të hostit, të cilin e marrim nga variablat e mjedisit.
Producenti është i konfiguruar, por transaksionet në Kafka menaxhojnë vetëm fushën e dukshmërisë së mesazhit. Pavarësisht nga statusi i transaksionit, mesazhi shkon menjëherë në temë, por ka veti sistemore shtesë.
Që këto mesazhe të mos lexohen nga Consumer para kohe, ai duhet të vendosë parametrin isolation.level në vlerën read_committed. Ky Consumer do të mund të lexojë mesazhe jo transaksionale si më parë, dhe ato transaksionale vetëm pas komitetit.
Nëse keni vendosur të gjitha cilësimet e përmendura më parë, atëherë keni konfiguruar dërgimin exactly once. Urime!
Por ka edhe një hollësi tjetër. Transactional.id, të cilin e kemi konfiguruar më lart, në të vërtetë është një prefiks i transaksionit. Në menaxherin e transaksioneve i shtohet një numër rendor. Identifikuesi i marrë jepet në transactional.id.expiration.ms, i cili konfigurohet në klasterin Kafka dhe ka një vlerë parazgjedhje "7 ditë". Nëse gjatë kësaj kohe aplikacioni nuk ka marrë asnjë mesazh, atëherë gjatë përpjekjes për dërgimin transaksional të ardhshëm do të merrni InvalidPidMappingException. Pas kësaj, koordinatorja e transaksioneve do të japë një numër të ri rendor për transaksionin e ardhshëm. Në këtë rast, mesazhi mund të humbasë, nëse InvalidPidMappingException nuk trajtohet siç duhet.
Në vend të përfundimeve
Siç mund të vërehet, nuk mjafton thjesht të dërgoni mesazhe në Kafka. Duhet të zgjidhni kombinime parametrash dhe të jeni të gatshëm për bërë ndryshime të shpejta. Në këtë artikull, kam përpiqur të tregoj me detaje konfigurimin e dërgimit exactly once dhe përshkruaj disa probleme të konfigurimeve client.id dhe transactional.id, me të cilat u ballafaquam. Më poshtë në formë të shkurtër janë vendosjet e Produserit dhe Consumerit.
Producenti:
- acks = all
- retries > 0
- enable.idempotence = true
- max.in.flight.requests.per.connection ≤ 5 (1 — për dërgim të rregullt)
- transactional.id = ${application-name}-${hostname}
Consumatori:
- isolation.level = read_committed
Për të minimizuar gabimet në aplikacionet e ardhshme, ne krijuam një mbulesë mbi konfigurimin spring, ku janë vendosur vlerat për disa nga parametrat e përmendur.
Ja dhe disa materiale për studim të pavarur:
Burimi: habr.com
