របៀបដែល Kafka ក្លាយជាការពិត

ហេហេ!

ខ្ញុំធ្វើការលើក្រុម Tinkoff ដែលកំពុងបង្កើតមជ្ឈមណ្ឌលជូនដំណឹងផ្ទាល់ខ្លួន។ ខ្ញុំភាគច្រើនអភិវឌ្ឍនៅ Java ដោយប្រើ Spring boot និងដោះស្រាយបញ្ហាបច្ចេកទេសផ្សេងៗដែលកើតឡើងនៅក្នុងគម្រោងមួយ។

ភាគច្រើននៃសេវាមីក្រូរបស់យើងទាក់ទងគ្នាទៅវិញទៅមកដោយអសមកាលតាមរយៈឈ្មួញកណ្តាលសារ។ ពីមុន យើងបានប្រើ IBM MQ ជាឈ្មួញកណ្តាល ដែលមិនអាចទប់ទល់នឹងបន្ទុកបានទៀតទេ ប៉ុន្តែនៅពេលជាមួយគ្នានោះមានការធានាលើការដឹកជញ្ជូនខ្ពស់។

ក្នុងនាមជាអ្នកជំនួស យើងត្រូវបានផ្តល់ជូន Apache Kafka ដែលមានសក្ដានុពលនៃការធ្វើមាត្រដ្ឋានខ្ពស់ ប៉ុន្តែជាអកុសល ទាមទារវិធីសាស្រ្តបុគ្គលស្ទើរតែទាំងអស់ក្នុងការកំណត់រចនាសម្ព័ន្ធសម្រាប់សេណារីយ៉ូផ្សេងៗ។ លើសពីនេះ យន្តការបញ្ជូនយ៉ាងហោចណាស់ម្តងដែលដំណើរការនៅក្នុង Kafka តាមលំនាំដើមមិនអនុញ្ញាតឱ្យរក្សាបាននូវកម្រិតដែលត្រូវការនៃភាពស៊ីសង្វាក់គ្នាក្រៅប្រអប់នោះទេ។ បន្ទាប់ ខ្ញុំនឹងចែករំលែកបទពិសោធន៍របស់យើងក្នុងការកំណត់រចនាសម្ព័ន្ធ Kafka ជាពិសេសខ្ញុំនឹងប្រាប់អ្នកពីរបៀបកំណត់រចនាសម្ព័ន្ធ និងរស់នៅជាមួយការចែកចាយភ្លាមៗ។

ធានាការដឹកជញ្ជូន និងច្រើនទៀត

ការកំណត់ដែលបានពិភាក្សាខាងក្រោមនឹងជួយការពារបញ្ហាមួយចំនួនជាមួយនឹងការកំណត់ការតភ្ជាប់លំនាំដើម។ ប៉ុន្តែជាដំបូងខ្ញុំចង់យកចិត្តទុកដាក់លើប៉ារ៉ាម៉ែត្រមួយដែលនឹងជួយសម្រួលដល់ការបំបាត់កំហុសដែលអាចកើតមាន។

នេះនឹងជួយ client.id សម្រាប់អ្នកផលិត និងអ្នកប្រើប្រាស់។ នៅ glance ដំបូងអ្នកអាចប្រើឈ្មោះកម្មវិធីជាតម្លៃហើយក្នុងករណីភាគច្រើនវានឹងដំណើរការ។ ទោះបីជាស្ថានភាពនៅពេលដែលកម្មវិធីប្រើប្រាស់អ្នកប្រើប្រាស់ជាច្រើន ហើយអ្នកផ្តល់ឱ្យពួកគេនូវ 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 នៅក្នុងកម្មវិធីជាមួយ Kafka នោះវាអាចជាបញ្ហា។ សម្រាប់ករណីនេះ វាជាការល្អបំផុតក្នុងការប្រើការរួមបញ្ចូលគ្នានៃឈ្មោះកម្មវិធី និងឧទាហរណ៍ឈ្មោះប្រធានបទជាតម្លៃ client.id ។ លទ្ធផលនៃការកំណត់រចនាសម្ព័ន្ធរបស់យើងអាចមើលឃើញនៅក្នុងលទ្ធផលពាក្យបញ្ជា kafka-consumer-groups ពីឧបករណ៍ប្រើប្រាស់ពី Confluent:

