🥇Разуменне брокераў паведамленняў. Вывучэнне механікі абмену паведамленнямі з дапамогай ActiveMQ і Kafka. Кіраўнік 3. Kafka

Працяг перакладу невялікай кнігі:
"Understanding Message Brokers",
аўтар: Jakub Korab, выдавецтва: O'Reilly Media, Inc., дата выдання: June 2017, ISBN: 9781492049296.

Папярэдняя перакладзеная частка: Разуменне брокераў паведамленняў. Вывучэнне механікі абмену паведамленнямі з дапамогай ActiveMQ і Kafka. Кіраўнік 1. Увядзенне

ГЛАВА 3

Кафка

Kafka была распрацавана ў LinkedIn для таго, каб абыйсці некаторыя абмежаванні традыцыйных брокераў паведамленняў і пазбегнуць неабходнасці наладжваць некалькі брокераў паведамленняў для розных узаемадзеянняў "кропка-кропка", што апісана ў дадзенай кнізе ў раздзеле "Вертыкальнае і гарызантальнае маштабаванне" на старонцы 28. Сцэнары выкарыстання у LinkedIn у асноўным засноўваліся на аднанакіраваным паглынанні вельмі вялікіх аб'ёмаў дадзеных, такіх як зграі на старонках і часопісы доступу, у той жа час дазваляючы выкарыстаць гэтыя дадзеныя некалькім сістэмам, не ўплываючы на прадукцыйнасць прадзюсараў ці іншых кансюмераў. Фактычна, чыннік існавання Kafka складаецца ў тым, каб атрымаць такую архітэктуру абмену паведамленнямі, якую апісвае Universal Data Pipeline.

З улікам гэтай канчатковай мэты, натуральна, узніклі і іншыя патрабаванні. Kafka павінна:

Быць надзвычай хуткай
Даваць вялікую прапускную здольнасць пры працы з паведамленнямі
Падтрымліваць мадэлі «Выдавец-падпісчык» і «Кропка-Кропка»
Не запавольвацца з дабаўленнем спажыўцоў. Напрыклад, прадукцыйнасць і чэргі, і топіка ў ActiveMQ пагаршаецца пры росце колькасці спажыўцоў на адрасаце
Быць гарызантальна якая маштабуецца; калі адзін брокер, які захоўвае (persists) паведамленні, можа рабіць гэта толькі на максімальнай хуткасці дыска, то для павелічэння прадукцыйнасці мае сэнс выйсці за межы аднаго асобніка брокера.
Размяжоўваць доступ да захоўвання і паўторнага вымання паведамленняў

Каб дасягнуць усяго гэтага, у Kafka прынята архітэктура, якая перавызначыла ролі і абавязкі кліентаў і брокераў абмену паведамленнямі. Мадэль JMS вельмі арыентавана на брокер, дзе ён адказвае за распаўсюджванне паведамленняў, а кліенты павінны турбавацца толькі аб адпраўцы і атрыманні паведамленняў. Kafka, з іншага боку, арыентавана на кліента, пры гэтым кліент бярэ на сябе многія функцыі традыцыйнага брокера, такія як справядлівае размеркаванне адпаведных паведамленняў сярод спажыўцоў, у абмен атрымліваючы надзвычай хуткі і які маштабуецца брокер. Для людзей, якія працавалі з традыцыйнымі сістэмамі абмену паведамленнямі, праца з Kafka патрабуе фундаментальных змен у поглядах.
Гэты інжынерны кірунак прывяло да стварэння інфраструктуры абмену паведамленнямі, здольнай на шмат парадкаў павялічыць прапускную здольнасць у параўнанні са звычайным брокерам. Як мы ўбачым, гэты падыход спалучаны з кампрамісамі, якія азначаюць, што Kafka не падыходзіць для вызначаных тыпаў нагрузак і ўсталяванага ПА.

Уніфікаваная мадэль адрасата

Каб выканаць патрабаванні, апісаныя вышэй, Kafka аб'яднала абмен паведамленнямі тыпу "публікацыя-падпіска" і "кропка-кропка" у рамках аднаго віду адрасата. топіка. Гэта збівае з панталыку людзей, якія працавалі з сістэмамі абмену паведамленнямі, дзе слова "топік" ставіцца да шырокавяшчальнага механізму, з якога (з топіка) чытанне не з'яўляецца надзейным (is nondurable). Топікі Kafka варта разглядаць як гібрыдны тып адрасата, у адпаведнасці з вызначэннем, дадзеным ва ўводзінах да гэтай кнігі.

У пакінутай частцы гэтага раздзела, калі мы відавочна не пакажам іншае, тэрмін "топік" будзе ставіцца да топіка Kafka.

Каб цалкам зразумець, як паводзяць сябе топікі і якія гарантыі яны падаюць, нам трэба спачатку разгледзець, як яны рэалізаваны ў Kafka.
У кожнага топіка ў Kafka ёсць свой часопіс.
Прадзюсары, якія адпраўляюць паведамленні ў Kafka, дапісваюць у гэты часопіс, а кансюмеры чытаюць з часопіса з дапамогай паказальнікаў, якія ўвесь час перамяшчаюцца наперад. Перыядычна Kafka выдаляе самыя старыя часткі часопіса, незалежна ад таго, ці былі паведамленні ў гэтых частках прачытаныя ці не. Цэнтральнай часткай дызайну Kafka з'яўляецца тое, што брокер не клапоціцца аб тым, прачытаныя ці паведамленні ці не - гэта адказнасць кліента.

