Tere Habr!
Töötan Tinkoffi meeskonnas, mis arendab oma teavituskeskust. Enamasti arendan Javas Spring booti kasutades ja lahendan erinevaid projekti käigus tekkivaid tehnilisi probleeme.
Enamik meie mikroteenuseid suhtlevad üksteisega asünkroonselt läbi sõnumimaakleri. Varem kasutasime maaklerina IBM MQ-d, mis ei tulnud enam koormusega toime, kuid samas oli kõrge tarnegarantiiga.
Asenduseks pakuti meile Apache Kafkat, millel on suur skaleerimispotentsiaal, kuid kahjuks nõuab see erinevate stsenaariumide jaoks peaaegu individuaalset lähenemist konfigureerimisele. Lisaks ei võimaldanud Kafkas vaikimisi töötav vähemalt ühekordse kohaletoimetamise mehhanism karbist väljas hoida vajalikku konsistentsi taset. Järgmisena jagan oma kogemusi Kafka konfigureerimisel, eelkõige räägin teile, kuidas konfigureerida ja elada täpselt ühe tarnekorraga.
Garanteeritud kohaletoimetamine ja palju muud
Allpool käsitletud sätted aitavad vältida mitmeid probleeme ühenduse vaikeseadetega. Kuid kõigepealt tahaksin pöörata tähelepanu ühele parameetrile, mis hõlbustab võimalikku silumist.
See aitab klient.id tootjale ja tarbijale. Esmapilgul saate väärtusena kasutada rakenduse nime ja enamikul juhtudel see töötab. Kuigi olukord, kui rakendus kasutab mitut tarbijat ja annate neile sama client.id, annab järgmise hoiatuse:
org.apache.kafka.common.utils.AppInfoParser — Error registering AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0Kui soovite kasutada JMX-i Kafkaga rakenduses, võib see olla probleem. Sel juhul on kõige parem kasutada kliendi.id väärtusena rakenduse nime ja näiteks teema nime kombinatsiooni. Meie konfiguratsiooni tulemust saab näha käsu väljundis kafka-tarbijarühmad Confluenti kommunaalteenustest:
Nüüd vaatame garanteeritud sõnumite edastamise stsenaariumi. Kafka Produceril on parameeter acks, mis võimaldab konfigureerida pärast seda, kui mitu kinnitust peab klastri juht sõnumi edukaks kirjutamiseks pidama. See parameeter võib võtta järgmisi väärtusi:
- 0 — kinnitust ei võeta arvesse.
- 1 on vaikeparameeter, kinnitamiseks on vaja ainult 1 koopiat.
- −1 — nõutav on kõigi sünkroonitud koopiate kinnitus (klastri häälestus min.insync.replicas).
Loetletud väärtuste põhjal on selge, et ackid, mis on võrdsed -1-ga, annavad tugevaima garantii, et sõnum ei lähe kaduma.
Nagu me kõik teame, on hajutatud süsteemid ebausaldusväärsed. Ajutiste rikete eest kaitsmiseks pakub Kafka Producer selle võimaluse proovib uuesti, mis võimaldab teil määrata kordussaatmiskatsete arvu tarne.timeout.ms. Kuna parameetri korduskatsete vaikeväärtus on Integer.MAX_VALUE (2147483647), saab sõnumite korduskatsete arvu reguleerida, muutes ainult delivery.timeout.ms.
Liigume täpselt ühekordse tarne poole
Loetletud seaded võimaldavad meie tootjal edastada sõnumeid kõrge garantiiga. Räägime nüüd sellest, kuidas tagada, et Kafka teemasse kirjutatakse ainult üks koopia sõnumist? Lihtsamal juhul peate selleks määrama parameetri Produceris võimaldama.idempotentsus tõeks. Idempotentsus garanteerib, et ühe teema konkreetsesse sektsiooni kirjutatakse ainult üks sõnum. Idempotentsuse võimaldamise eelduseks on väärtused acks = kõik, proovi uuesti > 0, max.in.flight.requests.per.connection ≤ 5. Kui arendaja ei ole neid parameetreid määranud, määratakse ülaltoodud väärtused automaatselt.
Kui idempotentsus on konfigureeritud, on vaja tagada, et samad sõnumid jõuaksid iga kord samadesse partitsioonidesse. Seda saab teha, määrates partitsiooni.class võtme ja parameetri väärtuseks Producer. Alustame võtmest. See peab olema iga esituse puhul sama. Seda saab hõlpsasti saavutada, kasutades mis tahes algse postituse ettevõtte ID-d. Parameetril partitioner.class on vaikeväärtus − . Selle partitsioonistrateegiaga toimime vaikimisi järgmiselt:
- Kui partitsioon on sõnumi saatmisel selgesõnaliselt määratud, siis kasutame seda.
- Kui partitsiooni pole määratud, kuid võti on määratud, valige partitsioon võtme räsi järgi.
- Kui partitsioon ja võti pole määratud, valige partitsioonid ükshaaval (ring-robin).
Samuti võtme kasutamine ja idempotentne saatmine parameetriga max.in.flight.requests.per.connection = 1 annab teile sujuva sõnumitöötluse tarbija jaoks. Samuti tasub meeles pidada, et kui teie klastris on konfigureeritud juurdepääsukontroll, on teil vaja õigusi, et teemasse idempotsionaalselt kirjutada.
Kui äkki jääb puudu idempotentse võtmega saatmise võimalustest või Produceri poolne loogika nõuab andmete järjepidevuse säilitamist erinevate sektsioonide vahel, siis tulevad appi tehingud. Lisaks saate aheltehingut kasutades tingimuslikult sünkroonida näiteks Kafka kirje andmebaasis oleva kirjega. Tootjale tehingute saatmise võimaldamiseks peab see olema idempotentne ja täiendavalt seadistatud tehinguline.id. Kui teie Kafka klastris on konfigureeritud juurdepääsukontroll, vajab tehingukirje, nagu ka idempotent kirje, kirjutamisõigusi, mille saab anda maskiga, kasutades faili tranzakcióal.id salvestatud väärtust.
Formaalselt saab tehingu identifikaatorina kasutada mis tahes stringi, näiteks rakenduse nime. Kuid kui käivitate sama rakenduse mitu eksemplari sama tehingual.id-ga, siis esimene käivitatud eksemplar peatatakse veaga, kuna Kafka peab seda zombiprotsessiks.
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.Selle probleemi lahendamiseks lisame rakenduse nimele järelliide hostinime kujul, mille saame keskkonnamuutujatest.
Tootja on konfigureeritud, kuid Kafka tehingud kontrollivad ainult sõnumi ulatust. Sõltumata tehingu olekust läheb teade kohe teemasse, kuid sellel on täiendavad süsteemiatribuudid.
Selleks, et Tarbija ei saaks selliseid sõnumeid enne tähtaega lugeda, peab ta parameetri määrama isolatsioon.tasand väärtusele read_committed value. Selline Tarbija saab lugeda tehinguväliseid sõnumeid nagu varem ja tehinguteateid alles pärast kohustust.
Kui olete kõik eelnevalt loetletud seaded määranud, olete konfigureerinud täpselt ühe tarnekorra. Palju õnne!
Kuid on veel üks nüanss. Transaction.id, mille me ülal seadistasime, on tegelikult tehingu eesliide. Tehinguhalduris lisatakse sellele järjenumber. Vastuvõetud tunnus väljastatakse tehinguline.id.aegumine.ms, mis on konfigureeritud Kafka klastris ja mille vaikeväärtus on "7 päeva". Kui selle aja jooksul ei ole rakendus ühtegi sõnumit saanud, siis järgmise tehingusaatmise proovimisel saate selle InvalidPidMappingException. Seejärel väljastab tehingu koordinaator järgmise tehingu jaoks uue järjekorranumbri. Sõnum võib aga kaduda, kui InvalidPidMappingExceptioni ei käsitleta õigesti.
Kogusummade asemel
Nagu näete, ei piisa ainult Kafkale sõnumite saatmisest. Peate valima parameetrite kombinatsiooni ja olema valmis kiireteks muudatusteks. Selles artiklis püüdsin üksikasjalikult näidata täpselt ühekordset kohaletoimetamise seadistust ja kirjeldasin mitmeid konfiguratsioonide client.id ja transactional.id probleeme, mis meil tekkisid. Allpool on kokkuvõte tootja ja tarbija seadetest.
Produtsent:
- acks = kõik
- korduskatsed > 0
- enable.idempotence = tõene
- max.in.flight.requests.per.connection ≤ 5 (1 korrapärase saatmise korral)
- transaktsiooniline.id = ${rakenduse-nimi}-${hostinimi}
Tarbija:
- isolation.level = read_committed
Vigade minimeerimiseks tulevastes rakendustes tegime vedrukonfiguratsioonile oma ümbrise, kus mõne loetletud parameetri väärtused on juba määratud.
Siin on paar materjali iseseisvaks õppimiseks:
Allikas: www.habr.com