ឥឡូវនេះសូមក្រឡេកមើលសេណារីយ៉ូសម្រាប់ការបញ្ជូនសារដែលមានការធានា។ អ្នកផលិត Kafka មានប៉ារ៉ាម៉ែត្រ អេកដែលអនុញ្ញាតឱ្យអ្នកកំណត់រចនាសម្ព័ន្ធបន្ទាប់ពីចំនួនប៉ុន្មានដែលទទួលស្គាល់អ្នកដឹកនាំចង្កោមត្រូវការដើម្បីពិចារណាសារដែលបានសរសេរដោយជោគជ័យ។ ប៉ារ៉ាម៉ែត្រនេះអាចយកតម្លៃដូចខាងក្រោមៈ

• 0 - ការទទួលស្គាល់នឹងមិនត្រូវបានពិចារណាទេ។
• 1 គឺជាប៉ារ៉ាម៉ែត្រលំនាំដើម មានតែការចម្លង 1 ប៉ុណ្ណោះដែលត្រូវបានទាមទារដើម្បីទទួលស្គាល់។
• −1 — ការទទួលស្គាល់ពីការចម្លងដែលបានធ្វើសមកាលកម្មទាំងអស់គឺត្រូវបានទាមទារ (ការដំឡើងចង្កោម min.insync.replicas).

ពីតម្លៃដែលបានរាយបញ្ជី វាច្បាស់ណាស់ថា acks ស្មើនឹង −1 ផ្តល់ការធានាខ្លាំងបំផុតថាសារនឹងមិនបាត់បង់ឡើយ។

ដូចដែលយើងដឹងស្រាប់ហើយ ប្រព័ន្ធចែកចាយមិនគួរឱ្យទុកចិត្ត។ ដើម្បីការពារប្រឆាំងនឹងកំហុសបណ្តោះអាសន្ន Kafka Producer ផ្តល់ជម្រើស ព្យាយាមម្តងទៀតដែលអនុញ្ញាតឱ្យអ្នកកំណត់ចំនួននៃការព្យាយាមផ្ញើឡើងវិញនៅក្នុង delivery.timeout.ms. ដោយសារប៉ារ៉ាម៉ែត្រព្យាយាមម្តងទៀតមានតម្លៃលំនាំដើមនៃ Integer.MAX_VALUE (2147483647) ចំនួននៃការព្យាយាមសារឡើងវិញអាចត្រូវបានកែតម្រូវដោយការផ្លាស់ប្តូរតែ delivery.timeout.ms ប៉ុណ្ណោះ។

យើងកំពុងឆ្ពោះទៅរកការចែកចាយភ្លាមៗ

ការកំណត់ដែលបានរាយបញ្ជីអនុញ្ញាតឱ្យអ្នកផលិតរបស់យើងបញ្ជូនសារជាមួយនឹងការធានាខ្ពស់។ ឥឡូវ​នេះ​សូម​និយាយ​អំពី​របៀប​ដើម្បី​ធានា​ថា​សារ​តែ​មួយ​ច្បាប់​ត្រូវ​បាន​សរសេរ​ទៅ​កាន់​ប្រធាន​បទ Kafka? ក្នុងករណីសាមញ្ញបំផុតដើម្បីធ្វើវាអ្នកត្រូវកំណត់ប៉ារ៉ាម៉ែត្រនៅលើអ្នកផលិត enable.idempotence ពិត។ Ideempotency ធានាថាមានតែសារមួយប៉ុណ្ណោះដែលត្រូវបានសរសេរទៅភាគជាក់លាក់នៃប្រធានបទមួយ។ លក្ខខណ្ឌជាមុនសម្រាប់ការធ្វើឱ្យមានភាពទន់ខ្សោយគឺជាតម្លៃ acks = ទាំងអស់, ព្យាយាមម្តងទៀត > 0, max.in.flight.requests.per.connection ≤ 5. ប្រសិនបើប៉ារ៉ាម៉ែត្រទាំងនេះមិនត្រូវបានបញ្ជាក់ដោយអ្នកអភិវឌ្ឍន៍ទេនោះតម្លៃខាងលើនឹងត្រូវបានកំណត់ដោយស្វ័យប្រវត្តិ។