Тэрміны «часопіс» і «паказальнік» не сустракаюцца ў дакументацыі Kafka. Гэтыя добра вядомыя тэрміны выкарыстоўваюцца тут, каб дапамагчы разуменню.

Гэтая мадэль цалкам адрозніваецца ад ActiveMQ, дзе паведамленні з усіх чэргаў захоўваюцца ў адным часопісе, а брокер пазначае паведамленні, як выдаленыя, пасля таго як яны былі прачытаныя.
Давайце зараз крыху паглыбімся і разгледзім часопіс топіка больш падрабязна.
Часопіс Kafka складаецца з некалькіх партыцый (Малюнак 3-1). Kafka гарантуе строгую спарадкаванасць у кожнай партіціі. Гэта азначае, што паведамленні, запісаныя ў партыцыю ў вызначаным парадку, будуць прачытаныя ў тым жа парадку. Кожная партыцыя рэалізавана ў выглядзе цыклічнага (rolling) файла часопіса, які змяшчае падмноства (subset) усіх паведамленняў, адпраўленых у топік яго прадзюсарамі. Створаны топік змяшчае па-змаўчанні адну партыцыю. Ідэя партыцый - гэта цэнтральная ідэя Kafka для гарызантальнага маштабавання.

Figure 3-1. Партыцыі Kafka

Калі прадзюсер адпраўляе паведамленне ў топік Kafka, ён вырашае, у якую партыцыю адправіць паведамленне. Мы разгледзім гэта больш падрабязна пазней.

Чытанне паведамленняў

Кліент, які хоча прачытаць паведамленні, кіруе найменным паказальнікам, званым група кансюмераў (consumer group), які паказвае на зрушэнне (offset) паведамлення ў партыцыі. Зрушэнне - гэта пазіцыя з нарастальным нумарам, якая пачынаецца з 0 у пачатку партыцыі. Гэтая група кансюмераў, на якую спасылаюцца ў API праз вызначаны карыстачом ідэнтыфікатар group_id, адпавядае аднаму лагічнаму спажыўцу або сістэме.

Большасць сістэм, выкарыстоўвалых абмен паведамленнямі, чытаюць дадзеныя з адрасата пасродкам некалькіх асобнікаў і струменяў для раўналежнай апрацоўкі паведамленняў. Такім чынам, звычайна будзе шмат асобнікаў кансюмераў, сумесна выкарыстоўвалых адну і тую ж групу кансюмераў.

Праблему чытання можна ўявіць наступным чынам:

Топік мае некалькі партыцый
Выкарыстоўваць топік можа адначасова мноства груп кансюмераў.
Група кансюмераў можа мець некалькі асобных экзэмпляраў.

Гэта нетрывіяльная праблема "многія да многіх". Каб зразумець, як Kafka звяртаецца з адносінамі паміж групамі кансюмераў, асобнікамі кансюмераў і партыямі, разгледзім шэраг паступова якія ўскладняюцца сцэнараў чытання.

Кансюмеры і групы кансюмераў

Давайце возьмем у якасці адпраўной кропкі топік з адной партыцыяй (Малюнак 3-2).

Figure 3/2. Кансюмер чытае з партыцыі

Калі асобнік кансюмера падлучаецца са сваім уласным group_id да гэтага топіка, яму прызначаецца партыцыя для чытання і зрушэнне ў гэтай партіціі. Палажэнне гэтага зрушэння канфігуруецца ў кліенце, як паказальнік на самую апошнюю пазіцыю (самае новае паведамленне) або самую раннюю пазіцыю (самае старое паведамленне). Кансюмер запытвае (polls) паведамленні з топіка, што прыводзіць да іх паслядоўнага чытання з часопіса.
Пазіцыя зрушэння рэгулярна коміціцца назад у Kafka і захоўваецца, як паведамленні ва ўнутраным топіку _consumer_offsets. Прачытаныя паведамленні ўсё роўна не выдаляюцца, у адрозненне ад звычайнага брокера, і кліент можа пераматаць (rewind) зрушэнне, каб паўторна апрацаваць ужо прагледжаныя паведамленні.

Калі падключаецца другі лагічны кансюмер, выкарыстоўваючы іншы group_id, ён кіруе другім паказальнікам, які не залежыць ад першага (Малюнак 3-3). Такім чынам, топік Kafka дзейнічае як чарга, у якой існуе адзін кансюмер і, як звычайны топік выдавец-падпісчык (pub-sub), на які падпісаны некалькі кансюмераў, з дадатковай перавагай, што ўсе паведамленні захоўваюцца і могуць апрацоўвацца некалькі разоў.

Figure 3-3. Два кансюмеры ў розных групах кансюмераў чытаюць з адной партыцыі

Кансюмеры ў групе кансюмераў

Калі адзін асобнік кансюмера чытае дадзеныя з партіціі, ён цалкам кантралюе паказальнік і апрацоўвае паведамленні, як апісана ў папярэдняй частцы.
Калі некалькі асобнікаў кансюмераў былі падлучаныя з адным і тым жа group_id да топіка з адной партыцыяй, то асобніку, які падлучыўся апошнім, будзе перададзены кантроль над паказальнікам і з гэтага моманту ён будзе атрымліваць усе паведамленні (Малюнак 3-4).

Figure 3-4. Два кансюмеры ў адной і той жа групе кансюмераў чытаюць з адной партыцыі

