ວິທີການ Kafka ກາຍເປັນຄວາມເປັນຈິງ

Hey Habr!

ຂ້ອຍເຮັດວຽກຢູ່ໃນທີມງານ Tinkoff, ເຊິ່ງກໍາລັງພັດທະນາສູນແຈ້ງເຕືອນຂອງຕົນເອງ. ຂ້ອຍສ່ວນຫຼາຍແມ່ນພັດທະນາໃນ Java ໂດຍໃຊ້ Spring boot ແລະແກ້ໄຂບັນຫາດ້ານວິຊາການຕ່າງໆທີ່ເກີດຂື້ນໃນໂຄງການ.

ການບໍລິການຈຸນລະພາກສ່ວນໃຫຍ່ຂອງພວກເຮົາຕິດຕໍ່ສື່ສານກັນແບບບໍ່ກົງກັນຜ່ານທາງນາຍໜ້າຂໍ້ຄວາມ. ກ່ອນຫນ້ານີ້, ພວກເຮົາໃຊ້ IBM MQ ເປັນນາຍຫນ້າ, ເຊິ່ງບໍ່ສາມາດຮັບມືກັບການໂຫຼດໄດ້, ແຕ່ໃນເວລາດຽວກັນມີການຮັບປະກັນການຈັດສົ່ງສູງ.

ເປັນການທົດແທນ, ພວກເຮົາໄດ້ຖືກສະເຫນີໃຫ້ Apache Kafka, ທີ່ມີທ່າແຮງການຂະຫຍາຍສູງ, ແຕ່ຫນ້າເສຍດາຍ, ຮຽກຮ້ອງໃຫ້ມີວິທີການເກືອບສ່ວນບຸກຄົນໃນການຕັ້ງຄ່າສໍາລັບສະຖານະການທີ່ແຕກຕ່າງກັນ. ນອກຈາກນັ້ນ, ກົນໄກການຈັດສົ່ງຢ່າງຫນ້ອຍຫນຶ່ງຄັ້ງທີ່ເຮັດວຽກຢູ່ໃນ Kafka ໂດຍຄ່າເລີ່ມຕົ້ນບໍ່ໄດ້ອະນຸຍາດໃຫ້ຮັກສາລະດັບຄວາມສອດຄ່ອງທີ່ຕ້ອງການອອກຈາກກ່ອງ. ຕໍ່ໄປ, ຂ້າພະເຈົ້າຈະແບ່ງປັນປະສົບການຂອງພວກເຮົາໃນການຕັ້ງຄ່າ Kafka, ໂດຍສະເພາະ, ຂ້າພະເຈົ້າຈະບອກທ່ານວິທີການ configure ແລະດໍາລົງຊີວິດກັບແນ່ນອນຫນຶ່ງຄັ້ງການຈັດສົ່ງ.

ຮັບປະກັນການຈັດສົ່ງ ແລະອື່ນໆອີກ

ການຕັ້ງຄ່າທີ່ສົນທະນາຂ້າງລຸ່ມນີ້ຈະຊ່ວຍປ້ອງກັນບັນຫາຈໍານວນຫນຶ່ງກັບການຕັ້ງຄ່າການເຊື່ອມຕໍ່ໃນຕອນຕົ້ນ. ແຕ່ທໍາອິດຂ້າພະເຈົ້າຢາກຈະເອົາໃຈໃສ່ກັບຕົວກໍານົດການຫນຶ່ງທີ່ຈະສ້າງຄວາມສະດວກໃນການດີບັກທີ່ເປັນໄປໄດ້.

ນີ້ຈະຊ່ວຍໃຫ້ 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 ຜູ້ຜະລິດມີພາລາມິເຕີ acks, ເຊິ່ງອະນຸຍາດໃຫ້ທ່ານສາມາດກໍາຫນົດຄ່າຫຼັງຈາກຈໍານວນການຮັບຮູ້ທີ່ຜູ້ນໍາກຸ່ມຕ້ອງການພິຈາລະນາຂໍ້ຄວາມທີ່ຂຽນສົບຜົນສໍາເລັດ. ພາລາມິເຕີນີ້ສາມາດເອົາຄ່າຕໍ່ໄປນີ້:

