🥇 Awọn iṣẹlẹ atunṣeto ti a gba lati ọdọ Kafka

Hello, Habr.

Laipe I pín rẹ iriri nipa kini awọn paramita ti a bi ẹgbẹ kan lo nigbagbogbo fun Olupese Kafka ati Olumulo lati sunmọ si ifijiṣẹ iṣeduro. Ninu àpilẹkọ yii Mo fẹ lati sọ fun ọ bi a ṣe ṣeto ilana atunṣe ti iṣẹlẹ ti a gba lati ọdọ Kafka nitori abajade wiwa igba diẹ ti eto ita.

Awọn ohun elo ode oni ṣiṣẹ ni agbegbe eka pupọ. Imọye iṣowo ti a we sinu akopọ imọ-ẹrọ ode oni, ṣiṣiṣẹ ni aworan Docker ti iṣakoso nipasẹ akọrin bi Kubernetes tabi OpenShift, ati sisọ pẹlu awọn ohun elo miiran tabi awọn solusan ile-iṣẹ nipasẹ pq ti awọn olulana ti ara ati foju. Ni iru agbegbe bẹẹ, ohunkan le fọ nigbagbogbo, nitorinaa awọn iṣẹlẹ atunṣe ti ọkan ninu awọn eto ita ko si jẹ apakan pataki ti awọn ilana iṣowo wa.

Bawo ni ṣaaju ki Kafka

Ni iṣaaju ninu iṣẹ akanṣe a lo IBM MQ fun ifijiṣẹ asynchronous. Ti eyikeyi aṣiṣe ba waye lakoko iṣiṣẹ iṣẹ naa, ifiranṣẹ ti o gba le wa ni gbe sinu isinyi-lẹta ti o ku (DLQ) fun sisọ afọwọṣe siwaju sii. A ṣẹda DLQ lẹgbẹẹ isinyi ti nwọle, ti gbe ifiranṣẹ naa sinu IBM MQ.

Ti aṣiṣe naa ba jẹ igba diẹ ati pe a le pinnu rẹ (fun apẹẹrẹ, ResourceAccessException lori ipe HTTP tabi MongoTimeoutException lori ibeere MongoDb kan), lẹhinna ilana atunyẹwo yoo ni ipa. Laibikita imọ-ọrọ eka ti ohun elo naa, ifiranṣẹ atilẹba ti gbe boya si isinyi eto fun fifiranṣẹ idaduro, tabi si ohun elo lọtọ ti a ṣe ni pipẹ sẹhin lati firanṣẹ awọn ifiranṣẹ. Eyi pẹlu nọmba atunṣe kan ninu akọsori ifiranṣẹ, eyiti o so mọ aarin idaduro tabi ipari ilana-ipele ohun elo. Ti a ba ti de opin ilana naa ṣugbọn eto ita ko si, lẹhinna ifiranṣẹ naa yoo gbe sinu DLQ fun sisọ afọwọṣe.

Wiwa Ojutu

Wiwa lori Intanẹẹti, o le wa awọn wọnyi ipinnu naa. Ni kukuru, o dabaa lati ṣẹda koko-ọrọ fun aarin idaduro kọọkan ati ṣe awọn ohun elo Olumulo ni ẹgbẹ, eyiti yoo ka awọn ifiranṣẹ pẹlu idaduro ti o nilo.

Pelu awọn ti o tobi nọmba ti rere agbeyewo, o dabi si mi ko šee igbọkanle aseyori. Ni akọkọ, nitori olupilẹṣẹ, ni afikun si imuse awọn ibeere iṣowo, yoo ni lati lo akoko pupọ ni imuse ẹrọ ti a ṣalaye.

Ni afikun, ti iṣakoso wiwọle ba ṣiṣẹ lori iṣupọ Kafka, iwọ yoo ni lati lo akoko diẹ ṣiṣẹda awọn akọle ati pese iraye si pataki si wọn. Ni afikun si eyi, iwọ yoo nilo lati yan paramita retention.ms ti o pe fun ọkọọkan awọn koko-ọrọ atungbiyanju ki awọn ifiranṣẹ le ni akoko lati firanṣẹ ati ma ṣe parẹ lati ọdọ rẹ. Imuse ati ibeere wiwọle yoo ni lati tun ṣe fun ọkọọkan ti o wa tabi iṣẹ tuntun.