Гэты рэжым апрацоўкі, у якім колькасць асобнікаў кансюмераў перавышае лік партыцый, можна разглядаць як разнавіднасць манапольнага спажыўца. Гэта можа быць карысна, калі вам патрэбна "актыўна-пасіўная" (або "гарачая-цёплая") кластарызацыя вашых асобнікаў кансюмераў, хоць паралельная праца некалькіх кансюмераў ("актыўна-актыўная" ці "гарачая-гарачая") нашмат больш тыповая, чым кансюмеры у рэжыме чакання.

Такія паводзіны размеркавання паведамленняў, апісанае вышэй, могуць выклікаць здзіўленне ў параўнанні з тым, як паводзіць сябе звычайная чарга JMS. У гэтай мадэлі паведамлення, адпраўленыя ў чаргу, будуць раўнамерна размеркаваны паміж двума кансюмерамі.

Часцей за ўсё, калі мы ствараем некалькі асобнікаў кансюмераў, мы робім гэта або для паралельнай апрацоўкі паведамленняў, або для павелічэння хуткасці чытання, або для павышэння ўстойлівасці працэсу чытання. Паколькі чытаць дадзеныя з партіціі можа адначасова толькі адзін асобнік кансюмера, то як гэта дасягаецца ў Kafka?

Адзін са спосабаў зрабіць гэта - выкарыстоўваць адзін асобнік кансюмера, каб прачытаць усе паведамленні і перадаць іх у пул патокаў. Хоць гэты падыход павялічвае прапускную здольнасць апрацоўкі, ён павялічвае складанасць логікі кансюмераў і нічога не робіць для павышэння ўстойлівасці сістэмы чытання. Калі адзін асобнік кансюмера адключаецца з-за збою сілкавання ці аналагічнай падзеі, то вычытка спыняецца.

Кананічным спосабам вырашэння гэтай праблемы ў Kafka з'яўляецца выкарыстанне бОбольшай колькасці партыцый.

Партыцыянаванне

Партыцыі з'яўляюцца асноўным механізмам распаралельвання чытання і маштабаванні топіка за межы прапускной здольнасці аднаго асобніка брокера. Каб лепш зразумець гэта, давайце разгледзім сітуацыю, калі існуе топік з двума партыцыямі і на гэты топік падпісваецца адзін кансюмер (Малюнак 3-5).

Figure 3-5. Адзін кансюмер чытае з некалькіх партыцый

У гэтым сцэнары кансюмеру даецца кантроль над паказальнікамі, якія адпавядаюць яго group_id у абодвух партыцыях, і пачынаецца чытанне паведамленняў з абедзвюх партыцый.
Калі ў гэты топік дадаецца дадатковы кансюмер для таго ж group_id, Kafka пераназначае (reallocate) адну з партыцый з першага на другі кансюмер. Пасля чаго кожны асобнік кансюмера будзе вычытваць з адной партіціі топіка (Малюнак 3-6).

Каб забяспечыць апрацоўку паведамленняў паралельна ў 20 патокаў, вам спатрэбіцца як мінімум 20 партыцый. Калі партыцый будзе менш, у вас застануцца кансюмеры, якім няма над чым працаваць, што апісана раней у абмеркаванні манапольных кансюмераў.

Figure 3-6. Два кансюмеры ў адной і той жа групе кансюмераў чытаюць з розных партыцый

Гэтая схема значна змяншае складанасць працы брокера Kafka у параўнанні з размеркаваннем паведамленняў, неабходным для падтрымкі чаргі JMS. Тут не трэба клапоціцца аб наступных момантах:

Які кансюмер павінен атрымаць наступнае паведамленне, засноўваючыся на цыклічным (round-robin) размеркаванні, бягучай ёмістасці буфераў папярэдняй выбаркі ці папярэдніх паведамленнях (як для груп паведамленняў JMS).
Якія паведамленні адпраўлены якім кансюмерам і ці павінны яны быць дастаўлены паўторна ў выпадку збою.

Усё, што павінен зрабіць брокер Kafka - гэта паслядоўна перадаваць паведамленні кансюмеру, калі апошні запытвае іх.

Аднак, патрабаванні да распаралельвання вычыткі і паўторнай адпраўцы няўдалых паведамленняў нікуды не дзяюцца – адказнасць за іх проста пераходзіць ад брокера да кліента. Гэта азначае, што яны павінны быць улічаныя ў вашым кодзе.

Адпраўка паведамленняў

Адказнасць за рашэнне, у якую партыцыю адправіць паведамленне, ускладаецца на прадзюсер гэтага паведамлення. Каб зразумець механізм, з дапамогай якога гэта робіцца, спачатку трэба разгледзець, што менавіта мы насамрэч адпраўляем.

У той час, як у JMS мы выкарыстоўваем структуру паведамлення з метададзенымі (загалоўкамі і ўласцівасцямі) і целам, утрымоўвальным карысную нагрузку (payload), у Kafka паведамленне з'яўляецца парай «ключ-значэнне». Карысная нагрузка паведамлення адпраўляецца, як значэнне (value). Ключ, з іншага боку, выкарыстоўваецца галоўным чынам для партыцыянавання і павінен змяшчаць спецыфічны для бізнес-логікі ключ, Каб змясціць звязаныя паведамленняў у тую ж партыцыю.

У Раздзеле 2 мы абмяркоўвалі сцэнар анлайн-ставак, калі злучаныя падзеі павінны апрацоўвацца па парадку адным кансюмерам:

Уліковы запіс карыстальніка настроена.
Грошы залічваюцца на рахунак.
Робіцца стаўка, якая выводзіць грошы з рахунку.