• 0 — ການ​ຮັບ​ຮູ້​ຈະ​ບໍ່​ໄດ້​ຮັບ​ການ​ພິ​ຈາ​ລະ​ນາ​.
• 1 ເປັນພາລາມິເຕີເລີ່ມຕົ້ນ, ພຽງແຕ່ 1 replica ຈໍາເປັນຕ້ອງໄດ້ຮັບຮູ້.
• −1 — ຕ້ອງ​ມີ​ການ​ຮັບ​ຮູ້​ຈາກ​ການ​ຈໍາ​ລອງ​ທີ່ synchronized ທັງ​ຫມົດ (ການ​ຕັ້ງ​ຄ່າ​ກຸ່ມ min.insync.replicas).

ຈາກຄ່າທີ່ລະບຸໄວ້ມັນເປັນທີ່ຊັດເຈນວ່າ acks ເທົ່າກັບ −1 ໃຫ້ການຄໍ້າປະກັນທີ່ເຂັ້ມແຂງທີ່ສຸດວ່າຂໍ້ຄວາມຈະບໍ່ສູນເສຍ.

ດັ່ງທີ່ພວກເຮົາທຸກຄົນຮູ້, ລະບົບການແຈກຢາຍແມ່ນບໍ່ຫນ້າເຊື່ອຖື. ເພື່ອປົກປ້ອງຄວາມຜິດຊົ່ວຄາວ, Kafka Producer ໃຫ້ທາງເລືອກ ລອງໃ່, ເຊິ່ງອະນຸຍາດໃຫ້ທ່ານສາມາດກໍານົດຈໍານວນຂອງຄວາມພະຍາຍາມສົ່ງຄືນພາຍໃນ delivery.timeout.ms. ເນື່ອງຈາກພາຣາມິເຕີ retries ມີມູນຄ່າເລີ່ມຕົ້ນຂອງ Integer.MAX_VALUE (2147483647), ຈໍານວນການລອງຂໍ້ຄວາມສາມາດປັບໄດ້ໂດຍການປ່ຽນພຽງແຕ່ delivery.timeout.ms ເທົ່ານັ້ນ.

ພວກເຮົາກໍາລັງກ້າວໄປສູ່ການຈັດສົ່ງຢ່າງແນ່ນອນ

ການຕັ້ງຄ່າທີ່ລະບຸໄວ້ອະນຸຍາດໃຫ້ຜູ້ຜະລິດຂອງພວກເຮົາສົ່ງຂໍ້ຄວາມທີ່ມີການຄໍ້າປະກັນສູງ. ຕອນນີ້ໃຫ້ເວົ້າກ່ຽວກັບວິທີການຮັບປະກັນວ່າພຽງແຕ່ຫນຶ່ງສໍາເນົາຂອງຂໍ້ຄວາມຖືກຂຽນໃສ່ຫົວຂໍ້ Kafka? ໃນກໍລະນີງ່າຍດາຍທີ່ສຸດ, ເພື່ອເຮັດສິ່ງນີ້, ທ່ານຈໍາເປັນຕ້ອງໄດ້ກໍານົດພາລາມິເຕີກ່ຽວກັບຜູ້ຜະລິດ enable.idempotence ກັບຄວາມຈິງ. Idempotency ຮັບປະກັນວ່າຂໍ້ຄວາມດຽວຖືກຂຽນໃສ່ສ່ວນສະເພາະຂອງຫົວຂໍ້ຫນຶ່ງ. precondition ສໍາ​ລັບ​ການ​ເຮັດ​ໃຫ້ ideempotency ແມ່ນ​ຄ່າ​ acks = ທັງໝົດ, ລອງໃໝ່ > 0, max.in.flight.requests.per.connection ≤ 5. ຖ້າຕົວກໍານົດການເຫຼົ່ານີ້ບໍ່ຖືກກໍານົດໂດຍຜູ້ພັດທະນາ, ຄ່າຂ້າງເທິງຈະຖືກຕັ້ງຄ່າໂດຍອັດຕະໂນມັດ.