Jẹ ki a ni bayi wo kini awọn ilana orisun omi ni gbogbogbo ati orisun omi-kafka ni pataki pese wa pẹlu fun atunṣe ifiranṣẹ. Orisun omi-kafka ni igbẹkẹle iyipada lori orisun omi-tun gbiyanju, eyiti o pese awọn abstractions fun ṣiṣakoso awọn oriṣiriṣi BackOffPolicies. Eyi jẹ ohun elo ti o rọ ni deede, ṣugbọn apadabọ pataki rẹ ni titoju awọn ifiranṣẹ fun fifiranṣẹ ni iranti ohun elo. Eyi tumọ si pe atunbere ohun elo naa nitori imudojuiwọn tabi aṣiṣe iṣẹ kan yoo ja si ipadanu gbogbo awọn ifiranṣẹ ni isunmọ atunto. Niwọn bi aaye yii ṣe pataki fun eto wa, a ko gbero siwaju sii.

orisun omi-kafka funrararẹ pese awọn imuse pupọ ti ContainerAwareErrorHandler, fun apẹẹrẹ SeekToCurrentErrorHandler, pẹlu eyiti o le ṣe ilana ifiranṣẹ nigbamii laisi aiṣedeede iyipada ni ọran ti aṣiṣe. Bibẹrẹ pẹlu ẹya orisun omi-kafka 2.3, o ṣee ṣe lati ṣeto BackOffPolicy.

Ọna yii ngbanilaaye awọn ifiranšẹ atunto lati ye awọn atunbẹrẹ ohun elo, ṣugbọn ko si ẹrọ DLQ sibẹsibẹ. A yan aṣayan yii ni ibẹrẹ ọdun 2019, ni igbagbọ ni ireti pe DLQ kii yoo nilo (a ni orire ati ni otitọ ko nilo rẹ lẹhin awọn oṣu pupọ ti ṣiṣiṣẹ ohun elo pẹlu iru eto atunṣe). Awọn aṣiṣe igba diẹ fa SeekToCurrentErrorHandler lati ina. Awọn aṣiṣe ti o ku ni a tẹjade ninu akọọlẹ, ti o yọrisi aiṣedeede, ati sisẹ tẹsiwaju pẹlu ifiranṣẹ atẹle.

Ipinnu ikẹhin

Imuse ti o da lori SeekToCurrentErrorHandler jẹ ki a ṣe agbekalẹ ẹrọ tiwa fun fifiranṣẹ awọn ifiranṣẹ.

Ni akọkọ, a fẹ lati lo iriri ti o wa ati faagun rẹ da lori ọgbọn ohun elo. Fun ohun elo kannaa laini, yoo dara julọ lati da kika awọn ifiranṣẹ titun duro fun igba diẹ ti a pato nipasẹ ilana atunyanju. Fun awọn ohun elo miiran, Mo fẹ lati ni aaye kan ti yoo fi ipa mu ilana atunwi naa. Ni afikun, aaye kan ṣoṣo yii gbọdọ ni iṣẹ ṣiṣe DLQ fun awọn ọna mejeeji.

Ilana atunwi funrararẹ gbọdọ wa ni ipamọ ninu ohun elo, eyiti o jẹ iduro fun gbigba pada aarin atẹle nigbati aṣiṣe igba diẹ ba waye.

Idaduro Olumulo fun Ohun elo Logic Linear

Nigbati o ba n ṣiṣẹ pẹlu orisun omi-kafka, koodu lati da Olumulo naa duro le dabi iru eyi:

public void pauseListenerContainer(MessageListenerContainer listenerContainer, 
                                   Instant retryAt) {
        if (nonNull(retryAt) && listenerContainer.isRunning()) {
            listenerContainer.stop();
            taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
            return;
        }
        // to DLQ
    }

Ninu apẹẹrẹ, retryAt ni akoko lati tun MessageListenerContainer bẹrẹ ti o ba tun nṣiṣẹ. Atun-ifilọlẹ yoo waye ni okun lọtọ ti a ṣe ifilọlẹ ni TaskScheduler, imuse eyiti o tun pese nipasẹ orisun omi.

A tun rii iye atunyẹwo ni ọna atẹle:

Awọn iye ti awọn tun-ipe counter ni a wo soke.
Da lori iye counter, aarin idaduro lọwọlọwọ ninu ilana atunyanwo ni a wa. Ilana naa jẹ ikede ninu ohun elo funrararẹ; a yan ọna kika JSON lati tọju rẹ.
Aarin akoko ti a rii ni titobi JSON ni nọmba awọn iṣẹju-aaya lẹhin eyiti iṣelọpọ yoo nilo lati tun ṣe. Nọmba awọn iṣẹju-aaya yii jẹ afikun si akoko lọwọlọwọ lati dagba iye fun atundaAt.
Ti a ko ba ri aarin, lẹhinna iye retryAt jẹ asan ati pe ifiranṣẹ naa yoo fi ranṣẹ si DLQ fun ṣiṣe itupalẹ afọwọṣe.

Pẹlu ọna yii, gbogbo ohun ti o ku ni lati ṣafipamọ nọmba awọn ipe ti o leralera fun ifiranṣẹ kọọkan ti n ṣiṣẹ lọwọlọwọ, fun apẹẹrẹ ni iranti ohun elo. Titọju kika atunyẹwo ni iranti kii ṣe pataki fun ọna yii, nitori ohun elo ọgbọn laini ko le mu sisẹ naa lapapọ. Ko dabi orisun omi-tun gbiyanju, atunbere ohun elo kii yoo fa ki gbogbo awọn ifiranṣẹ sọnu lati tun ṣe, ṣugbọn yoo tun bẹrẹ ilana naa.

Ọna yii ṣe iranlọwọ lati mu fifuye kuro ni eto ita, eyiti o le ma wa nitori ẹru iwuwo pupọ. Ni awọn ọrọ miiran, ni afikun si atunṣe, a ṣe aṣeyọri imuse ti ilana naa Opin Iyika monamona.

Ninu ọran wa, ẹnu-ọna aṣiṣe jẹ 1 nikan, ati lati dinku akoko idinku eto nitori awọn ijakadi nẹtiwọọki igba diẹ, a lo ilana atunyẹwo granular pupọ pẹlu awọn aaye arin kekere. Eyi le ma dara fun gbogbo awọn ohun elo ẹgbẹ, nitorinaa ibatan laarin ẹnu-ọna aṣiṣe ati iye aarin gbọdọ jẹ yiyan ti o da lori awọn abuda ti eto naa.

Ohun elo lọtọ fun sisẹ awọn ifiranṣẹ lati awọn ohun elo pẹlu ọgbọn-ipinnu ti kii ṣe ipinnu

Eyi ni apẹẹrẹ koodu ti o fi ifiranṣẹ ranṣẹ si iru ohun elo (Retryer), eyiti yoo tun fi ranṣẹ si koko DESTINATION nigbati akoko RETRY_AT ba ti de:


public <K, V> void retry(ConsumerRecord<K, V> record, String retryToTopic, 
                         Instant retryAt, String counter, String groupId, Exception e) {
        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());
        List<Header> arrayOfHeaders = 
            new ArrayList<>(Arrays.asList(headers.toArray()));
        updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);
        updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);
        updateHeader(arrayOfHeaders, ORIGINAL_PARTITION, 
                     () -> Integer.toString(record.partition()).getBytes());
        if (nonNull(retryAt)) {
            updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);
            updateHeader(arrayOfHeaders, SEND_TO, "retry"::getBytes);
            updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);
        } else {
            updateHeader(arrayOfHeaders, REASON, 
                         ExceptionUtils.getStackTrace(e)::getBytes);
            updateHeader(arrayOfHeaders, SEND_TO, "backout"::getBytes);
        }
        ProducerRecord<K, V> messageToSend =
            new ProducerRecord<>(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);
        kafkaTemplate.send(messageToSend);
    }

Apẹẹrẹ fihan pe ọpọlọpọ alaye ti wa ni gbigbe ni awọn akọle. Iye RETRY_AT ni a rii ni ọna kanna bi fun ẹrọ atunbere nipasẹ iduro Onibara. Ni afikun si DESTINATION ati RETRY_AT a kọja:

GROUP_ID, nipasẹ eyiti a ṣe akojọpọ awọn ifiranṣẹ fun itupalẹ afọwọṣe ati wiwa irọrun.
ORIGINAL_PARTITION lati gbiyanju lati tọju Onibara kanna fun tun-ṣiṣẹ. Paramita yii le jẹ asan, ninu ọran eyiti ipin tuntun yoo gba ni lilo bọtini record.key () ti ifiranṣẹ atilẹba naa.
Ṣe imudojuiwọn iye COUNTER lati tẹle ilana atunyanwo.
SEND_TO jẹ igbagbogbo ti o nfihan boya a fi ifiranṣẹ ranṣẹ fun atunṣe nigbati o ba de RETRY_AT tabi fi sii ni DLQ.
IDI – idi ti siseto ifiranṣẹ ti di idilọwọ.

