حالت تاریک
فصل ۴ — Producerها: منبع داده
این فصل شامل:
- ارسال پیامها و producer
- ساخت serializer و partitioner سفارشی خودمان
- بررسی گزینههای پیکربندی برای برآورده کردن نیازمندیهای یک شرکت
در فصل قبل، نیازمندیهایی را که یک سازمان ممکن است دربارهٔ دادههایش داشته باشد بررسی کردیم. برخی تصمیمهای طراحی که گرفتیم، تأثیرات عملی بر نحوهٔ ارسال داده به Kafka دارند. حالا از دروازهٔ یک Kafka producer وارد دنیای پلتفرم event-streaming میشویم. پس از خواندن این فصل، برای حل نیازمندیهای بنیادین یک پروژهٔ Kafka با تولید داده به چند روش مختلف، در مسیر درستی خواهید بود.
Producer، با وجود اهمیتش، فقط بخشی از این سیستم است. در واقع میتوانیم برخی گزینههای پیکربندی producer را تغییر دهیم یا آنها را در سطح broker یا topic تنظیم کنیم. این گزینهها را با پیشرفت در کتاب کشف خواهیم کرد، اما در این فصل نخستین نگرانی ما ورود داده به Kafka است.
۴.۱ یک مثال
Producer راهی برای push کردن داده به سیستم Kafka در پروژهٔ نمونهٔ ما فراهم میکند. بهعنوان یادآوری، شکل ۴-۱ نشان میدهد producerها در Kafka کجا قرار میگیرند.
شکل ۴-۱. Kafka producerها
Producer clients JVM application-message broker We will focus Examples Data in Kafka core on our alert • Databases (to partition) producers in • IOT events OS page cache (memory) this chapter. • Browser/user web events • Logs Flushed to disk Messages can be replayed Topic from the beginning of the Replay log and consumed again. Partition 0 0 1 2 3 4 Consumer clients Examples Data out • Logs are append only. • HDFS (from partition) • New entries added to the end. • S3 • No database storage, just disk. • Web applications • Each log is made up of entries • Metrics labeled with offset numbers. • Analytics engines ZooKeeper ensemble Message brokers (cluster) + Expanded ZooKeeper used for One of the brokers distributed configuration will be a controller. and management
با نگاه به شکل ۴-۱، روی گوشهٔ بالا-چپ (producer clientها) تمرکز کنیم که نمونههایی از دادهٔ تولیدشده در Kafka را نشان میدهد. این داده میتواند همان رویدادهای IoT باشد که در شرکت داستانیمان استفاده میکنیم. برای ملموستر کردن ایدهٔ تولید داده، تصور کنیم مثال عملیای نوشتهایم که ممکن است برای یکی از پروژههایمان داشته باشیم. به برنامهای نگاه کنیم که بازخورد کاربران دربارهٔ عملکرد وبسایت برای مشتریان را میگیرد.
در حال حاضر، کاربر فرمی در وبسایت ارسال میکند که ایمیلی به حساب پشتیبانی یا chatbot تولید میکند. هر از گاهی یکی از کارکنان پشتیبانی صندوق ورودی را بررسی میکند تا ببیند مشتریان چه پیشنهادها یا مشکلاتی داشتهاند. با نگاه به آینده، میخواهیم این اطلاعات همچنان به ما برسد، اما به روشی که داده از صندوق ورودی ایمیل قابلدسترستر باشد. اگر بهجای آن این پیام را به یک topic در Kafka بفرستیم، میتوانیم پاسخهای قویتر و متنوعتری تولید کنیم، نه فقط پاسخهای واکنشی ایمیل به مشتریان. انعطافپذیری از داشتن رویداد در Kafka برای استفادهٔ هر برنامهٔ مصرفکننده میآید.
ابتدا ببینیم استفاده از ایمیل بهعنوان بخشی از خط لولهٔ داده چه تأثیری دارد. با نگاه به شکل ۴-۲، تمرکز روی قالبی که داده پس از ارسال فرم بازخورد کاربر ذخیره میشود مفید است.
شکل ۴-۲. ارسال داده در ایمیل
HTML form User: Issue: User submits feedback form User Application generates email Data format SMTP protocol Delivered to: Data stored in mail server Received like Microsoft Exchange by: MIME version: User access of data with email client or ... Manual data extraction Data extraction with scripts with copy and paste
ایمیل سنتی از Simple Mail Transfer Protocol (SMTP) استفاده میکند و این در نحوهٔ ارائه و گاهی ذخیرهٔ خود رویداد ایمیل منعکس میشود. میتوانیم از کلاینتهای ایمیل مانند Microsoft® Outlook® برای بازیابی سریع داده استفاده کنیم، اما بهجای فقط خواندن ایمیل، چگونه میتوانیم داده را از آن سیستم برای کاربردهای دیگر بیرون بکشیم؟ کپی و paste گامهای دستی رایجی هستند، همچنین اسکریپتهای parse کردن ایمیل. (اسکریپتهای parse شامل استفاده از ابزار یا زبان برنامهنویسی و کتابخانهها یا فریمورکها برای درست کردن parse است.) در مقایسه، اگرچه Kafka از پروتکل خودش استفاده میکند، هیچ قالب خاصی برای دادهٔ پیام ما تحمیل نمیکند. باید بتوانیم داده را در هر قالبی که انتخاب کنیم بنویسیم.
یادداشت در فصل قبل، قالب Apache Avro را بهعنوان یکی از قالبهای رایج جامعهٔ Kafka بررسی کردیم. Protobuf و JSON هم بسیار محبوباند [1].
الگوی استفادهٔ دیگری که به ذهن میرسد این است که اعلانهای مشکلات مشتری یا قطعی وبسایت را بهعنوان هشدارهای موقت در نظر بگیریم که پس از پاسخ به مشتری حذف میشوند. اما این ورودی مشتری ممکن است بیش از یک هدف داشته باشد. اگر بتوانیم روند قطعیهایی که مشتریان گزارش میکنند را ببینیم چه؟ آیا سایت همیشه پس از ارسال کدهای کوپن فروش در ایمیلهای بازاریابی انبوه کند میشود؟ آیا این داده میتواند به ما کمک کند ویژگیهایی را که کاربران از سایت ما کم دارند پیدا کنیم؟ آیا ۴۰٪ ایمیلهای کاربران ما دربارهٔ مشکل یافتن تنظیمات Privacy حسابشان است؟ داشتن این داده در topicای که قابل replay یا خواندن توسط چند برنامه با اهداف مختلف باشد، ارزش بیشتری برای مشتری نسبت به ایمیل پشتیبانی یا bot خودکار که سپس حذف میشود ایجاد میکند.
همچنین اگر نیاز retention داریم، آنها توسط تیمهایی که زیرساخت ایمیل ما را اجرا میکنند کنترل میشود، در مقابل تنظیم پیکربندی که با Kafka کنترل میکنیم. با نگاه دوباره به شکل ۴-۳، توجه کنید که برنامه فرم HTML دارد اما به topic در Kafka مینویسد، نه به سرور ایمیل.
شکل ۴-۳. ارسال داده به Kafka
HTML form User: Issue: User submits feedback form User Format Application produces determined message to Kafka by you. Data format Kafka protocol { User: Timestamp: Data stored in Kafka brokers Issue: } Access with Access with ksqlDB Access with Connect custom consumer
با این رویکرد میتوانیم اطلاعات مهم را در هر قالبی که نیاز داریم استخراج کنیم و به روشهای مختلف استفاده شود. برنامههای مصرفکننده میتوانند با schemaها روی داده کار کنند و به یک قالب پروتکل واحد وابسته نباشند. میتوانیم این پیامها را برای use caseهای جدید نگه داریم و دوباره پردازش کنیم چون retention رویدادها را کنترل میکنیم. حالا که دیدیم چرا ممکن است از producer استفاده کنیم، بیایید جزئیات تعامل producer با brokerهای Kafka را سریع بررسی کنیم.
۴.۱.۱ نکات producer
وظیفهٔ producer شامل واکشی metadata دربارهٔ کلاستر است [2]. چون producerها فقط میتوانند به replica leader پارتیشنی که به آن اختصاص یافتهاند بنویسند، metadata به producer کمک میکند تعیین کند به کدام broker بنویسد، چون کاربر ممکن است فقط نام topic را بدون جزئیات دیگر داده باشد. این خوب است چون کاربر نهایی producer مجبور نیست فراخوانی جداگانهای برای گرفتن آن اطلاعات انجام دهد. با این حال، کاربر نهایی باید حداقل یک broker در حال اجرا برای اتصال داشته باشد و کتابخانهٔ کلاینت Java بقیه را خودش پیدا میکند.
چون این سیستم توزیعشده برای خطاهای لحظهای مانند قطعی شبکه طراحی شده، منطق retry از قبل داخلی است. اما اگر ترتیب پیامها ضروری است، مانند پیامهای audit ما، علاوه بر تنظیم retries روی عددی مثل ۳، باید مقدار max.in.flight.requests.per.connection را روی ۱ و acks (تعداد brokerهایی که تأیید برمیگردانند) را روی all بگذاریم [3] [4]. بهنظر ما این یکی از امنترین روشها برای اطمینان از رسیدن پیامهای producer به ترتیبی است که قصد داریم [4]. میتوانیم مقادیر acks و retries را بهعنوان پارامترهای پیکربندی تنظیم کنیم.
گزینهٔ دیگری که باید بدانید استفاده از idempotent producer است. اصطلاح idempotent به این معناست که ارسال همان پیام چند بار فقط یکبار تولید پیام را نتیجه میدهد. برای استفاده از idempotent producer میتوانیم enable.idempotence=true را تنظیم کنیم [5]. در مثالهای بعدی از idempotent producer استفاده نمیکنیم.
یک چیز که لازم نیست نگرانش باشیم این است که یک producer در راه دادهٔ producer دیگر قرار بگیرد. thread safety مشکل نیست چون داده بازنویسی نمیشود، بلکه توسط خود broker مدیریت و به log broker اضافه (append) میشود [6]. حالا وقت آن است ببینیم چگونه مقادیری مانند max.in.flight.requests.per.connection را در کد فعال کنیم.
۴.۲ گزینههای producer
یکی از چیزهای جالب وقتی شروع به کار با ارسال داده به Kafka کردیم، سادگی تنظیم گزینهها با کلاینتهای Java بود که در این کتاب روی آنها تمرکز میکنیم. اگر با سیستمهای صف یا پیامرسانی دیگر کار کردهاید، راهاندازی آن سیستمها میتواند شامل ارائهٔ لیست صفهای remote و local، hostnameهای manager، شروع اتصالها، connection factoryها، sessionها و موارد بیشتر باشد.
اگرچه کاملاً بدون دردسر نیست، producer از پیکربندی خودش برای واکشی بخش زیادی از اطلاعاتی که نیاز دارد، مانند لیست همهٔ brokerهای Kafka، کار میکند. با استفاده از مقدار bootstrap.servers بهعنوان نقطهٔ شروع، producer metadata دربارهٔ brokerها و پارتیشنها را برای همهٔ نوشتنهای بعدی واکشی میکند.
همانطور که گفتیم، Kafka اجازه میدهد با تغییر برخی مقادیر پیکربندی، رفتارهای کلیدی را عوض کنید. یکی از راههای برخورد با همهٔ نامهای کلید پیکربندی producer، استفاده از ثابتهای کلاس ProducerConfig هنگام توسعهٔ کد producer است (به http://mng.bz/ZYdA مراجعه کنید) و جستجوی برچسب Importance با مقدار «high» در وبسایت Confluent [7]. اما در مثالهایمان برای وضوح از خود نام propertyها استفاده میکنیم.
جدول ۴-۱ برخی از مهمترین پیکربندیهای producer را که مثالهای خاص ما را پشتیبانی میکنند فهرست میکند. در بخشهای بعد میبینیم برای تکمیل کار کارخانهمان چه چیزی لازم است.
جدول ۴-۱. پیکربندیهای مهم producer
| کلید | هدف |
|---|---|
acks | تعداد تأیید replicaهایی که producer قبل از برقراری موفقیت نیاز دارد |
bootstrap.servers | یک یا چند broker Kafka برای اتصال در راهاندازی |
value.serializer | کلاسی که برای serialization مقدار استفاده میشود |
key.serializer | کلاسی که برای serialization کلید استفاده میشود |
۴.۲.۱ پیکربندی لیست broker
از مثالهای نوشتن پیام به Kafka مشخص است که باید به producer بگوییم پیامها را به کدام topic بفرستد. بهخاطر بیاورید topicها از پارتیشنها ساخته میشوند، اما Kafka چگونه میداند پارتیشن topic کجاست؟ ما لازم نیست جزئیات آن پارتیشنها را هنگام ارسال پیام بدانیم. شاید یک تصویر این معما را روشن کند. یکی از گزینههای پیکربندی لازم برای producerها bootstrap.servers است. شکل ۴-۴ مثالی از producerای را نشان میدهد که فقط broker ۰ در لیست bootstrap serverهایش دارد، اما با شروع از یکی میتواند همهٔ سه broker کلاستر را یاد بگیرد.
شکل ۴-۴. Bootstrap serverها
1. Producer connects to bootstrap servers Broker 0 Broker 1 Broker 2 Producer Controller Our alert producers connect to our servers since they are local on 2. Metadata sent back to producer letting different ports on localhost. it know its leader resides on Broker 2, which it did not know about at first. Kafka knows about its other brokers.
bootstrap.servers میتواند چند broker اولیه یا فقط یکی مانند شکل ۴-۴ بگیرد. با اتصال به این broker، کلاینت metadata مورد نیازش را کشف میکند که شامل داده دربارهٔ brokerهای دیگر در کلاستر هم میشود [8].
این پیکربندی کلیدی است تا producer brokerای برای صحبت پیدا کند. پس از اتصال به کلاستر، میتواند metadata لازم برای جزئیاتی که قبلاً ندادیم (مانند محل replica leader پارتیشن روی دیسک) را بگیرد. کلاینتهای producer هم میتوانند با اطلاعات کلاستر، شکست leader پارتیشنی که به آن مینویسند را جبران کنند و leader جدید پیدا کنند. شاید متوجه شده باشید اطلاعات ZooKeeper بخشی از پیکربندی نیست. هر metadataای که producer نیاز دارد بدون اینکه کلاینت producer جزئیات کلاستر ZooKeeper را بدهد مدیریت میشود.
۴.۲.۲ چگونه سریع برویم (یا امنتر)
الگوهای پیام ناهمگام یکی از دلایلی است که بسیاری از سیستمهای صفگونه استفاده میکنند و این قابلیت قدرتمند در Kafka هم هست. میتوانیم در کد منتظر نتیجهٔ درخواست send producer بمانیم، یا موفقیت یا شکست را بهصورت ناهمگام با callbackها یا اشیاء Future مدیریت کنیم. اگر بخواهیم سریعتر برویم و منتظر پاسخ نمانیم، هنوز میتوانیم نتایج را بعداً با منطق سفارشی خودمان مدیریت کنیم.
property پیکربندی دیگری که به سناریوی ما میخورد کلید acks است که مخفف acknowledgments است. این کنترل میکند producer قبل از برگرداندن درخواست تکمیلشده چند تأیید از followerهای leader پارتیشن دریافت کند. مقادیر معتبر این property عبارتاند از: all، -1، 1 و 0 [9].
شکل ۴-۵ نشان میدهد پیامی با acks برابر ۰ چگونه رفتار میکند. تنظیم این مقدار روی ۰ احتمالاً کمترین latency را میدهد اما به قیمت امنیت. علاوه بر این، تضمینی داده نمیشود که broker پیام را دریافت کرده و retry هم انجام نمیشود [9]. بهعنوان use case نمونه، فرض کنید پلتفرم ردیابی وب داریم که کلیکهای صفحه را جمع میکند و این رویدادها را به Kafka میفرستد. در این موقعیت، از دست رفتن یک کلیک یا hover شاید مهم نباشد. اگر یکی گم شود، تأثیر تجاری واقعی ندارد.
شکل ۴-۵. property
acksبرابر ۰1. The producer writes 2. The leader doesn't wait to the leader of the to find out if the write partition. was successful. This is not what we want for our kinaction_audit producer. 3. Because we do not know if the leader write was successful, we are not aware of the state of any replica copies and if they were successful or not.
در اصل، رویداد در شکل ۴-۵ از producer فرستاده شد و فراموش شد. ممکن است پیام هرگز به پارتیشن نرسیده باشد. اگر بهاتفاق به replica leader رسید، producer نمیداند آیا کپیهای replica follower موفق بودهاند یا نه.
تنظیمی که برعکس مورد قبلی میدانیم acks با مقادیر all یا -1 است. مقادیر all یا -1 قویترین گزینهٔ موجود برای این تنظیم هستند. شکل ۴-۶ نشان میدهد مقدار all یعنی replica leader پارتیشن منتظر میماند تا کل لیست in-sync replicaها (ISRها) تکمیل را تأیید کنند [9]. بهعبارت دیگر، producer تا زمانی که همهٔ replicaهای یک پارتیشن موفق نشوند، تأیید موفقیت نمیگیرد. آسان است ببینیم سریعترین نیست بهخاطر وابستگی به brokerهای دیگر. در بسیاری موارد، ارزش پرداخت هزینهٔ عملکرد را دارد تا از از دست رفتن داده جلوگیری شود. با brokerهای زیاد در کلاستر، باید بدانیم leader منتظر چند broker است. brokerی که دیرترین پاسخ را میدهد، عامل تعیینکنندهٔ زمان دریافت پیام موفقیت توسط producer است.
شکل ۴-۶. property
acksبرابرall1. The producer writes to the leader of the partition. We use acks=all for our Leader broker kinaction_audit producer. 2. The leader waits for all brokers to reply with success or failure. 3. The producer receives notification when all of the replicas are updated.
شکل ۴-۷ تأثیر تنظیم مقدار acks روی ۱ و درخواست acknowledgment را نشان میدهد. acknowledgment یعنی گیرندهٔ پیام (replica leader پارتیشن مشخص) تأیید را به producer برمیگرداند. کلاینت producer منتظر آن acknowledgment میماند. اما followerها ممکن است قبل از شکست leader، پیام را کپی نکرده باشند. اگر آن وضعیت قبل از کپی رخ دهد، پیام هرگز روی replica followerهای آن پارتیشن ظاهر نمیشود [9]. شکل ۴-۷ نشان میدهد در حالی که پیام توسط replica leader تأیید و به producer فرستاده شد، شکست هر replica در کپی پیام باعث میشود انگار پیام هرگز به کلاستر نرسیده است.
شکل ۴-۷. property
acksبرابر ۱1. The producer writes to the leader of the partition. Our other producers might use this setting. 2. The partition leader replies that the message has made a successful call. 3. Before the leader that has success has time to copy the message to any follower replicas, the leader fails. This 4. These brokers never see means that the message could be the message even though lost to the remaining brokers. it was seen by the leader.
یادداشت این ارتباط نزدیکی با ایدههای semantics حداکثر یکبار و حداقل یکبار دارد که در فصل ۱ پوشش دادیم [10]. تنظیم
acksبخشی از آن تصویر بزرگتر است.
۴.۲.۳ Timestampها
نسخههای اخیر رکورد producer شامل timestamp روی رویدادهایی است که میفرستید. کاربر میتواند زمان را هنگام ارسال شیء ProducerRecord بهعنوان نوع Java long به constructor بدهد، یا زمان فعلی سیستم. زمان واقعی استفادهشده در پیام میتواند همان مقدار بماند، یا میتواند timestamp broker باشد که هنگام log شدن پیام رخ میدهد. تنظیم message.timestamp.type topic روی CreateTime از زمان تنظیمشده توسط کلاینت استفاده میکند، در حالی که LogAppendTime زمان broker را استفاده میکند [11].
چرا یکی را بر دیگری ترجیح دهید؟ شاید بخواهید از زمان ایجاد استفاده کنید تا زمان وقوع تراکنش (مثل سفارش فروش) را داشته باشید، نه زمان رسیدن به broker. زمان broker وقتی مفید است که زمان ایجاد داخل خود پیام مدیریت میشود یا زمان رویداد واقعی از نظر کسبوکار یا سفارش مرتبط نیست.
همانطور که همیشه، timestampها میتوانند پیچیده باشند. مثلاً ممکن است رکوردی با timestamp زودتر از رکورد قبلی بگیریم. این در مواردی رخ میدهد که شکستی افتاده و پیام دیگری با timestamp دیرتر قبل از retry رکورد اول commit شده باشد. داده در log بر اساس offset مرتب میشود، نه timestamp. اگرچه خواندن دادهٔ timestampدار اغلب نگرانی کلاینت consumer است، نگرانی producer هم هست چون producer اولین گامها را برای اطمینان از ترتیب پیام برمیدارد.
همانطور که گفتیم، به همین دلیل max.in.flight.requests.per.connection هنگام تصمیم برای اجازهٔ retry یا چند درخواست inflight همزمان مهم است. اگر retry رخ دهد و درخواستهای دیگر در اولین تلاش موفق شوند، پیامهای زودتر ممکن است بعد از دیرترها به log اضافه شوند. شکل ۴-۸ مثالی از زمانی که پیام از ترتیب خارج میشود میدهد. اگرچه پیام ۱ اول فرستاده شد، بهخاطر فعال بودن retryها بهصورت مرتب به log نمیرسد.
شکل ۴-۸. تأثیر retry روی ترتیب
Producer Message 1 Leader partition broker 1. Message 1 is sent and fails. Message 2 2. Message 2 is sent and succeeds. Message 1 retry 3. Message 1 is resent and appended to the leader log after message 2. We used this retry logic to determine if we need ordering with our alert events from chapter 3 and for topic kinaction_audit.
گزینهٔ دیگر هنگام استفاده از producer، ساخت producer interceptorها است. اینها در KIP-42 (Kafka Improvement Proposal) معرفی شدند. هدف اصلی آن پشتیبانی از اندازهگیری و مانیتورینگ بود [12]. در مقایسه با workflow Kafka Streams برای فیلتر یا تجمیع داده، یا حتی ساخت topicهای مختلف مخصوص دادهٔ تغییریافته، استفاده از این interceptorها شاید اولین انتخاب ما نباشد. در حال حاضر interceptor پیشفرضی در چرخهٔ حیات اجرا نمیشود. در فصل ۹ use caseای برای trace کردن پیامها از کلاینتهای producer تا consumer با interceptorهایی که trace ID اضافه میکنند نشان میدهیم.
۴.۳ تولید کد برای نیازمندیهایمان
بیایید اطلاعاتی که دربارهٔ نحوهٔ کار producerها جمع کردیم روی راهحلهای خودمان بهکار ببریم. با چکلیست auditای که در فصل ۳ برای استفاده با Kafka در کارخانهٔ e-bike طراحی کردیم شروع میکنیم. همانطور که در فصل ۳ گفتیم، میخواهیم هیچ پیام auditای هنگام تکمیل دستورات اپراتورها روی سنسورها از دست نرود. یکی از نیازمندیها این بود که نیازی به همبستگی (یا گروهبندی) رویدادها نیست. نیازمندی دیگر اطمینان از عدم از دست رفتن پیامها بود. لیستینگ ۴-۱ نشان میدهد چگونه پیکربندی producer را شروع کنیم و با تنظیم acks روی all برای acknowledgment پیام امن باشیم.
لیستینگ ۴-۱. پیکربندی audit producer
java
public class AuditProducer {
// Creates properties as before
// for our configuration
...
private static final Logger log = LoggerFactory.getLogger
(AuditProducer.class);Properties kaProperties = new Properties();
kaProperties.put( "bootstrap.servers",
"localhost:9092,localhost:9093,localhost:9094"); // Sets acks to all to get
kaProperties.put("acks", "all"); // the strongest guarantee
kaProperties.put("retries", "3");
kaProperties.put("max.in.flight.requests.per.connection", "1");
...
// Lets the client retry in case of
// failure so we don't have to
// implement our own failure logicتوجه کنید برای نگرانی از دست رفتن پیام، جزئیات پیکربندی که به producer میفرستیم را تغییر دادیم. تغییر پیکربندی acks ویژگی کوچک اما قدرتمندی است که تأثیر قابلتوجهی روی رسیدن یا نرسیدن پیام دارد. چون نیازی به همبستگی (گروهبندی) رویدادها نداریم، برای این پیامها از کلید استفاده نمیکنیم. اما بخش بنیادینی هست که میخواهیم تغییر دهیم تا قبل از ادامه منتظر نتیجه بمانیم. لیستینگ بعدی متد get را نشان میدهد که چگونه میتوانیم منتظر تکمیل پاسخ بهصورت همگام قبل از ادامهٔ کد بمانیم. توجه کنید این لیستینگ از مثالهای https://docs.confluent.io/2.0.0/clients/producer.html#examples الهام گرفته شده است.
لیستینگ ۴-۲. منتظر ماندن برای نتیجه
java
RecordMetadata result = // Waits on the response
producer.send(producerRecord).get(); // from the send call
log.info("kinaction_info offset = {}, topic = {}, timestamp = {}",
result.offset(), result.topic(), result.timestamp());
producer.close();منتظر ماندن مستقیم و همگام برای پاسخ تضمین میکند کد نتیجهٔ هر رکورد را قبل از ارسال پیام بعدی مدیریت کند. تمرکز روی تحویل پیامها بدون از دست رفتن است، بیشتر از سرعت!
تا اینجا در فصلهای قبل از چند serializer از پیش ساختهشده استفاده کردیم. برای پیامهای plain text، producer ما از serializer به نام StringSerializer استفاده میکند. و وقتی در فصل ۳ دربارهٔ Avro صحبت کردیم، به کلاس io.confluent.kafka.serializers.KafkaAvroSerializer رسیدیم. اما اگر قالب خاصی میخواهیم تولید کنیم چه؟ این اغلب هنگام کار با اشیاء سفارشی رخ میدهد. از serialization برای ترجمهٔ داده به قالبی که قابل انتقال، ذخیره و سپس بازیابی برای کلون دادهٔ اصلی باشد استفاده میکنیم. لیستینگ ۴-۳ کد کلاس Alert را نشان میدهد.
لیستینگ ۴-۳. کلاس Alert
java
public class Alert implements Serializable {
private final int alertId;
private String stageId;
private final String alertLevel;
private final String alertMessage;
public Alert(int alertId,
String stageId,
String alertLevel, // Holds the alert's ID,
String alertMessage) { // level, and messages
this.alertId = alertId;
this.stageId = stageId;
this.alertLevel = alertLevel;
this.alertMessage = alertMessage;
}
public int getAlertId() {
return alertId;
}
public String getStageId() {
return stageId;
}
public void setStageId(String stageId) {
this.stageId = stageId;
}
public String getAlertLevel() {
return alertLevel;
}
public String getAlertMessage() {
return alertMessage;
}
}لیستینگ ۴-۳ کدی را نشان میدهد که برای ساخت bean به نام Alert برای نگهداری اطلاعاتی که میخواهیم بفرستیم استفاده میکنیم. کسانی که با Java آشنا هستند میبینند لیستینگ چیزی بیش از getter و setter و constructor برای کلاس Alert نیست. حالا که قالبی برای شیء دادهٔ Alert داریم، وقت آن است در ساخت serializer سادهٔ alert به نام AlertKeySerde استفاده کنیم، همانطور که لیستینگ ۴-۴ نشان میدهد.
لیستینگ ۴-۴. serializer ما برای Alert
java
public class AlertKeySerde implements Serializer<Alert>,
Deserializer<Alert> {
public byte[] serialize(String topic, Alert key) { // Sends the topic and the
if (key == null) { // Alert object to our method
return null;
}
return key.getStageId()
.getBytes(StandardCharsets.UTF_8); // Converts objects to
} // bytes (our end goal)
public Alert deserialize
(String topic, byte[] value) { // The rest of the interface methods do
//could return Alert in future if needed // not need any logic at this point.
return null;
}
//...
}در لیستینگ ۴-۵، فعلاً فقط از این کلاس سفارشی بهعنوان key serializer استفاده میکنیم و value serializer را StringSerializer نگه میداریم. جالب است که میتوانیم کلیدها و مقادیر را با serializerهای مختلف روی همان پیام serialize کنیم. اما باید به serializerهای مورد نظر و مقادیر پیکربندی هر دو توجه کنیم. کد interface Serializer را پیادهسازی میکند و فقط فیلد stageId را بهعنوان کلید پیام برمیدارد. این مثال باید ساده باشد چون تمرکز روی تکنیک استفاده از serde است. گزینههای دیگر serde که اغلب استفاده میشوند پیادهسازیهای JSON و Avro هستند.
یادداشت اگر اصطلاح serde را میبینید یا میشنوید، یعنی serializer و deserializer هر دو توسط همان پیادهسازی آن interface مدیریت میشوند [13].
با این حال، هنوز رایج است هر interface جدا تعریف شود. فقط مواظب باشید
StringSerializerدر مقابلStringDeserializerاستفاده کنید؛ تفاوت ممکن است سخت دیده شود!
چیز دیگری که باید بدانید این است که دانستن نحوهٔ deserialize مقادیر، consumerها را در ارتباط با نحوهٔ serialize شدن مقادیر توسط producer درگیر میکند. نوعی توافق یا هماهنگکننده برای قالبهای داده برای کلاینتها لازم است، هرچند Kafka به دادهای که روی brokerها ذخیره میکند اهمیت نمیدهد.
هدف دیگر طراحی کارخانه، ثبت وضعیت روند alert مراحل برای ردیابی alertها در طول زمان بود. چون اطلاعات هر مرحله (نه همهٔ سنسورها همزمان) برایمان مهم است، مفید است به گروهبندی این رویدادها فکر کنیم. در این مورد، چون هر stage ID یکتا است، منطقی است آن ID را بهعنوان کلید استفاده کنیم. لیستینگ ۴-۵ property key.serializer را که تنظیم میکنیم و ارسال alert با سطح CRITICAL را نشان میدهد.
لیستینگ ۴-۵. alert trending producer
java
public class AlertTrendingProducer {
private static final Logger log =
LoggerFactory.getLogger(AlertTrendingProducer.class);
public static void main(String[] args)
throws InterruptedException, ExecutionException {
Properties kaProperties = new Properties();
kaProperties.put("bootstrap.servers",
"localhost:9092,localhost:9093,localhost:9094"); // Tells our producer client
kaProperties.put("key.serializer", // how to serialize our custom
AlertKeySerde.class.getName()); // Alert object into a key
kaProperties.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
try (Producer<Alert, String> producer =
new KafkaProducer<>(kaProperties)) {
Alert alert = new Alert(0, "Stage 0", "CRITICAL", "Stage 0 stopped");
ProducerRecord<Alert, String> producerRecord =
new ProducerRecord<>("kinaction_alerttrend",
alert, alert.getAlertMessage());
RecordMetadata result = producer.send(producerRecord).get();
log.info("kinaction_info offset = {}, topic = {}, timestamp = {}",
result.offset(), result.topic(), result.timestamp());
}
} // Instead of null for the second
} // parameter, uses the actual object
// we want to populate the keyبهطور کلی، همان کلید باید همان تخصیص پارتیشن را تولید کند و چیزی لازم نیست عوض شود. بهعبارت دیگر، همان stage IDها (کلیدها) فقط با استفاده از کلید درست گروهبندی میشوند. توزیع اندازهٔ پارتیشنها را زیر نظر میگیریم تا ببینیم آیا در آینده ناهموار میشوند، اما فعلاً با این رویکرد پیش میرویم. همچنین توجه کنید برای کلاسهای خاصی که در دستنوشته ساختیم، propertyهای کلاس را به روش دیگری تنظیم میکنیم تا گزینهٔ دیگری نشان دهیم. بهجای hardcode کردن مسیر کامل کلاس، میتوانید چیزی مانند AlertKeySerde.class.getName() یا حتی AlertKeySerde.class برای مقدار property استفاده کنید.
آخرین نیازمندی ما پردازش سریع alertها برای اطلاعرسانی به اپراتورها دربارهٔ قطعیهای بحرانی بود تا در این مورد هم بر اساس stage ID گروهبندی کنیم. یکی از دلایل این است که میتوانیم بفهمیم سنسور خراب شده یا بازیابی شده (هر تغییر وضعیت) با نگاه به فقط آخرین رویداد آن stage ID. به تاریخچهٔ بررسی وضعیت اهمیت نمیدهیم، فقط سناریوی فعلی. در این مورد، alertها را هم پارتیشنبندی میخواهیم.
تا اینجا در مثالهای نوشتن به Kafka، داده بدون metadata اضافی از کلاینت به topic هدایت شد. چون topicها از پارتیشنهایی روی brokerها ساخته میشوند، Kafka راه پیشفرضی برای ارسال پیام به پارتیشن مشخص دارد. پیشفرض برای پیام بدون کلید (که در مثالها تا اینجا استفاده کردیم) قبل از Kafka نسخه ۲.۴ استراتژی round-robin بود. در نسخههای بعد از ۲.۴، پیامهای بدون کلید از استراتژی sticky partition استفاده میکنند [14]. اما گاهی روشهای خاصی برای پارتیشنبندی داده داریم. یکی از راهها نوشتن کلاس partitioner یکتای خودمان است.
کلاینت هم میتواند با پیکربندی partitioner یکتا کنترل کند به کدام پارتیشن مینویسد. مثالی برای فکر کردن، سطوح alert از سرویس مانیتورینگ سنسور در فصل ۳ است. اطلاعات برخی سنسورها ممکن است مهمتر باشد؛ مثلاً روی مسیر بحرانی e-bike که در صورت عدم رسیدگی باعث downtime میشود. فرض کنید چهار سطح alert داریم: Critical، Major، Minor و Warning. میتوانیم partitionerای بسازیم که سطوح مختلف را در پارتیشنهای مختلف قرار دهد. کلاینتهای consumer ما همیشه مطمئن میشوند alertهای بحرانی را قبل از بقیه میخوانند.
اگر consumerها با پیامهای log شده همگام بمانند، alertهای بحرانی احتمالاً نگرانی بزرگی نیست. اما لیستینگ ۴-۶ نشان میدهد میتوانیم تخصیص پارتیشن را با کلاسی تغییر دهیم تا alertهای بحرانی به پارتیشن مشخص (مثل پارتیشن ۰) هدایت شوند. (توجه کنید alertهای دیگر هم ممکن است بهخاطر منطق ما روی پارتیشن ۰ بیایند، اما alertهای بحرانی همیشه آنجا میرسند.) منطق آینهٔ مثال DefaultPartitioner در خود Kafka است [15].
لیستینگ ۴-۶. Partitioner برای سطوح alert
java
public int partition(final String topic // AlertLevelPartitioner needs
# ... // to implement the partition
// method for its core logic.
int criticalLevelPartition = findCriticalPartitionNumber(cluster, topic);
return isCriticalLevel(((Alert) objectKey).getAlertLevel()) ?
criticalLevelPartition :
findRandomPartition(cluster, topic, objectKey);
}
//... // Critical alerts should end up
// the partition returned from
// findCriticalPartitionNumberبا پیادهسازی interface Partitioner، میتوانیم از متد partition پارتیشن مشخصی که میخواهیم producer به آن بنویسد را برگردانیم. در این مورد، مقدار کلید تضمین میکند هر رویداد CRITICAL به جای مشخصی برسد؛ مثلاً پارتیشن ۰ میتواند از متد findCriticalPartitionNumber برگردد. علاوه بر ساخت خود کلاس، لیستینگ ۴-۷ نشان میدهد باید کلید پیکربندی partitioner.class را برای استفادهٔ producer از کلاس ساختهشده تنظیم کنیم. پیکربندی که Kafka را قدرت میدهد برای بهرهگیری از کلاس جدید ما استفاده میشود.
لیستینگ ۴-۷. پیکربندی کلاس partitioner
java
Properties kaProperties = new Properties(); // Updates the producer configuration
//... // to reference and use the custom
kaProperties.put("partitioner.class", // partitioner AlertLevelPartitioner
AlertLevelPartitioner.class.getName());این مثال که همیشه شمارهٔ پارتیشن مشخصی برمیگردد میتواند گسترش یابد یا پویاتر شود. میتوانیم با کد سفارشی منطق خاص نیازهای کسبوکار را پیاده کنیم.
لیستینگ ۴-۸ پیکربندی producer برای افزودن مقدار partitioner.class بهعنوان partitioner مشخص ما را نشان میدهد. قصد این است داده در پارتیشن مشخص در دسترس باشد تا consumerهایی که داده را پردازش میکنند به alertهای بحرانی دسترسی داشته باشند و پس از رسیدگی به alertهای دیگر (در پارتیشنهای دیگر) بروند.
لیستینگ ۴-۸. Alert producer
java
public class AlertProducer {
public static void main(String[] args) {
Properties kaProperties = new Properties();
kaProperties.put("bootstrap.servers",
"localhost:9092,localhost:9093");
kaProperties.put("key.serializer", // Reuses the Alert
AlertKeySerde.class.getName()); // key serializer
kaProperties.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
kaProperties.put("partitioner.class",
AlertLevelPartitioner.class.getName()); // Uses the property partitioner.class
// to set our specific partitioner class
try (Producer<Alert, String> producer =
new KafkaProducer<>(kaProperties)) {
Alert alert = new Alert(1, "Stage 1", "CRITICAL", "Stage 1 stopped");
ProducerRecord<Alert, String>
producerRecord = new ProducerRecord<>
("kinaction_alert", alert, alert.getAlertMessage());
producer.send(producerRecord,
new AlertCallback()); // This is the first time we've used
} // a callback to handle the
} // completion or failure of a send.
}افزودنی که در لیستینگ ۴-۸ میبینیم، callback برای اجرا هنگام تکمیل است. اگرچه گفتیم ۱۰۰٪ نگران شکست پیام گاهبهگاه نیستیم بهخاطر فراوانی رویدادها، میخواهیم مطمئن شویم نرخ شکست بالایی نمیبینیم که میتواند نشانهٔ خطاهای مرتبط با برنامه باشد. لیستینگ بعدی مثالی از پیادهسازی interface Callback را نشان میدهد. callback فقط در صورت خطا پیام log میکند. توجه کنید این لیستینگ از مثالهای https://docs.confluent.io/2.0.0/clients/producer.html#examples الهام گرفته شده است.
لیستینگ ۴-۹. Alert callback
java
public class AlertCallback implements Callback { // Implements the Kafka
// Callback interface
private static final Logger log =
LoggerFactory.getLogger(AlertCallback.class);
public void onCompletion
(RecordMetadata metadata, // The completion can
Exception exception) { // have success or failure.
if (exception != null) {
log.error("kinaction_error", exception);
} else {
log.info("kinaction_info offset = {}, topic = {}, timestamp = {}",
metadata.offset(), metadata.topic(), metadata.timestamp());
}
}
}اگرچه در بیشتر مطالب روی نمونههای کوچک تمرکز میکنیم، فکر میکنیم دیدن استفاده از producer در پروژهٔ واقعی هم مفید است. همانطور که گفتیم، Apache Flume میتواند کنار Kafka برای ویژگیهای دادهٔ مختلف استفاده شود. وقتی Kafka را بهعنوان sink استفاده میکنیم، Flume داده را در Kafka قرار میدهد. شاید با Flume آشنا باشید یا نباشید، اما برای این بحث به مجموعهٔ ویژگیهایش علاقهمند نیستیم. میخواهیم ببینیم در موقعیت واقعی چگونه از کد producer Kafka استفاده میکند.
در مثالهای بعدی به Flume نسخه ۱.۸ ارجاع میدهیم (در https://github.com/apache/flume/tree/flume-1.8 اگر بخواهید کد منبع کاملتر را ببینید). لیستینگ ۴-۱۰ قطعهٔ پیکربندی را نشان میدهد که توسط agent Flume استفاده میشود.
لیستینگ ۴-۱۰. پیکربندی Flume sink
a1.sinks.k1.kafka.topic = kinaction_helloworld
a1.sinks.k1.kafka.bootstrap.servers = localhost:9092
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.compression.type = snappyبرخی propertyهای پیکربندی از لیستینگ ۴-۱۰ آشنا به نظر میرسند: topic، acks، bootstrap.servers. در مثالهای قبلی پیکربندیها را بهصورت Properties داخل کد تعریف کردیم. اما لیستینگ ۴-۱۰ مثالی از برنامهای است که مقادیر پیکربندی را externalize میکند، کاری که در پروژههای خودمان هم میتوانیم انجام دهیم. کد منبع KafkaSink از Apache Flume در http://mng.bz/JvpZ مثالی از گرفتن داده و قرار دادن آن در Kafka با کد producer است. لیستینگ ۴-۱۱ مثال دیگری از producer با ایدهٔ مشابه است: گرفتن فایل پیکربندی مانند لیستینگ ۴-۱۰ و بارگذاری آن مقادیر در نمونهٔ producer.
لیستینگ ۴-۱۱. خواندن پیکربندی Kafka producer از فایل
java
...
Properties kaProperties = readConfig();
String topic = kaProperties.getProperty("topic");
kaProperties.remove("topic");
try (Producer<String, String> producer =
new KafkaProducer<>(kaProperties)) {
ProducerRecord<String, String> producerRecord =
new ProducerRecord<>(topic, null, "event");
producer.send(producerRecord,
new AlertCallback()); // Our familiar producer.send
} // with a callback
private static Properties readConfig() {
Path path = Paths.get("src/main/resources/kafkasink.conf");
Properties kaProperties = new Properties(); // Reads an external file
try (Stream<String> lines = Files.lines(path)) // for configuration
lines.forEachOrdered(line ->
determineProperty(line, kaProperties));
} catch (IOException e) {
System.out.println("kinaction_error" + e);
}
return kaProperties;
}
private static void determineProperty // Parses configuration properties
(String line, Properties kaProperties) { // and sets those values
if (line.contains("bootstrap")) {
kaProperties.put("bootstrap.servers", line.split("=")[1]);
} else if (line.contains("acks")) {
kaProperties.put("acks", line.split("=")[1]);
} else if (line.contains("compression.type")) {
kaProperties.put("compression.type", line.split("=")[1]);
} else if (line.contains("topic")) {
kaProperties.put("topic", line.split("=")[1]);
}
...
}اگرچه برخی کد در لیستینگ ۴-۱۱ حذف شده، قطعات اصلی Kafka producer شاید آشنا به نظر برسند. تنظیم پیکربندی و متد send producer باید شبیه کدی باشد که در این فصل نوشتیم. و حالا امیدواریم اعتماد به نفس دارید بفهمید کدام propertyهای پیکربندی تنظیم شدهاند و چه تأثیری دارند.
یک تمرین برای خواننده مقایسهٔ AlertCallback.java با کلاس callback Kafka Sink یعنی SinkCallback در کد منبع http://mng.bz/JvpZ است. هر دو مثال از شیء RecordMetadata برای اطلاعات بیشتر دربارهٔ فراخوانیهای موفق استفاده میکنند. این اطلاعات میتواند به ما کمک کند بیشتر بدانیم پیام producer کجا نوشته شده، شامل پارتیشن و offset در آن پارتیشن مشخص.
درست است که میتوانید از برنامههایی مانند Flume بدون هیچگاه کندن در کد منبعشان موفق باشید. اما فکر میکنیم اگر بخواهید بدانید داخلاً چه میگذرد یا troubleshooting پیشرفته لازم دارید، مهم است بدانید ابزارها چه میکنند. با دانش بنیادین جدید producerها، باید مشخص باشد میتوانید با این تکنیکها خودتان برنامههای قدرتمند بسازید.
۴.۳.۱ نسخههای کلاینت و broker
نکتهٔ مهم این است که نسخههای broker و کلاینت Kafka همیشه لازم نیست یکی باشند. اگر broker با Kafka نسخه ۰.۱۰.۰ اجرا میکنید و کلاینت producer Java شما ۰.۱۰.۲ است، broker این ارتقای نسخهٔ پیام را مدیریت میکند [16]. اما اینکه میتوانید به معنای این نیست که در همهٔ موارد باید انجام دهید. برای بررسی بیشتر سازگاری دوطرفهٔ نسخه، به KIP-97 (http://mng.bz/7jAQ) نگاهی بیندازید.
با شروع ورود داده به Kafka از مانع مهمی عبور کردیم. حالا که عمیقتر در اکوسیستم Kafka هستیم، مفاهیم دیگری برای فتح داریم قبل از اتمام راهحل end-to-end. سؤال بعدی این است: چگونه میتوانیم این داده را بیرون بکشیم تا برنامههای دیگر مصرف کنند؟ ایدههایی دربارهٔ ورود داده به Kafka داریم، پس میتوانیم یاد بگیریم چگونه داده را برای برنامههای دیگر مفید کنیم و به روشهای درست بیرون بیاوریم. کلاینتهای consumer بخش حیاتی این کشف هستند و مانند producerها، رفتارهای مختلف مبتنی بر پیکربندی برای برآوردن نیازمندیهای مختلف مصرف وجود دارد.
خلاصه
- کلاینتهای producer به توسعهدهندگان راهی برای ورود داده به Kafka میدهند.
- تعداد زیادی گزینهٔ پیکربندی برای کنترل رفتار کلاینت بدون کد سفارشی در دسترس است.
- داده روی brokerها در چیزی به نام پارتیشنها ذخیره میشود.
- کلاینت میتواند با ارائهٔ منطق خودش از طریق interface
Partitionerکنترل کند داده به کدام پارتیشن نوشته شود. - Kafka بهطور کلی داده را دنبالهای از بایت میبیند. اما serializerهای سفارشی میتوانند با قالبهای دادهٔ خاص کار کنند.
منابع
- J. Kreps. "Why Avro for Kafka Data?" Confluent blog (February 25, 2015). https://www.confluent.io/blog/avro-kafka-data/ (accessed November 23, 2017).
- "Sender.java." Apache Kafka. GitHub (n.d.). https://github.com/apache/kafka/blob/299eea88a5068f973dc055776c7137538ed01c62/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java (accessed August 20, 2021).
- "Producer Configurations: Retries." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_retries (accessed May 29, 2020).
- "Producer Configurations: max.in.flight.requests.per.connection." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#max.in.flight.requests.per.connection (accessed May 29, 2020).
- "Producer Configurations: enable.idempotence." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#producerconfigs_enable.idempotence (accessed May 29, 2020).
- "KafkaProducer." Apache Software Foundation (n.d.). https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html (accessed July 7, 2019).
- "Producer Configurations." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html (accessed May 29, 2020).
- "Producer Configurations: bootstrap.servers." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#bootstrap.servers (accessed May 29, 2020).
- "Producer Configurations: acks." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#acks (accessed May 29, 2020).
- "Documentation: Message Delivery Semantics." Apache Software Foundation (n.d.). https://kafka.apache.org/documentation/#semantics (accessed May 30, 2020).
- "Topic Configurations: message.timestamp.type." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/topic-configs.html#topicconfigs_message.timestamp.type (accessed July 22, 2020).
- KIP-42: "Add Producer and Consumer Interceptors," Wiki for Apache Kafka, Apache Software Foundation. https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors (accessed April 15, 2019).
- "Kafka Streams Data Types and Serialization." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/streams/developer-guide/datatypes.html (accessed August 21, 2021).
- J. Olshan. "Apache Kafka Producer Improvements with the Sticky Partitioner." Confluent blog (December 18, 2019). https://www.confluent.io/blog/apache-kafka-producer-improvements-sticky-partitioner/ (accessed August 21, 2021).
- "DefaultPartitioner.java," Apache Software Foundation. GitHub (n.d.). https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java (accessed March 22, 2020).
- C. McCabe. "Upgrading Apache Kafka Clients Just Got Easier." Confluent blog (July 18, 2017). https://www.confluent.io/blog/upgrading-apache-kafka-clients-just-got-easier/ (accessed August 21, 2021).