ເມື່ອ idempotency ຖືກກໍານົດ, ມັນເປັນສິ່ງຈໍາເປັນເພື່ອຮັບປະກັນວ່າຂໍ້ຄວາມດຽວກັນສິ້ນສຸດລົງໃນການແບ່ງປັນດຽວກັນທຸກໆຄັ້ງ. ອັນນີ້ສາມາດເຮັດໄດ້ໂດຍການຕັ້ງລະຫັດ partitioner.class ແລະພາລາມິເຕີເປັນ Producer. ໃຫ້ເລີ່ມຕົ້ນດ້ວຍກຸນແຈ. ມັນຕ້ອງຄືກັນສໍາລັບການຍື່ນສະເຫນີແຕ່ລະຄົນ. ນີ້ສາມາດເຮັດໄດ້ງ່າຍໂດຍການໃຊ້ ID ທຸລະກິດໃດໆຈາກການຕອບຕົ້ນສະບັບ. ພາຣາມິເຕີ partitioner.class ມີມູນຄ່າເລີ່ມຕົ້ນ − DefaultPartitioner. ດ້ວຍກົນລະຍຸດການແບ່ງປັນນີ້, ໂດຍຄ່າເລີ່ມຕົ້ນພວກເຮົາປະຕິບັດແບບນີ້:

• ຖ້າການແບ່ງປັນຖືກລະບຸຢ່າງຊັດເຈນໃນເວລາສົ່ງຂໍ້ຄວາມ, ຫຼັງຈາກນັ້ນພວກເຮົາໃຊ້ມັນ.
• ຖ້າການແບ່ງພາຕິຊັນບໍ່ໄດ້ຖືກລະບຸ, ແຕ່ກະແຈຖືກລະບຸ, ເລືອກພາທິຊັນໂດຍ hash ຂອງຄີ.
• ຖ້າການແບ່ງພາຕິຊັນແລະກະແຈບໍ່ໄດ້ຖືກລະບຸ, ເລືອກພາທິຊັນຫນຶ່ງໂດຍຫນຶ່ງ (round-robin).

ນອກຈາກນີ້, ການນໍາໃຊ້ການສົ່ງທີ່ສໍາຄັນແລະ ideempotent ກັບພາລາມິເຕີ max.in.flight.requests.per.connection = 1 ເຮັດ​ໃຫ້​ທ່ານ​ມີ​ການ​ປະ​ມວນ​ຜົນ​ຂໍ້​ຄວາມ​ສະ​ດວກ​ໃນ​ການ​ບໍ​ລິ​ໂພກ​. ມັນຍັງເປັນມູນຄ່າທີ່ຈື່ໄວ້ວ່າຖ້າການຄວບຄຸມການເຂົ້າເຖິງຖືກຕັ້ງຄ່າຢູ່ໃນກຸ່ມຂອງທ່ານ, ຫຼັງຈາກນັ້ນ, ທ່ານຈະຕ້ອງມີສິດທິໃນການຂຽນຫົວຂໍ້ໃດຫນຶ່ງທີ່ບໍ່ມີທ່າແຮງ.