នៅពេលដែល idempotency ត្រូវបានកំណត់រចនាសម្ព័ន្ធ វាចាំបាច់ដើម្បីធានាថាសារដូចគ្នាបញ្ចប់នៅក្នុងភាគថាសដូចគ្នារាល់ពេល។ នេះអាចត្រូវបានធ្វើដោយកំណត់កូនសោ partitioner.class និងប៉ារ៉ាម៉ែត្រទៅអ្នកផលិត។ ចូរចាប់ផ្តើមជាមួយនឹងគន្លឹះ។ វាត្រូវតែដូចគ្នាសម្រាប់ការដាក់ស្នើនីមួយៗ។ នេះអាចសម្រេចបានយ៉ាងងាយស្រួលដោយប្រើលេខសម្គាល់អាជីវកម្មណាមួយពីប្រកាសដើម។ ប៉ារ៉ាម៉ែត្រ partitioner.class មានតម្លៃលំនាំដើម − ភាគថាសលំនាំដើម. ជាមួយនឹងយុទ្ធសាស្ត្របែងចែកនេះ តាមលំនាំដើម យើងធ្វើសកម្មភាពដូចនេះ៖

• ប្រសិនបើភាគថាសត្រូវបានបញ្ជាក់យ៉ាងច្បាស់នៅពេលផ្ញើសារ នោះយើងប្រើវា។
• ប្រសិនបើភាគថាសមិនត្រូវបានបញ្ជាក់ ប៉ុន្តែកូនសោត្រូវបានបញ្ជាក់ សូមជ្រើសរើសភាគថាសដោយសញ្ញានៃគ្រាប់ចុច។
• ប្រសិនបើភាគថាស និងគ្រាប់ចុចមិនត្រូវបានបញ្ជាក់ សូមជ្រើសរើសភាគថាសម្តងមួយៗ (ជុំ-រ៉ូប៊ីន)។

ផងដែរដោយប្រើសោ និង ideempotent ផ្ញើជាមួយប៉ារ៉ាម៉ែត្រមួយ។ max.in.flight.requests.per.connection = 1 ផ្តល់ឱ្យអ្នកនូវដំណើរការសារដែលសម្រួលនៅលើអ្នកប្រើប្រាស់។ វាក៏គួរចងចាំផងដែរថា ប្រសិនបើការគ្រប់គ្រងការចូលប្រើត្រូវបានកំណត់រចនាសម្ព័ន្ធនៅលើចង្កោមរបស់អ្នក នោះអ្នកនឹងត្រូវការសិទ្ធិក្នុងការសរសេរដោយអចេតនាចំពោះប្រធានបទមួយ។

ប្រសិនបើភ្លាមៗនោះអ្នកខ្វះសមត្ថភាពនៃការបញ្ជូនដោយកូនសោ ឬតក្កវិជ្ជានៅខាងអ្នកផលិតតម្រូវឱ្យរក្សាភាពស៊ីសង្វាក់គ្នានៃទិន្នន័យរវាងភាគថាសផ្សេងៗគ្នា នោះប្រតិបត្តិការនឹងមកជួយសង្គ្រោះ។ លើសពីនេះទៀត ដោយប្រើប្រតិបត្តិការខ្សែសង្វាក់ អ្នកអាចធ្វើសមកាលកម្មកំណត់ត្រានៅក្នុង Kafka ជាឧទាហរណ៍ ជាមួយនឹងកំណត់ត្រានៅក្នុងមូលដ្ឋានទិន្នន័យ។ ដើម្បីបើកដំណើរការការផ្ញើទៅអ្នកផលិត វាត្រូវតែគ្មានសមត្ថភាព និងកំណត់បន្ថែម transactional.id. ប្រសិនបើចង្កោម Kafka របស់អ្នកមានការគ្រប់គ្រងការចូលប្រើដែលបានកំណត់រចនាសម្ព័ន្ធ នោះកំណត់ត្រាប្រតិបត្តិការ ដូចជាកំណត់ត្រា ideempotent នឹងត្រូវការការអនុញ្ញាតក្នុងការសរសេរ ដែលអាចត្រូវបានផ្តល់ដោយរបាំងដោយប្រើតម្លៃដែលបានរក្សាទុកនៅក្នុង transactional.id ។

ជាផ្លូវការ ខ្សែអក្សរណាមួយ ដូចជាឈ្មោះកម្មវិធី អាចត្រូវបានប្រើជាអត្តសញ្ញាណប្រតិបត្តិការ។ ប៉ុន្តែប្រសិនបើអ្នកបើកដំណើរការឧទាហរណ៍ជាច្រើននៃកម្មវិធីដូចគ្នាជាមួយ transactional.id ដូចគ្នា នោះឧទាហរណ៍ដែលបានបើកដំណើរការដំបូងនឹងត្រូវបញ្ឈប់ដោយមានកំហុស ព្រោះ Kafka នឹងចាត់ទុកវាជាដំណើរការខ្មោចឆៅ។

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.