Калі кожная падзея ўяўляе сабой паведамленне, адпраўленае ў топік, то ў гэтым выпадку натуральным ключом будзе ідэнтыфікатар уліковага запісу.
Калі паведамленне адпраўляецца з выкарыстаннем Kafka Producer API, яно перадаецца функцыі партыцыянавання, якая, улічваючы паведамленне і бягучы стан кластара Kafka, вяртае ідэнтыфікатар партыцыі, у якую павінна быць даслана паведамленне. Гэтая функцыя рэалізаваная ў Java праз інтэрфейс Partitioner.

Гэты інтэрфейс выглядае наступным чынам:

interface Partitioner {
    int partition(String topic,
        Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
}

Рэалізацыя Partitioner для вызначэння партіціі выкарыстоўвае па змаўчанні алгарытм хэшавання ключа (general-purpose hashing algorithm over the key) або цыклічны перабор (round-robin), калі ключ не паказаны. Гэта значэнне па-змаўчанні працуе добра ў большасці выпадкаў. Аднак, у будучыні вы захочаце напісаць свой уласны.

Напісанне ўласнай стратэгіі партыцыянавання

Давайце разгледзім прыклад, калі вы хочаце адправіць метададзеныя разам з карыснай нагрузкай паведамлення. Карысная нагрузка ў нашым прыкладзе - гэта інструкцыя для ўнясення дэпазіту на гульнявы рахунак. Інструкцыя - гэта тое, што мы хацелі б гарантавана не мадыфікаваць пры перадачы і хочам быць упэўненыя, што толькі давераная вышэйстаячая сістэма можа ініцыяваць гэтую інструкцыю. У гэтым выпадку адпраўляючая і прымаючая сістэмы ўзгадняюць выкарыстанне подпісу для праверкі сапраўднасці паведамлення.
У звычайным JMS мы проста вызначаем уласцівасць "подпіс паведамлення" і дадаем яго да паведамлення. Тым не менш, Kafka не дае нам механізм для перадачы метададзеных - толькі ключ і значэнне.

Паколькі значэнне - гэта карысная нагрузка банкаўскага перакладу (bank transfer payload), цэласнасць якой мы хочам захаваць, у нас не застаецца іншага выбару, акрамя вызначэння структуры дадзеных для выкарыстання ў ключы. Мяркуючы, што нам патрэбен ідэнтыфікатар уліковага запісу для партыцыянавання, бо ўсе паведамленні, якія адносяцца да ўліковага запісу, павінны апрацоўвацца па парадку, мы прыдумаем наступную структуру JSON:

{
  "signature": "541661622185851c248b41bf0cea7ad0",
  "accountId": "10007865234"
}

Паколькі значэнне подпісу будзе вар'іравацца ў залежнасці ад карыснай нагрузкі, дэфолтная стратэгія хэшавання інтэрфейсу Partitioner не будзе надзейна групаваць звязаныя паведамленні. Таму нам трэба будзе напісаць сваю ўласную стратэгію, якая будзе аналізаваць гэты ключ і падзяляць (partition) значэнне accountId.

Kafka уключае кантрольныя сумы для выяўлення пашкоджанні паведамленняў у сховішча і мае поўны набор функцый бяспекі. Нават у гэтым выпадку часам з'яўляюцца спецыфічныя для галіны патрабаванні, такія як прыведзенае вышэй.

Карыстацкая стратэгія партыцыянавання павінна гарантаваць, што ўсе злучаныя паведамленні апынуцца ў адной партыцыі. Хоць гэта здаецца простым, але патрабаванне можа быць ускладнена з-за важнасці парадкавання звязаных паведамленняў і таго, наколькі фіксавана колькасць партыцый у топіцы.

Колькасць партыцый у топіцы можа змяняцца з часам, бо іх можна дадаць, калі трафік выходзіць за межы першапачатковых чаканняў. Такім чынам, ключы паведамленняў могуць быць злучаны з партыцыяй, у якую яны былі першапачаткова адпраўленыя, маючы на ўвазе частка стану, якое павінна быць размеркавана паміж асобнікамі прадзюсара.

Іншым фактарам, які трэба ўлічваць, з'яўляецца раўнамернасць размеркавання паведамленняў паміж партыямі. Як правіла, ключы не размяркоўваюцца раўнамерна па паведамленнях, і хэш-функцыі не гарантуюць справядлівае размеркаванне паведамленняў для невялікага набору ключоў.
Важна адзначыць, што, як бы вы ні вырашылі падзяліць паведамленні, сам падзельнік, магчыма, давядзецца выкарыстоўваць паўторна.

Разгледзім патрабаванне рэплікацыі дадзеных паміж кластарамі Kafka у розных геаграфічных размяшчэннях. Для гэтай мэты Kafka пастаўляецца з прыладай каманднага радка пад назовам MirrorMaker, які выкарыстоўваецца для чытання паведамленняў з аднаго кластара і перадачы іх у іншы.

MirrorMaker павінен разумець ключы які рэплікуецца топіка, каб падтрымліваць адносны парадак паміж паведамленнямі пры рэплікацыі паміж кластарамі, паколькі колькасць партыцый для гэтага топіка можа не супадаць у двух кластарах.

Карыстальніцкія стратэгіі партыцыянавання сустракаюцца адносна рэдка, бо дэфолтныя хэшаванне ці цыклічны перабор паспяхова працуюць у большасці сцэнараў. Аднак, калі вам патрабуюцца строгія гарантыі парадкавання ці вам неабходна атрымаць метададзеныя з карысных нагрузак, то партыцыянаванне - гэта тое, на што вам варта зірнуць больш падрабязна.

Перавагі маштабаванасці і прадукцыйнасці Kafka абумоўлены пераносам некаторых абавязкаў традыцыйнага брокера на кліента. У гэтым выпадку прымаецца рашэнне аб размеркаванні патэнцыйна злучаных паведамленняў па некалькіх кансюмерам, якія працуюць раўналежна.

JMS брокеры таксама павінны мець справу з такімі патрабаваннямі. Цікава, што механізм адпраўкі звязаных паведамленняў аднаму і таму ж кансюмеру, рэалізаваны праз JMS Message Groups (разнавіднасць стратэгіі балансавання sticky load balancing (SLB)), таксама патрабуе, каб адпраўнік пазначаў паведамленні, як звязаныя. У выпадку JMS, брокер адказвае за адпраўку гэтай групы звязаных паведамленняў аднаму кансюмеру з многіх і перадачу правоў уласнасці на групу калі кансюмер адваліўся.

Пагадненні па прадзюсары

Партыцыянаванне - гэта не адзінае, што неабходна ўлічваць пры адпраўцы паведамленняў. Давайце разгледзім метады send () класа Producer у Java API:

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

Варта адразу адзначыць, што абодва метаду вяртаюць Future, што паказвае на тое, што аперацыя адпраўкі не выконваецца неадкладна. У выніку атрымліваецца, што паведамленне (ProducerRecord) запісваецца ў буфер адпраўкі для кожнай актыўнай партіціі і перадаецца брокеру фонавым струменем у бібліятэцы кліента Kafka. Хоць гэта робіць працу неверагодна хуткай, гэта азначае, што нявопытна напісанае прыкладанне можа страціць паведамленні, калі яго працэс будзе спынены.

Як заўсёды, ёсць спосаб зрабіць аперацыю адпраўкі больш надзейнай за рахунак прадукцыйнасці. Памер гэтага буфера можна ўсталяваць у 0, і струмень які адпраўляе прыкладання будзе змушаны чакаць, пакуль перадача паведамлення брокеру не будзе завершана, наступным чынам:

RecordMetadata metadata = producer.send(record).get();

Яшчэ раз аб чытанні паведамленняў

Чытанне паведамленняў мае дадатковыя складанасці, аб якіх неабходна паразважаць. У адрозненне ад API JMS, які можа запускаць слухача паведамленняў (message listener) у адказ на паступленне паведамлення, інтэрфейс Спажывец Kafka толькі апытвае (polling). Давайце падрабязней разгледзім метад poll (), які выкарыстоўваецца для гэтай мэты:

ConsumerRecords < K, V > poll(long timeout);

Вяртаецца значэнне метаду - гэта кантэйнерная структура, якая змяшчае некалькі аб'ектаў ConsumerRecord з патэнцыйна некалькіх партыцый. ConsumerRecord сам па сабе з'яўляецца аб'ектам-холдэрам для пары ключ-значэнне з адпаведнымі метададзенымі, такімі, як партыцыя, з якой ён атрыманы.

Як абмяркоўвалася ў Главе 2, мы павінны ўвесь час памятаць, што адбываецца з паведамленнямі пасля іх паспяховай ці няўдалай апрацоўкі, напрыклад, калі кліент не можа апрацаваць паведамленне ці калі ён перарывае працу. У JMS гэта апрацоўвалася праз рэжым пацверджання (acknowledgement mode). Брокер альбо выдаліць паспяховае апрацаванае паведамленне, альбо паўторна даставіць неапрацаванае ці зафейленае (пры ўмове, што былі выкарыстаныя транзакцыі).
Kafka працуе зусім па-іншаму. Паведамленні не выдаляюцца ў брокеры пасля вычыткі і адказнасць за тое, што адбываецца пры збоі, ляжыць на самым вычытвальным кодзе.

Як мы ўжо казалі, група кансюмераў звязана са зрушэннем у часопісе. Пазіцыя ў часопісе, звязаная з гэтым зрушэннем, адпавядае наступнаму паведамленню, якое будзе выдадзена ў адказ на poll (). Вырашальнае значэнне пры чытанні мае момант часу, калі гэтае зрушэнне павялічваецца.

Вяртаючыся да мадэлі чытання, разгледжанай раней, апрацоўка паведамлення складаецца з трох этапаў:

Выняць паведамленне для чытання.
Апрацаваць паведамленне.
Пацвердзіць паведамленне.

Кансюмер Kafka пастаўляецца з опцыяй канфігурацыі enable.auto.commit. Гэта часта выкарыстоўваная налада па змаўчанні, як гэта звычайна бывае з наладамі, якія змяшчаюць слова "аўто".

Да Kafka 0.10 кліент, які выкарыстоўваў гэты параметр, адпраўляў зрушэнне апошняга прачытанага паведамлення пры наступным выкліку poll () пасля апрацоўкі. Гэта азначала, што любыя паведамленні, якія ўжо былі вынятыя (fetched), маглі быць паўторна апрацаваны, калі кліент іх ужо апрацаваў, але быў нечакана знішчаны перад выклікам poll (). Паколькі брокер не захоўвае ніякага стану адносна таго, колькі разоў паведамленне было прачытана, наступны кансюмер, які здабывае гэтае паведамленне, не будзе ведаць, што адбылося нешта дрэннае. Гэтыя паводзіны былі псеўда-транзакцыйнымі. Зрушэнне каміцілася толькі ў выпадку паспяховай апрацоўкі паведамлення, але калі кліент перапыняў працу, брокер зноў адпраўляў тое ж самае паведамленне іншаму кліенту. Такія паводзіны адпавядалі гарантыі дастаўкі паведамленняўпа меншай меры адзін раз«.

У Kafka 0.10 код кліента быў зменены такім чынам, што коміт стаў перыядычна запускацца бібліятэкай кліента, у адпаведнасці з настройкай. auto.commit.interval.ms. Гэтыя паводзіны знаходзяцца недзе паміж рэжымамі JMS AUTO_ACKNOWLEDGE і DUPS_OK_ACKNOWLEDGE. Пры выкарыстанні аўтакаміта паведамлення маглі быць пацверджаны незалежна ад таго, ці былі яны фактычна апрацаваны - гэта магло адбыцца ў выпадку павольнага кансюмера. Калі кансюмер перапыняў працу, паведамленні здабываліся наступным кансюмерам, пачынальна з закаммічанай пазіцыі, што магло прывесці да пропуску паведамлення. У гэтым выпадку Kafka не губляла паведамленні, які чытае код проста не апрацоўваў іх.

Гэты рэжым мае тыя ж далягляды, што і ў версіі 0.9: паведамленні могуць быць апрацаваны, але ў выпадку збою, зрушэнне можа быць не закаммічана, што патэнцыйна можа прывесці да задваення дастаўкі. Чым больш паведамленняў вы здабываеце пры выкананні poll (), тым больш гэтая праблема.

Як абмяркоўвалася ў раздзеле «Чытанне паведамленняў з чаргі» на стар. 21, у сістэме абмену паведамленнямі няма такога паняцця, як аднаразовая дастаўка паведамлення, калі прыняць да ўвагі рэжымы збояў.

У Kafka ёсць два спосабу зафіксаваць (закаміціць) зрушэнне (афсет): аўтаматычна і ўручную. У абодвух выпадках паведамленні могуць апрацоўвацца некалькі разоў, калі паведамленне было апрацавана, але адбыўся збой да комміта. Вы таксама можаце наогул не апрацоўваць паведамленне, калі комiт адбыўся ў фоне і ваш код быў завершаны да таго, як ён прыступіў да апрацоўкі (магчыма ў Kafka 0.9 і больш ранніх версіях).

Кіраваць працэсам комита зрушэння ўручную можна ў API кансюмера Kafka, усталяваўшы параметр enable.auto.commit у значэнне false і відавочна выклікаўшы адзін з наступных метадаў:

void commitSync();
void commitAsync();

Калі вы імкнецеся апрацаваць паведамленне "хоць бы адзін раз", вы павінны закамаміціць зрушэнне ўручную з дапамогай commitSync (), выканаўшы гэтую каманду адразу пасля апрацоўкі паведамленняў.

Гэтыя метады не дазваляюць пацвярджаць (acknowledged) паведамленні да таго, як яны будуць апрацаваны, але яны нічога не робяць для ўхілення патэнцыйнага задваення апрацоўкі, у той жа час ствараючы бачнасць транзакцыйнасці. У Kafka адсутнічаюць транзакцыі. У кліента няма магчымасці зрабіць наступнае:

Аўтаматычна адкаціць (roll back) зафейленае паведамленне. Кансюмеры самі павінны апрацоўваць выключэнні, якія ўзнікаюць з-за праблемных пэйлаадаў і адключэнняў бэкэнда, бо яны не могуць спадзявацца на паўторную дастаўку паведамленняў брокерам.
Адправіць паведамленні ў некалькі топікаў у рамках адной атамарнай аперацыі. Як мы хутка ўбачым, кантроль над рознымі топікі і партыцыямі можа знаходзіцца на розных машынах у кластары Kafka, якія не каардынуюць транзакцыі пры адпраўцы. На момант напісання гэтага артыкула была праведзена пэўная праца, каб зрабіць гэта магчымым з дапамогай KIP-98.
Звязаць чытанне аднаго паведамлення з аднаго топіка з адпраўкай іншага паведамлення ў іншы топік. Ізноў жа, архітэктура Kafka залежыць ад мноства незалежных машын, якія працуюць як адна шына і не робіцца ніякіх спроб схаваць гэта. Напрыклад, не існуе кампанентаў API, якія дазволілі б звязаць Кансюмер и прадзюсер у транзакцыі. У JMS гэта забяспечваецца аб'ектам сесія, з якога ствараюцца MessageProducers и MessageConsumers.

Калі мы не можам спадзявацца на транзакцыі, як мы можам забяспечыць семантыку, бліжэйшую да той, якую падаюць традыцыйныя сістэмы абмену паведамленнямі?

Калі існуе верагоднасць таго, што зрушэнне кансюмера можа павялічыцца да таго, як паведамленне было апрацавана, напрыклад, падчас збою кансюмера, то ў кансюмера няма спосабу пазнаць, ці прапусціла яго група кансюмераў паведамлення, калі ёй прызначаюць партыцыю. Такім чынам, адна са стратэгій складаецца ў перамотцы (rewind) зрушэнні на папярэднюю пазіцыю. API кансюмера Kafka падае наступныя метады для гэтага:

void seek(TopicPartition partition, long offset);
void seekToBeginning(Collection < TopicPartition > partitions);

метад seek () можа выкарыстоўвацца з метадам
offsetsForTimes (Map timestampsToSearch) для перамоткі ў стан у які-небудзь пэўны момант у мінулым.

Няяўна, выкарыстанне гэтага падыходу азначае, што, вельмі верагодна, некаторыя паведамленні, якія былі апрацаваны раней, будуць прачытаныя і апрацаваны нанова. Каб пазбегнуць гэтага, мы можам выкарыстоўваць ідэмпатэнтнае чытанне, як апісана ў Раздзеле 4, для адсочвання раней прагледжаных паведамленняў і выключэнні дублікатаў.

Як альтэрнатыва, код вашага кансюмера можа быць простым, калі дапушчальная страта або дубліраванне паведамленняў. Калі мы разглядаем сцэнары выкарыстання, для якіх звычайна выкарыстоўваецца Kafka, напрыклад, апрацоўка падзей логаў, метрык, адсочванне клікаў і г.д., мы разумеем, што страта асобных паведамленняў наўрад ці акажа значны ўплыў на навакольныя прыкладанні. У такіх выпадках значэнні па-змаўчанні суцэль дапушчальныя. З іншага боку, калі вашаму з дадаткам трэба перадаваць плацяжы, вы павінны старанна клапаціцца аб кожным асобным паведамленні. Усё зводзіцца да кантэксту.

Асабістыя назіранні паказваюць, што з ростам інтэнсіўнасці паведамленняў, каштоўнасць кожнага асобнага паведамлення змяншаецца. Паведамленні вялікага аб'ёму становяцца, як правіла, каштоўнымі, калі іх разглядаць у агрэгаванай форме.

Высокая даступнасць (High Availability)

Падыход Kafka у стаўленні высокай даступнасці істотна адрозніваецца ад падыходу ActiveMQ. Kafka распрацавана на базе гарызантальна якія маштабуюцца кластараў, у якіх усе асобнікі брокера прымаюць і раздаюць паведамленні адначасова.

Кластар Kafka складаецца з некалькіх асобнікаў брокера, якія працуюць на розных серверах. Kafka была распрацавана для працы на звычайным аўтаномным жалезе, дзе кожны вузел мае сваё ўласнае наадварот сховішча. Выкарыстанне сеткавых сховішчаў (SAN) не рэкамендуецца, паколькі множныя вылічальныя вузлы могуць канкураваць за часЫе інтэрвалы сховішчы і ствараць канфлікты.

Kafka - гэта стала ўключаная сістэма. Многія буйныя карыстальнікі Kafka ніколі не гасяць свае кластары і праграмнае забеспячэнне заўсёды забяспечвае абнаўленне шляхам паслядоўнага рэстарту. Гэта дасягаецца за кошт гарантавання сумяшчальнасці з папярэдняй версіяй для паведамленняў і ўзаемадзеянняў паміж брокерамі.

Брокеры падлучаныя да кластара сервераў Зоопарк, Які дзейнічае, як рэестр канфігурацыйных дадзены і выкарыстоўваецца для каардынацыі роляў кожнага брокера. ZooKeeper сам з'яўляецца размеркаванай сістэмай, якая забяспечвае высокую даступнасць з дапамогай рэплікацыі інфармацыі шляхам устанаўлення кворуму.

У базавым выпадку топік ствараецца ў кластары Kafka з наступнымі ўласцівасцямі:

Колькасць партыцый. Як абмяркоўвалася раней, дакладнае значэнне, якое выкарыстоўваецца тут, залежыць ад жаданага ўзроўню паралельнага чытання.
Каэфіцыент (фактар) рэплікацыі вызначае, колькі асобнікаў брокера ў кластары павінны змяшчаць часопісы для гэтай партіціі.

Выкарыстоўваючы ZooKeepers для каардынацыі, Kafka спрабуе справядліва размеркаваць новыя партіціі паміж брокерамі ў кластары. Гэта робіцца адным асобнікам, які выконвае ролю Кантролера.

У рантайме для кожнай партіціі топіка кантролер прызначае брокеру ролі лідэра (leader, master, кіроўнага) і паслядоўнікаў (followers, slaves, падпарадкаваных). Брокер, які выступае ў якасці лідэра для дадзенай партіціі, адказвае за прыём усіх паведамленняў, адпраўленых яму прадзюсарамі, і распаўсюджванне паведамленняў па кансюмерах. Пры адпраўцы паведамленняў у партыцыю топіка яны рэплікуюцца на ўсе вузлы брокера, якія выступаюць у якасці паслядоўнікаў для гэтай партіціі. Кожны вузел, які змяшчае часопісы для партіціі, называецца рэплікай. Брокер можа выступаць у якасці лідэра для адных партыцый і ў якасці паслядоўніка для іншых.

Паслядоўнік, які змяшчае ўсе паведамленні, якія захоўваюцца ў лідэра, называецца сінхранізаванай рэплікай (рэплікай, якая знаходзіцца ў сінхранізаваным стане, in-sync replica). Калі брокер, які выступае ў якасці лідэра для партіціі, адключаецца, любы брокер, які знаходзіцца ў актуалізаваным або сінхранізаваным стане для гэтай партіціі, можа ўзяць на сябе ролю лідэра. Гэта неверагодна ўстойлівы дызайн.

Часткай канфігурацыі прадзюсара з'яўляецца параметр acks, Які вызначае, колькі рэплік павінна пацвердзіць (acknowledge) атрыманне паведамлення, перш чым паток прыкладання працягне адпраўку: 0, 1 ці ўсё. Калі зададзена значэнне усё, то пры атрыманні паведамлення лідэр адправіць пацверджанне (confirmation) зваротна прадзюсару, як толькі атрымае пацверджанне (acknowledgements) запісы ад некалькіх рэплік (уключаючы саму сябе), вызначаных наладай топіка min.insync.replicas (па змаўчанні 1). Калі паведамленне не можа быць паспяхова рэпліцыравана, то прадзюсер выкліча выключэнне для прыкладання (NotEnoughReplicas або NotEnoughReplicasAfterAppend).

У тыповай канфігурацыі ствараецца топік з каэфіцыентам рэплікацыі 3 (1 лідэр, 2 паслядоўнікі для кожнай партыцыі) і параметр min.insync.replicas усталёўваецца ў значэнне 2. У гэтым выпадку, кластар будзе дапушчаць, каб адзін з брокераў, якія кіруюць партыцыяй топіка, мог адключацца без уплыву на кліенцкія прыкладанні.

Гэта вяртае нас да ўжо знаёмага кампрамісу паміж прадукцыйнасцю і надзейнасцю. Рэплікацыя адбываецца за кошт дадатковага часу чакання пацверджанняў (acknowledgments) ад паслядоўнікаў. Хоць, паколькі яна выконваецца паралельна, рэплікацыя, як мінімум на тры вузлы, мае такую ж прадукцыйнасць, як і на два (ігнаруючы павелічэнне выкарыстання прапускной здольнасці сеткі).

Выкарыстоўваючы гэтую схему рэплікацыі, Kafka спрытна пазбягае неабходнасці забяспечваць фізічны запіс кожнага паведамлення на дыск з дапамогай аперацыі sync (). Кожнае паведамленне, адпраўленае прадзюсарам, будзе запісана ў часопіс партыцыі, але, як абмяркоўвалася ў Раздзеле 2, запіс у файл першапачаткова выконваецца ў буфер аперацыйнай сістэмы. Калі гэтае паведамленне рэплікавана на іншы асобнік Kafka і знаходзіцца ў яго памяці, страта лідэра не азначае, што само паведамленне было страчана - яго можа ўзяць на сябе сінхранізаваная рэпліка.
Адмова ад неабходнасці выконваць аперацыю sync () азначае, што Kafka можа прымаць паведамленні са хуткасцю, з якой яна можа запісваць іх у памяць. І наадварот, чым даўжэй можна пазбегнуць скіду (flushing) памяці на дыск, тым лепш. Па гэтай прычыне нярэдкія выпадкі, калі брокерам Kafka вылучаецца 64 Гб памяці ці больш. Такое выкарыстанне памяці азначае, што адзін асобнік Kafka можа лёгка працаваць на хуткасцях у шмат тысяч разоў хутчэй, чым традыцыйны брокер паведамленняў.

Kafka таксама можна наладзіць для прымянення аперацыі sync () да пакетаў паведамленняў. Паколькі ўсё ў Kafka арыентавана на працу з пакетамі, гэта насамрэч працуе даволі добра для шматлікіх сцэнараў выкарыстання і з'яўляецца карыснай прыладай для карыстачоў, якія патрабуюць вельмі моцных гарантый. Вялікая частка чыстай прадукцыйнасці Kafka злучана з паведамленнямі, якія адпраўляюцца брокеру ў выглядзе пакетаў, і з тым, што гэтыя паведамленні счытваюцца з брокера паслядоўнымі блокамі з дапамогай zero-copy аперацый (аперацыямі, падчас якіх не выконваецца задача капіявання дадзеных з адной вобласці памяці ў іншую). Апошняе з'яўляецца вялікім выйгрышам з пункта гледжання прадукцыйнасці і рэсурсаў і магчыма толькі дзякуючы выкарыстанню ляжалай у аснове структуры дадзеных часопіса, якая вызначае схему партіціі.

У кластары Kafka магчымая значна больш высокая прадукцыйнасць, чым пры выкарыстанні аднаго брокера Kafka, паколькі партіціі топіка могуць гарызантальна маштабавацца на мностве асобных машын.

Вынікі

У гэтым раздзеле мы разгледзелі, як архітэктура Kafka пераасэнсоўвае адносіны паміж кліентамі і брокерамі, каб забяспечыць неверагодна ўстойлівы канвеер абмену паведамленнямі, з прапускной здольнасцю ў шмат разоў большай, чым у звычайнага брокера паведамленняў. Мы абмеркавалі функцыянальнасць, якую яна выкарыстоўвае для дасягнення гэтай мэты, і коратка разгледзелі архітэктуру прыкладанняў, якія забяспечваюць гэтую функцыянальнасць. У наступным раздзеле мы разгледзім агульныя праблемы, якія неабходна вырашаць прыкладанням на аснове абмену паведамленнямі, і абмяркуем стратэгіі іх рашэння. Мы завершым кіраўніка, абмаляваўшы, як разважаць аб тэхналогіях абмену паведамленнямі ў цэлым, каб вы маглі ацаніць іх прыдатнасць для вашых сцэнарыяў выкарыстання.

Папярэдняя перакладзеная частка: Разуменне брокераў паведамленняў. Вывучэнне механікі абмену паведамленнямі з дапамогай ActiveMQ і Kafka. Кіраўнік 1

Пераклад выкананы: tele.gg/middle_java

Працяг будзе…

Толькі зарэгістраваныя карыстачы могуць удзельнічаць у апытанні. Увайдзіце, Калі ласка.

Ці выкарыстоўваецца Kafka у вашай арганізацыі?

Так
Няма
Раней выкарыстоўвалася, зараз няма
Плануем выкарыстоўваць

Прагаласавалі 38 карыстальнікаў. Устрымаліся 8 карыстальнікаў.

Крыніца: habr.com

Разуменне брокераў паведамленняў. Вывучэнне механікі абмену паведамленнямі з дапамогай ActiveMQ і Kafka. Кіраўнік 3. Kafka