ຖ້າທັນທີທັນໃດທ່ານຂາດຄວາມສາມາດຂອງການສົ່ງໂດຍກະແຈຫຼືເຫດຜົນໃນດ້ານຜູ້ຜະລິດຮຽກຮ້ອງໃຫ້ມີການຮັກສາຄວາມສອດຄ່ອງຂອງຂໍ້ມູນລະຫວ່າງພາທິຊັນທີ່ແຕກຕ່າງກັນ, ຫຼັງຈາກນັ້ນ, ການເຮັດທຸລະກໍາຈະມາຊ່ວຍ. ນອກຈາກນັ້ນ, ການນໍາໃຊ້ການເຮັດທຸລະກໍາລະບົບຕ່ອງໂສ້, ທ່ານສາມາດ synchronize ບັນທຶກໃນ Kafka, ສໍາລັບຕົວຢ່າງ, ດ້ວຍບັນທຶກໃນຖານຂໍ້ມູນ. ເພື່ອເປີດໃຊ້ທຸລະກໍາການສົ່ງກັບຜູ້ຜະລິດ, ມັນຕ້ອງມີຄວາມເຂັ້ມງວດແລະກໍານົດເພີ່ມເຕີມ transactional.id. ຖ້າກຸ່ມ Kafka ຂອງເຈົ້າມີການຄວບຄຸມການເຂົ້າເຖິງທີ່ຖືກຕັ້ງຄ່າ, ຫຼັງຈາກນັ້ນບັນທຶກການເຮັດທຸລະກໍາ, ເຊັ່ນບັນທຶກ idempotent, ຈະຕ້ອງການການອະນຸຍາດການຂຽນ, ເຊິ່ງສາມາດໄດ້ຮັບໂດຍຫນ້າກາກໂດຍໃຊ້ມູນຄ່າທີ່ເກັບໄວ້ໃນ transactional.id.

ຢ່າງເປັນທາງການ, ສະຕຣິງໃດນຶ່ງ, ເຊັ່ນຊື່ແອັບພລິເຄຊັນ, ສາມາດໃຊ້ເປັນຕົວລະບຸການເຮັດທຸລະກໍາໄດ້. ແຕ່ຖ້າທ່ານເປີດຫຼາຍໆຕົວຢ່າງຂອງແອັບພລິເຄຊັນດຽວກັນກັບ transactional.id ດຽວກັນ, ຕົວຢ່າງທີ່ເປີດຕົວຄັ້ງທໍາອິດຈະຖືກຢຸດເຊົາດ້ວຍຄວາມຜິດພາດ, ເພາະວ່າ Kafka ຈະພິຈາລະນາມັນເປັນຂະບວນການ zombie.

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 ຄ່າ read_committed. ຜູ້ບໍລິໂພກດັ່ງກ່າວຈະສາມາດອ່ານຂໍ້ຄວາມທີ່ບໍ່ແມ່ນທຸລະກໍາໄດ້ຄືກັບກ່ອນ, ແລະຂໍ້ຄວາມການເຮັດທຸລະກໍາພຽງແຕ່ຫຼັງຈາກຄໍາຫມັ້ນສັນຍາ.
ຖ້າທ່ານໄດ້ກໍານົດການຕັ້ງຄ່າທັງຫມົດທີ່ລະບຸໄວ້ກ່ອນຫນ້ານີ້, ຫຼັງຈາກນັ້ນທ່ານໄດ້ກໍານົດການຈັດສົ່ງຢ່າງແນ່ນອນ. ຊົມເຊີຍ!

ແຕ່ມີຫນຶ່ງ 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

ເພື່ອຫຼຸດຜ່ອນຄວາມຜິດພາດໃນຄໍາຮ້ອງສະຫມັກໃນອະນາຄົດ, ພວກເຮົາເຮັດ wrapper ຂອງພວກເຮົາເອງໃນໄລຍະການຕັ້ງຄ່າພາກຮຽນ spring, ບ່ອນທີ່ຄ່າສໍາລັບບາງຕົວກໍານົດການທີ່ໄດ້ລະບຸໄວ້ແລ້ວ.

ນີ້ແມ່ນເອກະສານສອງສາມຢ່າງສໍາລັບການສຶກສາດ້ວຍຕົນເອງ:

• KIP-98 - ແທ້ຈິງແລ້ວເມື່ອການຈັດສົ່ງແລະການສົ່ງຂໍ້ຄວາມ
• ລາຍລະອຽດຂອງການຕັ້ງຄ່າ

ແຫຼ່ງຂໍ້ມູນ: www.habr.com