ដើម្បីដោះស្រាយបញ្ហានេះ យើងបន្ថែមបច្ច័យទៅឈ្មោះកម្មវិធីក្នុងទម្រង់ជា hostname ដែលយើងទទួលបានពីអថេរបរិស្ថាន។

អ្នកផលិតត្រូវបានកំណត់រចនាសម្ព័ន្ធ ប៉ុន្តែប្រតិបត្តិការនៅលើ Kafka គ្រប់គ្រងតែវិសាលភាពនៃសារប៉ុណ្ណោះ។ ដោយមិនគិតពីស្ថានភាពប្រតិបត្តិការ សារភ្លាមៗទៅកាន់ប្រធានបទ ប៉ុន្តែមានគុណលក្ខណៈប្រព័ន្ធបន្ថែម។

ដើម្បីបងា្ករសារបែបនេះមិនត្រូវបានអានដោយអ្នកប្រើប្រាស់ជាមុនគាត់ត្រូវកំណត់ប៉ារ៉ាម៉ែត្រ isolation.level ដើម្បីអាន_តម្លៃដែលបានសន្យា។ អ្នកប្រើប្រាស់បែបនេះនឹងអាចអានសារដែលមិនមែនជាប្រតិបត្តិការដូចពីមុន ហើយសារប្រតិបត្តិការបានលុះត្រាតែមានការប្តេជ្ញាចិត្ត។
ប្រសិនបើអ្នកបានកំណត់ការកំណត់ទាំងអស់ដែលបានរាយបញ្ជីមុននេះ នោះអ្នកបានកំណត់រចនាសម្ព័ន្ធភ្លាមៗនៅពេលចែកចាយ។ អបអរសាទរ!

ប៉ុន្តែមាន nuance មួយបន្ថែមទៀត។ Transactional.id ដែលយើងកំណត់រចនាសម្ព័ន្ធខាងលើ គឺពិតជាបុព្វបទប្រតិបត្តិការ។ នៅលើកម្មវិធីគ្រប់គ្រងប្រតិបត្តិការ លេខលំដាប់មួយត្រូវបានបន្ថែមទៅវា។ អត្តសញ្ញាណដែលបានទទួលត្រូវបានចេញឱ្យ transactional.id.expiration.msដែលត្រូវបានកំណត់រចនាសម្ព័ន្ធនៅលើចង្កោម Kafka និងមានតម្លៃលំនាំដើមនៃ "7 ថ្ងៃ" ។ ប្រសិនបើក្នុងអំឡុងពេលនេះ កម្មវិធីមិនបានទទួលសារណាមួយទេ នោះនៅពេលដែលអ្នកព្យាយាមផ្ញើប្រតិបត្តិការបន្ទាប់ អ្នកនឹងទទួលបាន InvalidPidMappingException. បន្ទាប់មក អ្នកសម្របសម្រួលប្រតិបត្តិការនឹងចេញលេខលំដាប់ថ្មីសម្រាប់ប្រតិបត្តិការបន្ទាប់។ ទោះយ៉ាងណាក៏ដោយ សារអាចនឹងត្រូវបាត់បង់ ប្រសិនបើ InvalidPidMappingException មិនត្រូវបានគ្រប់គ្រងត្រឹមត្រូវ។

ជំនួសឱ្យចំនួនសរុប

ដូចដែលអ្នកអាចឃើញវាមិនគ្រប់គ្រាន់ទេក្នុងការគ្រាន់តែផ្ញើសារទៅ Kafka ។ អ្នកត្រូវជ្រើសរើសបន្សំនៃប៉ារ៉ាម៉ែត្រ ហើយត្រៀមខ្លួនដើម្បីធ្វើការផ្លាស់ប្តូររហ័ស។ នៅក្នុងអត្ថបទនេះ ខ្ញុំបានព្យាយាមបង្ហាញយ៉ាងលម្អិតអំពីការរៀបចំការចែកចាយភ្លាមៗ ហើយបានពិពណ៌នាអំពីបញ្ហាជាច្រើនជាមួយនឹងការកំណត់រចនាសម្ព័ន្ធ client.id និង transactional.id ដែលយើងជួបប្រទះ។ ខាងក្រោមនេះគឺជាសេចក្តីសង្ខេបនៃការកំណត់របស់អ្នកផលិត និងអ្នកប្រើប្រាស់។

ក្រុមហ៊ុនផលិត:

1. acks = ទាំងអស់។
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 - ភ្លាមៗនៅពេលដែលការដឹកជញ្ជូន និងការផ្ញើសារតាមប្រតិបត្តិការ
• ការពិពណ៌នាអំពីការកំណត់

ប្រភព: www.habr.com