Retryer tọju awọn ifiranṣẹ pamọ fun fifiranṣẹ ati ṣiṣe itupalẹ afọwọṣe ni PostgreSQL. Aago kan bẹrẹ iṣẹ kan ti o wa awọn ifiranṣẹ pẹlu RETRY_AT ti o si fi wọn ranṣẹ pada si apakan ORIGINAL_PARTITION ti koko DESTINATION pẹlu bọtini record.key().

Ni kete ti o ti firanṣẹ, awọn ifiranṣẹ ti paarẹ lati PostgreSQL. Ṣiṣayẹwo awọn ifiranṣẹ afọwọṣe waye ni UI ti o rọrun ti o ṣe ajọṣepọ pẹlu Retryer nipasẹ REST API. Awọn ẹya akọkọ rẹ jẹ fifiranṣẹ tabi piparẹ awọn ifiranṣẹ lati DLQ, wiwo alaye aṣiṣe ati wiwa awọn ifiranṣẹ, fun apẹẹrẹ nipasẹ orukọ aṣiṣe.

Niwọn igba ti iṣakoso wiwọle ti ṣiṣẹ lori awọn iṣupọ wa, o jẹ dandan lati ni afikun lati beere iraye si koko-ọrọ ti Retryer n tẹtisi, ki o jẹ ki Retryer lati kọ si koko DESTINATION. Eyi korọrun, ṣugbọn, ko dabi ọna koko aarin, a ni DLQ ti o ni kikun ati UI lati ṣakoso rẹ.

Awọn ọran wa nigbati akọle ti nwọle ba ka nipasẹ ọpọlọpọ awọn ẹgbẹ olumulo oriṣiriṣi, eyiti awọn ohun elo wọn ṣe imuse awọn ọgbọn oriṣiriṣi. Ṣiṣe atunṣe ifiranṣẹ nipasẹ Retryer fun ọkan ninu awọn ohun elo wọnyi yoo ja si ni ẹda-ẹda lori ekeji. Lati daabobo lodi si eyi, a ṣẹda koko-ọrọ lọtọ fun tun-ṣiṣẹ. Awọn koko-ọrọ ti nwọle ati tun gbiyanju le jẹ kika nipasẹ Olumulo kanna laisi awọn ihamọ eyikeyi.

Nipa aiyipada ọna yii ko pese iṣẹ ṣiṣe fifọ Circuit, sibẹsibẹ o le ṣe afikun si ohun elo nipa lilo orisun omi-awọsanma-netflix tabi titun orisun omi awọsanma Circuit fifọ, N murasilẹ awọn aaye nibiti a ti pe awọn iṣẹ ita si awọn abstractions ti o yẹ. Ni afikun, o di ṣee ṣe lati yan a nwon.Mirza fun olopobobo apẹrẹ, eyiti o tun le wulo. Fun apẹẹrẹ, ni orisun omi-cloud-netflix eyi le jẹ adagun okun tabi semaphore kan.

ipari

Bi abajade, a ni ohun elo lọtọ ti o gba wa laaye lati tun sisẹ ifiranṣẹ kan ti eto ita eyikeyi ko ba si fun igba diẹ.

Ọkan ninu awọn anfani akọkọ ti ohun elo ni pe o le ṣee lo nipasẹ awọn ọna ṣiṣe ita ti nṣiṣẹ lori iṣupọ Kafka kanna, laisi awọn iyipada pataki ni ẹgbẹ wọn! Iru ohun elo kan yoo nilo lati wọle si koko-ọrọ atunyẹwo, fọwọsi awọn akọle Kafka diẹ ki o fi ifiranṣẹ ranṣẹ si Retryer. Ko si ye lati gbe eyikeyi afikun amayederun. Ati pe lati le dinku nọmba awọn ifiranṣẹ ti o gbe lati ohun elo si Retryer ati sẹhin, a ṣe idanimọ awọn ohun elo pẹlu ọgbọn laini ati tun ṣe ilana wọn nipasẹ iduro Olumulo.

orisun: www.habr.com

Awọn iṣẹlẹ atunṣe ti a gba lati ọdọ Kafka