کافکا، توسعه پذیری و پردازش موازی

خلاصه بر کافکا

کافکا به طور خلاصه یک صف است که پیام از یک‌ طرف وارد (produce) و از طرف دیگر استفاده (consume) می‌شود. از دلایل وجود کافکا نیاز به پلتفرمی برای پردازش real-time داده‌ها با کارایی و شاخص اطمینان بالا (reliability)، قابلیت توسعه‌پذیری‌ (‌scalability) و در نهایت پردازش موازی است.

زمانی که نیاز به پردازش و عملیات روی پیام‌هایی (log) دارید که مهم هستند و هیچ‌کدام نباید از دست بروند، کافکا به کارتان خواهد آمد. کافکا تضمین خواهد کرد که هیچ‌کدام از پیام‌های شما از دست نخواهد رفت. حتی اگر تا میلیون پیام در ثانیه برای پردازش داشته باشید!

چه زمانی نیاز به پردازش موازی پیام‌ها داریم؟

در کاربردهای بسیاری رخ می‌دهد که عملیات پردازش روی پیام‌ها وقت‌گیر و یا سنگین هستند. در این کاربردها به طبع تمایل داریم به‌صورت موازی پردازش‌ را انجام دهیم ولی نیاز به کسی داریم که این تقسیم کار را به طور صحیح و قابل اطمینانی انجام دهد. اگر یکی از workerها دچار مشکل شد و یا پردازشی با خطا مواجه شد این وضعیت را مدیریت کند.

برای بررسی توسعه‌پذیری و پردازش موازی کافکا نیاز به بررسی دقیق‌تر ساختار کافکا داریم.

مفاهیم consumer group ،partition ،topic و consumer

• مفهوم topic و partition

همانطور که احتمالا از پیش می‌دانید topic در کافکا نامی است که داده ها در آن ذخیره میشوند.

پارتیشن (partition) در کافکا در واقع تکه‌کردن یک topic به قسمت‌های مختلف است به‌طوری‌که partitionهای مختلف می‌توانند در nodeهای متفاوت از یک cluster کافکا قرار گیرند. این قابلیت، توسعه‌پذیری را به کافکا می‌بخشد.

• مفهوم consumer‌ و consumer group و offset

همان‌طور که مشخص است consumer پیام‌ها را از topic می‌خواند و بسته به نیاز عملیاتی انجام می‌دهد. با هر بار خوانده‌شدن پیام توسط یک consumer‌ مقدار offset زیاد می‌شود که درواقع اعلام می‌کند که برای مصرف پیام بعدی از کدام پیام صف استفاده شود. هر consumer آفست خود را دارد و مکرراً این آفست را به کافکا نیز می‌فرستد. این آفست‌ها در یکی از topicهای کافکا خودبه‌خود ذخیره می‌شوند. به‌این‌ترتیب اگر consumer به دلایلی از کار افتاد، پیام‌های کاملاً مصرف نشده و مصرف شده مشخص هستند.

مفهوم consumer‌ جزئی از مفهوم consumer group است. کافکا امکان استفاده از چندین consumer روی یک topic را می‌دهد. هر consumer‌ به یک یا چند partition‌ از topic کافکا متصل می‌شود؛ اما چند consumer به یک partition نمی‌توانند متصل شوند. می‌توان به طور دستی نحوه اتصال consumerها به partitionها را تعیین نمود؛ اما به طور پیش‌فرض اگر تعداد consumerها از تعداد partitionها کمتر باشد، برخی از consumerها به بیش از یک partition متصل می‌شوند و از همه آن‌ها پیام دریافت می‌کنند.

اگر تعداد consumerها و partitionها برابر باشد‌‌، هر consumer به یک partition ‌متصل می‌شود و اگر تعداد ‌‌‌consumerها بیشتر باشد، ‌consumerهای اضافی بیکار (در حالت رزرو یا پشتیبان) می‌مانند.

در هر یک حالات بالا در صورت ازکارافتادن یکی از consumerها، سایر consumerها به partition رها شده متصل می‌شوند و پیام‌های آن را دریافت میکنند. همانطور که گفته شد کافکا offset هر partition را در یک topic جداگانه ذخیره می‌کند. لذا در صورت از کار افتادن یک consumer و جایگزین شدن consumer دیگر، اختلالی در روند دریافت پیام‌ها ایجاد نمی‌شود. لازم به ذکر است که خود کافکا اضافه شدن و یا کم شد consumerها را مدیریت می‌کند.

حال نگاهی عمیق‌تر به مفهوم consumer group می‌اندازیم. دیدیم که در یک گروه همه consumerها با هم عمل می‌کنند و در صورت ازکارافتادن یکی از آن‌ها دیگری کار قبلی را ادامه می‌دهد. این بدان معنی است که یک consumer group یک application است که تضمین می‌شود تمام پیام‌ها را دریافت می‌کند. اما اگر نیاز داشته باشیم که در application دیگری به طور موازی هم به تمام این پیام‌‌ها دسترسی داشته باشیم چه باید بکنیم؟ خیلی ساده از یک consumer group دیگر استفاده می‌کنیم. گروه‌های مختلف در کافکا مستقل از هم عمل می‌کنند و هر گروه مستقلاً تمام پیام‌های topic را دریافت می‌کند.

(یادآوری این مورد نیز خالی از لطف نیست که یک partition نمی‌تواند به بیش از یک consumer در یک گروه متصل شود.)

مفاهیم poll و heartbeat

مفاهیم poll و heartbeat دو مفهوم مربوط به consume کردن از topic هستند که در سناریوهای پردازش موازی و multi-consumer اهمیت بیشتری پیدا می‌کنند. به‌ویژه زمانی که پردازش و یا کارهای زمان‌بری با هر پیام انجام می‌شود.

• مفهوم heartbeat:

هر consumer یک ضربان با خود کافکا (broker) دارد. وجود این ضربان ضروری است چرا که کافکا برای تخصیص partitionها به consumerها نیاز به دانستن تعداد و وضعیت consumerها دارد. برای مثال اگر یک consumer فعال به هر دلیلی غیرفعال شود و از رده‌ خارج شود، کافکا consumer دیگری را جایگزین consumer فعلی می‌کند. یکی از پارامترهایی که از طریق آن فعال‌بودن یک consumer برای کافکا احراز می‌شود، heartbeat است. به این طریق که هر consumer هر پنج ثانیه (به طور پیش‌فرض در اکثر موارد) یک ضربان با کافکا (broker) خواهد داشت و از این طریق سلامت آن مشخص می‌شود. البته heartbeat تنها پارامتری نیست که ممکن است یک consumer را از دید کافکا غیرفعال کند؛ اما در بسیاری از کاربردها مهم‌ترین پارامتر است. به یکی دیگر از این پارامترها در ادامه متن اشاره خواهد شد.

• مفهوم poll:

به تابعی که یک consumer اجرا می‌کند تا پیام‌ها را از ‌topic بگیرد، poll گفته می‌شود. این تابع یک ورودی دارد که ماکزیمم زمانی است که consumer برای پیام جدید صبر می‌کند. در حالت ساده poll یک حلقه while با شرط کمتر بودن زمان انتظار از ماکزیمم زمان انتظار است. لازم به ذکر است کهconsumerها معمولاً این عمل را به‌صورت batch انجام می‌دهند، بدین معنی که با هر بار درخواست از کافکا (broker) چند پیام را به‌صورت هم‌زمان (در صورت وجود) دریافت می‌کنند.

تا اینجا با مفاهیم موردنیاز استفاده از کافکا آشنا شده‌ایم و از روش کافکا در مدیریت consumer groupها نیز اطلاع نسبی داریم.

همانند هر ابزاری، کافکا و consumerهایش تنظیماتی دارند که برای استفاده بهتر از آن ابزار، درک و استفاده بهینه از این تنظیمات ضروری است. در ادامه تنها به دو مورد از این تنظیمات مربوط به consumer که برای استفاده از کافکا برای پردازش موازی نسبتاً مهم‌تر هستند اشاره خواهد شد.

• مفاهیم max.poll.records و max.poll.interval.ms:

مفهوم max.poll.interval.ms همان‌طور که از نامش پیداست، به حداکثر زمان مجاز بین دو poll متوالی گفته می‌شود. اگر یک consumer پس از انجام poll، تا حداکثر max.poll.interval.ms میلی‌ثانیه poll نکند، از نظر کافکا این consumer مرده (dead) محسوب می‌شود و کافکا partitionهایش را بین سایر consumerها rebalance می‌کند. مفهوم max.poll.records هم به همان batch بودن فرایند گرفتن پیام‌ها از کافکا اشاره دارد. این دو تنظیم در همه consumerهای موجود و نوشته شده برای کافکا به همین صورت (و یا شاید با اندکی تغییر) هستند.

دو تنظیم بالا معمولاً با هم مهم هستند. فرض کنید که یک task برای انجام به ۱۰ ثانیه زمان نیاز دارد، max.poll.records و max_poll_interval نیز به ترتیب ۵۰۰ و ۵ دقیقه (۳۰۰ ثانیه) هستند. در این حالت زمان بین دو poll یک consumer مقدار ۵۰۰ ثانیه خواهد بود. به طبع در میانه پردازش پیام‌ها کافکا این consumer را از consumer group بیرون می‌اندازد.

مورد ذکر شده در بالا یکی از موارد مهم تنظیمات کافکا و consumerهایش بود که اشاره شد و هدف اصلی آن، علاوه بر ارتباط آن با موضوع مقاله یعنی پردازش موازی، اشاره به اهمیت دقت به تنظیمات یک ابزار بوده است.

در این مقاله مفاهیم اولیه استفاده از کافکا و بخشی از کاربردهای آن بیان شد. کافکا ویژگی‌ها و کاربردها متعدد دیگری نیز دارد که بیشتر آن‌ها در ذیل مفاهیم مطرح شده قرار می‌گیرند و بسته به کاربرد موردنیاز می‌توان از کافکا کمک گرفت.