حالت تاریک
فصل ۵ — Consumerها: باز کردن قفل داده
این فصل شامل:
- بررسی consumer و نحوهٔ کار آن
- استفاده از consumer groupها برای هماهنگی خواندن داده از topicها
- یادگیری دربارهٔ offsetها و نحوهٔ استفاده از آنها
- بررسی گزینههای پیکربندی مختلف که رفتار consumer را تغییر میدهند
در فصل قبل شروع به نوشتن داده در سیستم Kafka کردیم. اما همانطور که میدانید، این فقط بخشی از داستان است. Consumerها داده را از Kafka میگیرند و آن مقادیر را به سیستمها یا برنامههای دیگر میدهند. چون consumerها کلاینتهایی هستند که خارج از brokerها وجود دارند، مانند کلاینتهای producer میتوانند به زبانهای برنامهنویسی مختلف نوشته شوند. توجه کنید وقتی نحوهٔ کار را در این فصل بررسی میکنیم، سعی میکنیم به پیشفرضهای کلاینت consumer Java تمایل داشته باشیم. پس از خواندن این فصل، برای حل نیازمندیهای کسبوکار قبلی با مصرف داده به چند روش مختلف در مسیر خواهیم بود.
۵.۱ یک مثال
کلاینت consumer برنامهای است که به topic یا topicهایی که برایشان مهم است subscribe میکند [1]. مانند کلاینتهای producer، فرآیندهای consumer واقعی میتوانند روی ماشینهای جدا اجرا شوند و لازم نیست روی سرور خاصی باشند. در واقع بیشتر کلاینتهای consumer در محیط production روی hostهای جدا هستند. تا زمانی که کلاینتها بتوانند به brokerهای Kafka متصل شوند، میتوانند پیام بخوانند. شکل ۵-۱ دوباره دامنهٔ گستردهٔ Kafka را معرفی میکند و consumerهایی را نشان میدهد که خارج از brokerها برای گرفتن داده از Kafka اجرا میشوند.
شکل ۵-۱. نمای کلی کلاینتهای consumer در Kafka
Producer clients JVM application-message broker Examples Kafka core Like our Data in • Databases (to partition) kinaction_alert OS page cache (memory) • IOT events producer • 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. Like our • HDFS (from partition) • New entries added to the end. kinaction_alert • S3 • No database storage, just disk. consumer • 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
چرا مهم است بدانیم consumer به topicها subscribe میکند (پیامها را pull میکند) و push نمیشود؟ در این موقعیت قدرت کنترل پردازش به consumer منتقل میشود. شکل ۵-۱ نشان میدهد کلاینتهای consumer در اکوسیستم کلی Kafka کجا قرار میگیرند. کلاینتها مسئول خواندن داده از topicها و در دسترس قرار دادن آن برای برنامه (مانند داشبوردهای metrics یا موتورهای analytics) یا ذخیره در سیستمهای دیگر هستند. خود consumerها نرخ مصرف را کنترل میکنند.
با consumer در رانندهصند، اگر شکستی رخ دهد و برنامههای consumer دوباره آنلاین شوند، میتوانند دوباره pull کنند. لازم نیست همیشه consumerها روشن باشند تا اعلانها را مدیریت (یا از دست بدهند). اگرچه میتوانید برنامههایی بسازید که جریان دادهٔ مداوم یا حتی فشار انباشته (back pressure) بهخاطر حجم را مدیریت کنند، باید بدانید listener برای brokerها نیستید؛ consumerها داده را pull میکنند. برای خوانندگانی که قبلاً از Kafka استفاده کردهاند، شاید بدانید دلایلی هست که نمیخواهید consumerها مدت طولانی خاموش باشند. وقتی جزئیات بیشتری دربارهٔ topicها بحث کنیم، میبینیم داده ممکن است بهخاطر محدودیتهای اندازه یا زمان که کاربر تعریف میکند از Kafka حذف شود.
۵.۱.۱ گزینههای consumer
در بحثمان متوجه propertyهایی میشوید که به آنهایی که برای کلاینتهای producer لازم بود مرتبطاند. همیشه باید brokerهایی را که در راهاندازی کلاینت میتوانیم به آنها متصل شویم بدانیم. یک «gotcha» کوچک این است که مطمئن شوید deserializerهای کلید و مقدار با serializerهایی که پیام را با آنها تولید کردید مطابقت دارند. مثلاً اگر با StringSerializer تولید کنید اما با LongDeserializer مصرف کنید، استثنایی میگیرید که باید رفع کنید.
جدول ۵-۱ برخی از مقادیر پیکربندی را که هنگام نوشتن consumerهای خودمان باید بدانیم فهرست میکند [2].
جدول ۵-۱. پیکربندی consumer
| کلید | هدف |
|---|---|
bootstrap.servers | یک یا چند broker Kafka برای اتصال در راهاندازی |
value.deserializer | برای deserialization مقدار لازم است |
key.deserializer | برای deserialization کلید لازم است |
group.id | نامی که برای پیوستن به consumer group استفاده میشود |
client.id | شناسه برای شناسایی کاربر (در فصل ۱۰ استفاده میکنیم) |
heartbeat.interval.ms | فاصلهٔ pingهای consumer به group coordinator |
یکی از راههای برخورد با همهٔ نامهای کلید پیکربندی consumer، استفاده از ثابتهای کلاس ConsumerConfig (به http://mng.bz/oGgy مراجعه کنید) و جستجوی برچسب Importance با مقدار «high» در وبسایت Confluent (http://mng.bz/drdv) است. اما در مثالهایمان برای وضوح از خود نام propertyها استفاده میکنیم. لیستینگ ۵-۱ چهار تا از این کلیدها را در عمل نشان میدهد. مقادیر پیکربندی در جدول ۵-۱ تعیین میکنند consumer چگونه با brokerها و consumerهای دیگر تعامل کند.
حالا به خواندن از topic با یک consumer همانطور که در فصل ۲ کردیم برمیگردیم. برای این مثال، برنامهای مشابه شروع احتمالی Kafka در LinkedIn داریم که با رویدادهای فعالیت کاربر (در فصل ۱ ذکر شد) سروکار دارد [3]. فرض کنید فرمول خاصی داریم که از زمان حضور کاربر در صفحه و تعداد تعاملاتش استفاده میکند و بهعنوان مقدار به topic فرستاده میشود تا نرخ کلیک آینده با پروموشن جدید پیشبینی شود. تصور کنید consumer را اجرا میکنیم و همهٔ پیامهای topic را پردازش میکنیم و از کاربرد فرمول (در این مورد ضرب در عدد جادویی) راضی هستیم.
لیستینگ ۵-۱ مثالی از نگاه به رکوردهای topic kinaction_promos و چاپ مقداری بر اساس دادهٔ هر رویداد است. این لیستینگ شباهتهای زیادی به کد producer فصل ۴ دارد، جایی که propertyها برای تعیین رفتار consumer استفاده میشوند. استفاده از deserializer برای کلیدها و مقادیر با داشتن serializer برای producerها متفاوت است و بسته به topicای که مصرف میکنیم فرق میکند.
یادداشت لیستینگ ۵-۱ لیستینگ کد کامل نیست بلکه برای برجسته کردن خطوط خاص consumer است. بهخاطر داشته باشید consumer میتواند به چند topic subscribe کند، اما در این مورد فقط به topic
kinaction_promosعلاقهمندیم.
در لیستینگ، حلقهای هم برای poll کردن پارتیشنهای topic که consumer به آنها اختصاص یافته برای پردازش پیامها استفاده میشود. این حلقه با مقدار Boolean toggle میشود. این نوع حلقه میتواند خطا ایجاد کند، بهویژه برای برنامهنویسان مبتدی! پس چرا این حلقه؟ بخشی از ذهنیت streaming رویدادها را بهعنوان جریان پیوسته در نظر میگیرد و در منطق منعکس میشود. توجه کنید این مثال ۲۵۰ را برای مدت poll به میلیثانیه استفاده میکند. این timeout نشان میدهد فراخوانی چقدر thread اصلی برنامه را با انتظار block میکند، اما وقتی رکوردها آمادهٔ تحویل باشند میتواند فوراً برگردد [4]. این مقدار چیزی است که میتوانید بر اساس نیاز برنامههایتان fine-tune کنید. مرجع (و جزئیات بیشتر) برای سبک Java 8 استفاده از addShutdownHook در لیستینگ بعدی در https://docs.confluent.io/platform/current/streams/developer-guide/write-streams.html دیده میشود.
لیستینگ ۵-۱. Promotion consumer
java
...
private volatile boolean keepConsuming = true;
public static void main(String[] args) { // Defines group.id. (We'll
Properties kaProperties = new Properties(); // discuss this with
kaProperties.put("bootstrap.servers", // consumer groups.)
"localhost:9092,localhost:9093,,localhost:9094");
kaProperties.put("group.id",
"kinaction_webconsumer");
kaProperties.put("enable.auto.commit", "true");
kaProperties.put("auto.commit.interval.ms", "1000");
kaProperties.put("key.deserializer", // Defines
// deserializers for
"org.apache.kafka.common.serialization.StringDeserializer");
// the key and values
kaProperties.put("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer");
WebClickConsumer webClickConsumer = new WebClickConsumer();
webClickConsumer.consume(kaProperties);
Runtime.getRuntime()
.addShutdownHook(
new Thread(webClickConsumer::shutdown)
);
}
private void consume(Properties kaProperties) { // Passes the properties
try (KafkaConsumer<String, String> consumer = // into the KafkaConsumer
new KafkaConsumer<>(kaProperties)) { // constructor
consumer.subscribe(
List.of("kinaction_promos") // Subscribes to one topic,
); // kinaction_promos
while (keepConsuming) { // Uses a loop to poll
ConsumerRecords<String, String> records = // for topic records
consumer.poll(Duration.ofMillis(250));
for (ConsumerRecord<String, String> record : records) {
log.info("kinaction_info offset = {}, key = {}",
record.offset(),
record.key());
log.info("kinaction_info value = {}",
Double.parseDouble(record.value()) * 1.543);
}
}
}
}
private void shutdown() {
keepConsuming = false;
}
}پس از تولید مقدار برای هر پیام در topic در لیستینگ ۵-۱، میفهمیم فرمول مدلسازی درست نیست! حالا چه کنیم؟ تلاش برای محاسبهٔ مجدد دادهای که از نتایج نهایی داریم (با فرض سختتر بودن اصلاح نسبت به مثال) و سپس اعمال فرمول جدید؟
اینجاست که میتوانیم از دانش رفتار consumer در Kafka برای replay پیامهایی که قبلاً پردازش کردیم استفاده کنیم. با نگهداشتن دادهٔ خام، نگران بازسازی دادهٔ اصلی نیستیم. اشتباهات توسعهدهنده، اشتباهات منطق برنامه و حتی شکست برنامههای وابسته قابل اصلاح است چون داده پس از مصرف از topicها حذف نمیشود. این هم توضیح میدهد چگونه «سفر در زمان» به نوعی با Kafka ممکن است.
بیایید به نحوهٔ توقف consumer بپردازیم. قبلاً دیدید Ctrl-C برای پایان پردازش یا توقف فرآیند در ترمینال. اما روش درست شامل فراخوانی متد close روی consumer است [23].
لیستینگ ۵-۲ consumerای را نشان میدهد که روی thread اجرا میشود و کلاس دیگری shutdown را کنترل میکند. وقتی کد لیستینگ ۵-۲ شروع میشود، thread با نمونهٔ consumer اجرا میشود. با فراخوانی متد public shutdown، کلاس دیگری میتواند Boolean را تغییر دهد و consumer را از poll برای رکوردهای جدید متوقف کند. متغیر stopping نگهبان ماست که تصمیم میگیرد پردازش ادامه یابد یا نه. فراخوانی متد wakeup هم باعث پرتاب WakeupException میشود که به بلوک finally منجر به بستن صحیح منبع consumer میشود [5]. لیستینگ ۵-۲ از https://kafka.apache.org/26/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html بهعنوان مستندات مرجع استفاده کرده است.
لیستینگ ۵-۲. بستن consumer
java
public class KinactionStopConsumer implements Runnable {
private final KafkaConsumer<String, String> consumer;
private final AtomicBoolean stopping =
new AtomicBoolean(false);
...
public KinactionStopConsumer(KafkaConsumer<String, String> consumer) {
this.consumer = consumer;
}
public void run() {
try {
consumer.subscribe(List.of("kinaction_promos"));
while (!stopping.get()) { // The variable stopping
ConsumerRecords<String, String> records = // determines whether to
consumer.poll(Duration.ofMillis(250)); // continue processing.
...
}
} catch (WakeupException e) { // The client shutdown hook
if (!stopping.get()) throw e; // triggers WakeupException.
} finally {
consumer.close(); // Stops the client and informs
} // the broker of the shutdown
}
public void shutdown() { // Calls shutdown from a different
stopping.set(true); // thread to stop the client properly
consumer.wakeup();
}
}با حرکت به موضوع بعد، برای پیشروی بیشتر باید offsetها را بفهمیم و چگونه برای کنترل نحوهٔ خواندن داده توسط consumerها استفاده شوند.
۵.۱.۲ درک مختصات ما
یکی از مواردی که تا اینجا فقط گذرا دربارهاش صحبت کردیم مفهوم offsetها است. offsetها را بهعنوان موقعیتهای index در log استفاده میکنیم که consumer به broker میفرستد. این به log میگوید کدام پیامها را میخواهد مصرف کند و از کجا. اگر به مثال console consumer ما برگردید، از پرچم --from-beginning استفاده کردیم. این پشت صحنه پارامتر پیکربندی consumer auto.offset.reset را روی earliest تنظیم میکند. با آن پیکربندی، باید همهٔ رکوردهای آن topic برای پارتیشنهایی که به آنها متصل هستید را ببینید، حتی اگر قبل از شروع console consumer فرستاده شده باشند. بخش بالای شکل ۵-۲ خواندن از ابتدای log هر بار در این حالت را نشان میدهد.
شکل ۵-۲. offsetهای Kafka [6]
Example kinaction_alert offset numbers From beginning, reads starting at 0 Spot 0 Spot 1 Spot 2 Spot 3 Spot 0 Spot 1 Spot 2 Spot 3 Spot 4 Latest reads start on next message. Offset numbers do not change.
اگر گزینهٔ auto.offset.reset را اضافه نکنید، پیشفرض latest است. شکل ۵-۲ این حالت را هم نشان میدهد. در این مورد، هیچ پیامی از producer نمیبینید مگر بعد از شروع consumer بفرستید. این گزینه میگوید پیامهایی که قبلاً در پارتیشن topic که consumer میخواند هستند را نادیده بگیرید؛ فقط آنچه بعد از شروع poll کلاینت consumer میآید را پردازش میخواهیم. میتوانید این را آرایهٔ بینهایتی با index شروع از ۰ بدانید. اما بهروزرسانی برای index مجاز نیست. هر تغییر باید به انتهای log اضافه شود.
توجه کنید offsetها برای هر پارتیشن همیشه افزایش مییابند. وقتی پارتیشن topic offset ۰ را دید، حتی اگر آن پیام بعداً حذف شود، شمارهٔ offset دوباره استفاده نمیشود. برخی ممکن است با مشکل اعدادی که تا حد بالای نوع داده افزایش مییابند مواجه شده باشند. هر پارتیشن دنبالهٔ offset خودش را دارد، پس امید است ریسک کم باشد.
برای پیامی که به topic نوشته شده، مختصات یافتن پیام چیست؟ ابتدا پارتیشن داخل topic که به آن نوشته شده را پیدا میکنیم، سپس offset مبتنی بر index. همانطور که شکل ۵-۳ نشان میدهد، consumerها معمولاً از replica leader پارتیشن consumer میخوانند. این replica leader consumer میتواند بهخاطر تغییرات leadership در طول زمان با replica leader هر producer متفاوت باشد؛ اما از نظر مفهومی عموماً مشابهاند.
شکل ۵-۳. Leaderهای پارتیشن
Topic: 3 partitions, 2 replicas Broker Partition 1 Partition 1 leader 0 1 2 3 4 P2 copy Broker Partition 2 Partition 2 leader 0 1 2 3 Each partition has its own log sequence. Consumers will only P3 copy read from partition leaders. Broker Partition 3 We will read from kinaction_alert later in Partition 3 leader 0 1 this chapter from its leader. P1 copy
همچنین وقتی دربارهٔ پارتیشنها صحبت میکنیم، داشتن همان شمارهٔ offset در پارتیشنهای مختلف مشکلی ندارد. تفکیک پیامها باید جزئیات پارتیشنی که در topic صحبت میکنیم و offset را هم شامل شود.
بهعنوان یادداشت جانبی، اگر بهخاطر مسئلهای مانند نگرانی latency شبکه (مثلاً کلاستری که مراکز داده را در بر میگیرد) نیاز به fetch از replica follower دارید، KIP-392 این قابلیت را در نسخه ۲.۴.۰ معرفی کرد [7]. وقتی با اولین کلاسترها شروع میکنید، توصیه میکنیم با رفتار پیشفرض شروع کنید و فقط وقتی تأثیر واقعی لازم است به این ویژگی برسید. اگر کلاستر را در سایتهای فیزیکی مختلف ندارید، احتمالاً فعلاً به این ویژگی نیاز ندارید.
پارتیشنها نقش مهمی در نحوهٔ پردازش پیامها دارند. اگرچه topic نام منطقی برای آنچه consumerها به آن علاقهمندند است، از replica leader پارتیشنهای اختصاصیافته میخوانند. اما consumerها چگونه پارتیشن اتصال را پیدا میکنند؟ و نه فقط کدام پارتیشن، leader آن پارتیشن کجاست؟ برای هر گروه consumer، broker مشخصی نقش group coordinator را بر عهده میگیرد [8]. کلاینت consumer با این coordinator صحبت میکند تا تخصیص پارتیشن و جزئیات دیگر لازم برای مصرف پیامها را بگیرد.
تعداد پارتیشنها هم در بحث مصرف نقش دارد. برخی consumerها با consumer بیشتر از پارتیشنها کاری نمیگیرند. مثال: چهار consumer و فقط سه پارتیشن. چرا ممکن است با این موافق باشید؟ در برخی موارد میخواهید مطمئن شوید اگر consumer ناگهان از بین رفت، نرخ مصرف مشابهی حفظ شود. group coordinator نه فقط در ابتدای راهاندازی گروه تعیین میکند کدام consumerها کدام پارتیشنها را بخوانند، بلکه وقتی consumer اضافه یا شکست میخورد و از گروه خارج میشود هم [8]. و در حالتی که پارتیشن بیشتر از consumer باشد، consumer در صورت نیاز بیش از یک پارتیشن را مدیریت میکند.
شکل ۵-۴ نمای کلی از چگونگی خواندن چهار consumer همهٔ داده روی brokerها را نشان میدهد، جایی که topic subscribeشده replica leader پارتیشنها را بهطور یکنواخت پخش کرده، یکی روی هر سه broker. در این شکل، داده تقریباً هماندازه است که همیشه اینطور نیست. یک consumer بدون کار آماده است چون هر replica leader پارتیشن فقط توسط یک consumer مدیریت میشود.
شکل ۵-۴. یک consumer اضافی Kafka
This consumer reads This consumer reads one section of the one section of the total data. total data. Data Data Java consumer 2 Java consumer 1 + poll() + poll() + subscribe() + subscribe() Data Our examples in this book usually only have 3 partitions, so 4 consumers would have 1 be without work. This consumer sits ready but does not read any data. Java consumer 6 Java consumer 3 This consumer reads + poll() + poll() one section of the + subscribe() + subscribe() total data.
چون تعداد پارتیشنها مقدار consumerهای موازی را تعیین میکند، برخی میپرسند چرا همیشه عدد بزرگی مثل ۵۰۰ پارتیشن انتخاب نکنید. این جستجوی throughput بالاتر رایگان نیست [9]. به همین دلیل باید آنچه با شکل جریان دادهتان بهتر میخورد انتخاب کنید.
یک ملاحظهٔ کلیدی این است که پارتیشنهای زیاد ممکن است latency end-to-end را افزایش دهد. اگر میلیثانیه در برنامهتان مهم است، شاید نتوانید تا replicate شدن پارتیشن بین brokerها صبر کنید [9]. این برای داشتن in-sync replicaها کلیدی است و قبل از در دسترس بودن پیام برای تحویل به consumer انجام میشود. همچنین باید مصرف حافظهٔ consumerها را زیر نظر بگیرید. اگر نگاشت ۱ به ۱ پارتیشن به consumer ندارید، نیاز حافظهٔ هر consumer با اختصاص پارتیشنهای بیشتر افزایش مییابد [9].
اگر در مستندات قدیمی Kafka به پیکربندیهای کلاینت consumer برای Apache ZooKeeper برخورد کردید، مگر از کلاینت consumer قدیمی استفاده کنید، Kafka consumerها را مستقیماً به ZooKeeper وابسته نمیکند. اگرچه consumerها قبلاً از ZooKeeper برای ذخیرهٔ offsetهایی که تا نقطهای مصرف کردهاند استفاده میکردند، حالا offsetها اغلب داخل topic داخلی Kafka ذخیره میشوند [10]. بهعنوان یادداشت جانبی، کلاینتهای consumer مجبور نیستند offsetهایشان را در هیچکدام از این مکانها ذخیره کنند، اما احتمالاً اینطور است. اگر میخواهید ذخیرهٔ offset خودتان را مدیریت کنید میتوانید! میتوانید در فایل محلی، ذخیرهٔ ابری با ارائهدهندهای مانند AWS™، یا پایگاه داده ذخیره کنید. یکی از مزایای دور شدن از ذخیرهٔ ZooKeeper کاهش وابستگی کلاینتها به ZooKeeper بود.
۵.۲ نحوهٔ تعامل consumerها
چرا مفهوم consumer groupها محوری است؟ مهمترین دلیل احتمالاً این است که مقیاسپذیری با افزودن یا حذف consumer از گروه تحت تأثیر قرار میگیرد. Consumerهایی که بخشی از همان گروه نیستند دانش offset هماهنگ مشترکی ندارند.
لیستینگ ۵-۳ مثالی از گروهی به نام kinaction_team0group است. اگر بهجای آن group.id جدیدی (مثل GUID تصادفی) بسازید، consumer جدیدی بدون offset ذخیرهشده و بدون consumer دیگر در گروه شروع میکنید [11]. اگر به گروه موجود (یا گروهی که قبلاً offset ذخیره کرده) بپیوندید، consumer میتواند کار را با دیگران تقسیم کند یا حتی از جایی که در اجراهای قبلی خواندن متوقف شده بود ادامه دهد [1].
لیستینگ ۵-۳. پیکربندی consumer برای consumer group
java
Properties kaProperties = new Properties(); // group.id determines
kaProperties.put("group.id", "kinaction_team0group"); // consumer behavior
// with other consumers.اغلب consumerهای زیادی از همان topic میخوانند. جزئیات مهمی که برای نیاز به group ID جدید تصمیم میگیرید این است که consumerها بخشی از یک برنامه کار میکنند یا جریانهای منطقی جدا. چرا مهم است؟
دو use case برای دادهای از سیستم منابع انسانی را در نظر بگیرید. یک تیم دربارهٔ تعداد استخدام از ایالتهای خاص میپرسد و تیم دیگر بیشتر به داده برای تأثیر بر بودجهٔ سفر مصاحبهها علاقهمند است. آیا کسی از تیم اول به کار تیم دیگر اهمیت میدهد یا هر تیم فقط بخشی از پیامها را میخواهد مصرف کند؟ احتمالاً نه! چگونه این جداسازی را حفظ کنیم؟ پاسخ اختصاص group.id جدا به هر برنامه است. هر consumer که همان group.id دیگری را استفاده کند بهعنوان همکاری برای مصرف پارتیشنها و offsetهای topic بهعنوان یک برنامهٔ منطقی در نظر گرفته میشود.
۵.۳ ردیابی
در الگوهای استفادهای که تا اینجا داشتیم، زیاد دربارهٔ نحوهٔ نگهداری سابقهٔ آنچه هر کلاینت خوانده صحبت نکردیم. بیایید کوتاه بگوییم برخی message brokerها در سیستمهای دیگر چگونه پیام را مدیریت میکنند. در برخی سیستمها consumerها ثبت نمیکنند چه خواندهاند. پیام را pull میکنند و پس از acknowledgment دیگر در صف وجود ندارد. این برای پیام واحدی که دقیقاً یک برنامه باید پردازش کند خوب کار میکند. برخی سیستمها از topic برای publish پیام به همهٔ subscriberها استفاده میکنند. و اغلب subscriberهای آینده این پیام را کاملاً از دست میدهند چون هنگام رویداد فعال در لیست گیرنده نبودند.
شکل ۵-۵ سناریوهای message broker غیر Kafka را نشان میدهد، از جمله حذف اغلب پیامها پس از مصرف. همچنین الگوی دومی که پیام از منبع اصلی میآید و سپس به صفهای دیگر replicate میشود. در سیستمهایی که پیام مصرف میشد و برای بیش از یک consumer در دسترس نبود، این رویکرد لازم است تا هر برنامهٔ جدا کپی بگیرد.
شکل ۵-۵. سناریوهای broker دیگر
Read once and acknowledge Queue The only message Producer 0 1 2 3 Consumer available to read is 3. Multiple consumers want the same message. Our kinaction_alert topic messages are not removed. Queue 0 0 1 2 Consumer 0 Queue 1 For each consumer to get the same message, some systems Producer 0 1 2 Consumer 1 fan out with copies of the data. Queue 2 0 1 2 Consumer 2
میتوانید تصور کنید با محبوب شدن رویداد بهعنوان منبع اطلاعات، تعداد کپیها زیاد میشود. بهجای داشتن کپی کامل صف (جز برای replication یا failover)، Kafka میتواند چند برنامه را از همان replica leader پارتیشن سرو کند.
Kafka، همانطور که در فصل اول گفتیم، محدود به یک consumer نیست. حتی اگر برنامهٔ مصرفکننده هنگام ایجاد اولین پیام روی topic وجود نداشته باشد، تا زمانی که Kafka پیام را در log نگه دارد، هنوز میتواند داده را پردازش کند. چون پیامها برای consumerهای دیگر حذف نمیشوند یا فقط یکبار تحویل داده نمیشوند، کلاینتهای consumer به راهی برای ثبت جایی که در topic خواندهاند نیاز دارند. علاوه بر این، چون برنامههای زیادی میتوانند همان topic را بخوانند، مهم است offsetها و پارتیشنها مخصوص consumer group مشخص باشند. مختصات کلیدی برای همکاری کلاینتهای consumer ترکیب یکتای موارد زیر است: group، topic و شمارهٔ پارتیشن.
۵.۳.۱ Group coordinator
همانطور که گفتیم، group coordinator با کلاینتهای consumer همکاری میکند تا ثبت کند آن گروه مشخص داخل topic تا کجا خوانده [8]. مختصات پارتیشن topic و group ID آن را به مقدار offset مشخص میکند.
با نگاه به شکل ۵-۶، میبینیم میتوانیم از offset commitها بهعنوان مختصات برای فهمیدن از کجا بعد بخوانیم استفاده کنیم. مثلاً در شکل، consumerای که بخشی از گروه kinaction_teamoffka0 است و پارتیشن ۰ اختصاص یافته، آمادهٔ خواندن offset ۳ بعدی است.
شکل ۵-۶. مختصات
Topic: kinaction_alert Group ID kinaction_teamoffka0 last read: Partition 0 0 1 2 3 Partition 0, offset 2 Partition 1 Three partitions 0 1 2 3 Partition 1, offset 1 Partition 2 0 1 2 Partition 2, offset 1 This offset information tells you where you are and what to consume next!
شکل ۵-۷ سناریویی را نشان میدهد که همان پارتیشنهای مورد علاقه روی سه broker جدا برای دو consumer group مختلف kinaction_teamoffka0 و kinaction_teamsetka1 وجود دارد. consumerهای هر گروه کپی خودشان از داده پارتیشنها روی هر broker را میگیرند. مگر بخشی از همان گروه باشند با هم کار نمیکنند. عضویت صحیح گروه برای مدیریت دقیق metadata هر گروه مهم است.
شکل ۵-۷. Consumerها در گروههای جدا [12]
Consumers from different groups ignore each other, getting their own copy of the data. Java consumer 1: Java consumer 2: kinaction_teamoffka0 kinaction_teamoffka0 Data Data + poll() + poll() + subscribe() + subscribe() Java consumer 4: Java consumer 5: kinaction_teamsetka1 Data kinaction_teamsetka1 + poll() + poll() + subscribe() + subscribe() Java consumer 3: Java consumer 6: kinaction_teamoffka0 kinaction_teamsetka1 + poll() + poll() + subscribe() + subscribe() Multiple consumers can read the same data because they have different consumer IDs.
بهعنوان قاعدهٔ کلی، فقط یک consumer از هر consumer group میتواند یک پارتیشن را بخواند. بهعبارت دیگر، در حالی که پارتیشن ممکن است توسط consumerهای زیاد خوانده شود، فقط توسط یک consumer از هر گروه در هر زمان خوانده میشود. شکل ۵-۸ برجسته میکند یک consumer میتواند دو replica leader پارتیشن بخواند، در حالی که consumer دوم فقط دادهٔ replica leader پارتیشن سوم را میخواند [8]. یک replica پارتیشن واحد نباید بین بیش از یک consumer با همان ID تقسیم یا به اشتراک گذاشته شود.
شکل ۵-۸. Consumerهای Kafka در یک گروه
Two consumers reading a 3-partition This consumer reads one kinaction_alert topic would have This consumer reads two section of the total data. one consumer reading 2 partitions. sections of the total data. Data Data Java consumer Java consumer + poll() + poll() + subscribe() + subscribe() Data
یکی از نکات جالب عضویت در consumer group این است که وقتی consumer شکست میخورد، پارتیشنهایی که میخواند دوباره اختصاص مییابند [8]. consumer موجود جای خواندن پارتیشنی را میگیرد که قبلاً consumer خارجشده از گروه میخواند.
جدول ۵-۱ heartbeat.interval.ms را فهرست کرد که مقدار pingها به group coordinator را تعیین میکند [13]. این heartbeat راه ارتباط consumer با coordinator برای اطلاع از پاسخ بهموقع و کار مداوم است [8].
شکست کلاینت consumer در ارسال heartbeat در مدت زمان میتواند به چند روش رخ دهد، مانند توقف کلاینت consumer با خاتمهٔ فرآیند یا شکست بهخاطر استثنای fatal. اگر کلاینت اجرا نمیشود، نمیتواند پیام به group coordinator برگرداند [8].
۵.۳.۲ استراتژی تخصیص پارتیشن
موردی که باید بدانیم نحوهٔ اختصاص consumerها به پارتیشنها است. این مهم است چون به فهمیدن هر consumer چند پارتیشن ممکن است پردازش کند کمک میکند. property partition.assignment.strategy تعیین میکند کدام پارتیشنها به هر consumer اختصاص یابند [14]. Range و RoundRobin ارائه شدهاند، همچنین Sticky و CooperativeSticky [15].
range assigner از یک topic برای یافتن تعداد پارتیشنها (مرتبشده بر اساس شماره) استفاده میکند و سپس بر اساس تعداد consumerها تقسیم میشود. اگر تقسیم یکنواخت نباشد، consumerهای اول (با ترتیب الفبایی) پارتیشنهای باقیمانده را میگیرند [16]. مطمئن شوید توزیع پارتیشنی که consumerها میتوانند مدیریت کنند دارید و در صورت نیاز استراتژی تخصیص را عوض کنید اگر برخی کلاینتهای consumer همهٔ منابعشان را مصرف میکنند در حالی که دیگران خوباند. شکل ۵-۹ نشان میدهد سه کلاینت سه تا از هفت پارتیشن کل را میگیرند و پارتیشن بیشتری از آخرین کلاینت دارند.
استراتژی round-robin جایی است که پارتیشنها بهطور یکنواخت در ردیف consumerها توزیع میشوند [1]. شکل ۵-۹ شکل اصلاحشدهای از مقاله «What I have learned from Kafka partition assignment strategy» است که مثال سه کلاینت بخشی از همان consumer group و تخصیص round-robin برای یک topic با هفت پارتیشن را نشان میدهد [17]. consumer اول پارتیشن اول، consumer دوم پارتیشن دوم و همینطور تا تمام شدن پارتیشنها.
شکل ۵-۹. تخصیص پارتیشنها
Range RoundRobin Topic with 7 partitions Topic with 7 partitions 1 2 3 4 5 6 7 1 2 3 4 5 6 7 Client 1 Client 2 Client 3 Client 1 Client 2 Client 3 (1, 2, 3) (4, 5) (6, 7) (1, 4, 7) (2, 5) (3, 6)
استراتژی sticky در نسخه ۰.۱۱.۰ اضافه شد [18]. اما چون در بیشتر مثالهای داخلی از range assigner استفاده میکنیم و round-robin را هم دیدیم، به Sticky و CooperativeSticky عمیق نمیپردازیم.
۵.۴ علامتگذاری جایمان
یکی از موارد مهم فکر کردن به نیاز برای اطمینان از اینکه برنامهها همهٔ پیامهای topic را میخوانند. اشکالی دارد چند تا از دست برود، یا هر پیام هنگام خواندن تأیید شود؟ تصمیم واقعی به نیازمندیها و trade-offهایی که حاضرید بپذیرید برمیگردد. برای دیدن هر پیام حاضرید کمی سرعت قربانی کنید؟ این گزینهها در این بخش بحث میشوند.
یک گزینه استفاده از enable.auto.commit روی true است، پیشفرض کلاینتهای consumer [19]. offsetها از طرف ما commit میشوند. یکی از بهترین بخشهای این گزینه این است که فراخوانی دیگری برای commit offsetهای مصرفشده لازم نیست.
brokerهای Kafka پیام را دوباره میفرستند اگر بهخاطر شکست کلاینت consumer بهصورت خودکار تأیید نشوند. اما چه دردسری ممکن است پیش بیاید؟ اگر پیامهایی که از آخرین poll میگیریم را در thread جدا پردازش کنیم، offset commit خودکار ممکن است بهعنوان خواندهشده علامت بخورد حتی اگر با آن offsetهای مشخص واقعاً تمام نشده باشد. اگر پیامی در پردازش ما شکست بخورد و نیاز به retry داشتیم چه؟ با poll بعدی، میتوانیم مجموعهٔ offset بعد از آنچه قبلاً بهعنوان مصرفشده commit شده را بگیریم [8]. ممکن است و آسان است پیامهایی که انگار مصرف شدهاند از دست بروند با وجود اینکه توسط منطق consumer پردازش نشدهاند.
وقتی به آنچه commit میکنید نگاه میکنید، توجه کنید زمانبندی ممکن است کامل نباشد. اگر متد commit روی consumer با metadata مشخص offset برای commit فراخوانی نکنید، ممکن است رفتار تعریفنشدهای بر اساس زمانبندی pollها، تایمرهای منقضیشده یا حتی منطق threading خودتان داشته باشید. اگر باید مطمئن شوید رکوردی در زمان مشخص هنگام پردازش یا offset مشخص commit شود، مطمئن شوید metadata offset را به متد commit میفرستید.
بیایید بیشتر با صحبت دربارهٔ commitهای مخصوص کد که با enable.auto.commit روی false فعال میشوند کاوش کنیم. این روش میتواند بیشترین مدیریت روی زمان واقعی مصرف پیام و commit آن توسط برنامه را اعمال کند. تضمینهای تحویل at-least-once با این الگو قابل دستیابیاند.
مثالی که پیام باعث ایجاد فایل در Hadoop در مکان مشخص میشود را در نظر بگیرید. وقتی پیام میگیرید، پیامی در offset ۹۹۹ poll میکنید. در حین پردازش، consumer بهخاطر خطا متوقف میشود. چون کد هرگز offset ۹۹۹ را commit نکرد، دفعهٔ بعد consumer همان گروه از آن پارتیشن شروع به خواندن کند، پیام offset ۹۹۹ را دوباره میگیرد. با دریافت دو بار پیام، کلاینت توانست کار را بدون از دست دادن پیام تکمیل کند. از طرف دیگر، پیام را دو بار گرفتید! اگر به هر دلیلی پردازش واقعاً کار کرد و نوشتن موفق بود، کد باید با احتمال duplicateها کنار بیاید.
حالا بخشی از کدی که برای کنترل offsetها استفاده میکنیم را ببینیم. همانطور که با producer هنگام ارسال پیام کردیم، میتوانیم offsetها را هم بهصورت همگام یا ناهمگام commit کنیم. لیستینگ ۵-۴ commit همگام را نشان میدهد. برای commitSync در آن لیستینگ، مهم است commit به شکلی انجام شود که هر پیشرفت دیگری در کد را تا موفقیت یا شکست block کند [20].
لیستینگ ۵-۴. منتظر ماندن برای commit
java
consumer.commitSync();
// commitSync waits for
#// Any code here will wait on line before
// a success or fail.مانند producerها، میتوانیم callback هم استفاده کنیم. لیستینگ ۵-۵ نحوهٔ ساخت commit ناهمگام با callback با پیادهسازی interface OffsetCommitCallback (متد onComplete) با lambda expression را نشان میدهد [21]. این نمونه اجازه میدهد پیامهای log برای تعیین موفقیت یا شکست داشته باشیم حتی اگر کد قبل از رفتن به دستور بعدی منتظر نماند.
لیستینگ ۵-۵. Commit با callback
java
public static void commitOffset(long offset,
int partition,
String topic,
KafkaConsumer<String, String> consumer) {
OffsetAndMetadata offsetMeta = new OffsetAndMetadata(++offset, "");
Map<TopicPartition, OffsetAndMetadata> kaOffsetMap = new HashMap<>();
kaOffsetMap.put(new TopicPartition(topic, partition), offsetMeta);
consumer.commitAsync(kaOffsetMap, (map, e) -> { // A lambda that creates an
if (e != null) { // OffsetCommitCallback instance
for (TopicPartition key : map.keySet()) {
log.info("kinaction_error: offset {}", map.get(key).offset());
}
} else {
for (TopicPartition key : map.keySet()) {
log.info("kinaction_info: offset {}", map.get(key).offset());
}
}
});
}اگر به فصل ۴ برگردید، این شبیه نحوهٔ استفاده از sendهای ناهمگام با callback برای acknowledgmentها است. برای پیادهسازی callback خودتان باید از interface OffsetCommitCallback استفاده کنید. میتوانید تعریف متد onComplete را برای مدیریت استثناها یا موفقیتها طبق نیاز تعریف کنید.
چرا الگوی commit همگام یا ناهمگام را انتخاب کنید؟ بهخاطر داشته باشید latency بالاتر است اگر منتظر فراخوانی blocking بمانید. این عامل زمان اگر نیازمندیها شامل consistency داده باشد ممکن است ارزش تأخیر را داشته باشد [21]. این تصمیمها به میزان کنترلی که برای اطلاع به Kafka کدام پیامها را منطق شما پردازششده میداند کمک میکنند.
۵.۵ خواندن از topic فشردهشده (compacted)
Consumerها باید از خواندن از topic compacted آگاه باشند. Kafka پارتیشن log را در فرآیند پسزمینه compact میکند و رکوردهایی با همان کلید ممکن است حذف شوند جز آخرین آنها. فصل ۷ بیشتر دربارهٔ نحوهٔ کار این topicها میرود، اما بهاختصار باید رکوردهایی با همان مقدار کلید را بهروز کنیم. اگر به تاریخچهٔ پیامها نیاز ندارید بلکه فقط آخرین مقدار را میخواهید، شاید بپرسید این مفهوم با log تغییرناپذیری که فقط رکورد به انتها اضافه میکند چگونه کار میکند. بزرگترین «gotcha» برای consumerهایی که ممکن است خطا ایجاد کند این است که هنگام خواندن رکوردها از topic compacted، consumerها هنوز میتوانند چند ورودی برای یک کلید بگیرند [22]! چگونه ممکن است؟ چون compaction روی فایلهای log روی دیسک اجرا میشود، ممکن است هر پیامی که در حافظه وجود دارد را هنگام پاکسازی نبیند.
کلاینتها باید این مورد را که بیش از یک مقدار برای هر کلید وجود دارد مدیریت کنند و در صورت نیاز منطقی داشته باشند که همه را نادیده بگیرند جز آخرین مقدار. برای علاقهمند کردن شما به topicهای compacted، توجه کنید Kafka topic داخلی compacted خودش به نام __consumer_offsets را استفاده میکند که مستقیماً به offsetهای consumer شما مربوط است [23]. compaction اینجا منطقی است چون برای ترکیب مشخص consumer group، پارتیشن و topic، فقط آخرین مقدار لازم است چون آخرین offset مصرفشده را دارد.
۵.۶ بازیابی کد برای نیازمندیهای کارخانه
بیایید اطلاعاتی که دربارهٔ نحوهٔ کار consumerها جمع کردیم بهکار ببریم و ببینیم آیا میتوانیم روی راهحلهای طراحیشده در فصل ۳ برای استفاده با Kafka در کارخانهٔ e-bike از دیدگاه کلاینت consumer کار کنیم. همانطور که در فصل ۳ گفتیم، میخواهیم هیچ پیام auditای هنگام تکمیل دستورات اپراتورها روی سنسورها از دست نرود. ابتدا گزینههای خواندن offsetها را ببینیم.
۵.۶.۱ گزینههای خواندن
اگرچه گزینهٔ جستجوی پیام با کلید در Kafka نیست، میتوان به offset مشخص seek کرد. با فکر کردن به log پیامها بهعنوان آرایهٔ همیشه افزایشی که هر پیام index دارد، چند گزینه داریم از جمله شروع از ابتدا، رفتن به انتها، یا یافتن offset بر اساس زمانهای مشخص. این گزینهها را ببینیم.
مسئلهای که ممکن است پیش بیاید این است که میخواهیم از ابتدای topic بخوانیم حتی اگر قبلاً این کار را کردهایم. دلایل میتواند شامل خطاهای منطق و تمایل به replay کل log یا شکست در خط لولهٔ داده پس از شروع با Kafka باشد. پیکربندی مهم برای این رفتار تنظیم auto.offset.reset روی earliest است [24]. تکنیک دیگر اجرای همان منطق با group ID متفاوت است. در اثر، یعنی topicهای commit offset که Kafka داخلاً استفاده میکند نمیتوانند مقدار offset پیدا کنند اما میتوانند از اولین index یافتشده شروع کنند چون topic commit offset برای consumer group جدید دادهای ندارد.
لیستینگ ۵-۶ مثالی از تنظیم property auto.offset.reset روی "earliest" برای seek به offset مشخص است [24]. تنظیم group ID روی UUID تصادفی هم برای شروع بدون تاریخچه offset برای consumer group کمک میکند. این نوع reset را میتوانیم برای نگاه به kinaction_alerttrend با منطق کد متفاوت برای تعیین روندها روی همهٔ دادهٔ آن topic استفاده کنیم.
لیستینگ ۵-۶. Earliest offset
java
Properties kaProperties = new Properties(); // Creates a group ID for
kaProperties.put("group.id", // which Kafka does not
UUID.randomUUID().toString()); // have a stored offset
kaProperties.put("auto.offset.reset", "earliest"); // Uses the earliest offset
// retained in our logsگاهی فقط میخواهید منطق را از زمان راهاندازی consumer شروع کنید و پیامهای گذشته را فراموش کنید [24]. شاید داده برای ارزش کسبوکار در topic شما خیلی قدیمی شده. لیستینگ ۵-۷ propertyهایی را که برای این رفتار شروع با latest offset تنظیم میکنید نشان میدهد. اگر میخواهید مطمئن شوید offset consumer قبلی پیدا نمیشود و بهجای آن به latest offset Kafka برای subscriptionهایتان پیشفرض میشود، استفاده از UUID جز برای تست لازم نیست. اگر فقط به alertهای جدید ورودی به topic kinaction_alert علاقهمندیم، این میتواند راهی برای consumer باشد که فقط آن alertها را ببیند.
لیستینگ ۵-۷. Latest offset
java
Properties kaProperties = new Properties(); // Creates a group ID for
kaProperties.put("group.id", // which Kafka does not
UUID.randomUUID().toString()); // have a stored offset
kaProperties.put("auto.offset.reset", "latest"); // Uses the latest
// record offsetیکی از روشهای جستجوی offset پیچیدهتر offsetsForTimes است. این متد به شما اجازه میدهد mapی از topicها و پارتیشنها و timestamp برای هر کدام بفرستید تا mapی از offset و timestamp برای topicها و پارتیشنهای دادهشده برگردانید [25]. در موقعیتهایی که offset منطقی مشخص نیست اما timestamp مشخص است مفید است. مثلاً اگر استثنایی مرتبط با رویدادی که log شده دارید، شاید بتوانید از consumer برای تعیین دادهای که حدود timestamp مشخص پردازش شده استفاده کنید. تلاش برای یافتن رویداد audit با زمان ممکن است برای topic kinaction_audit ما برای یافتن دستورات در حال وقوع استفاده شود.
همانطور که لیستینگ ۵-۸ نشان میدهد، میتوانیم offset و timestamp را برای هر topic یا پارتیشن وقتی هر کدام را به timestamp map میکنیم بازیابی کنیم. پس از گرفتن map metadata برگشتی از فراخوانی offsetsForTimes، میتوانیم مستقیماً به offset مورد علاقه seek کنیم با seek به offset برگشتی برای هر کلید مربوطه.
لیستینگ ۵-۸. Seek به offset با timestampها
java
...
Map<TopicPartition, OffsetAndTimestamp> kaOffsetMap =
consumer.offsetsForTimes(timeStampMapper); // Finds the first offset greater or
... // equal to that timeStampMapper
// We need to use the map we get
consumer.seek(partitionOne,
kaOffsetMap.get(partitionOne).offset()); // Seeks to the first offset
// provided in kaOffsetMapیک چیز که باید بدانید offset برگشتی اولین پیام با timestampای است که معیارهای شما را برآورده میکند. اما بهخاطر resend پیام توسط producer در شکستها یا تفاوت در زمان افزودن timestamp (شاید توسط consumerها)، زمانها ممکن است نامرتب به نظر برسند.
Kafka هم به شما امکان یافتن offsetهای دیگر را میدهد همانطور که در Javadoc consumer قابل ارجاع است [26]. با همهٔ این گزینهها، ببینیم چگونه به use case ما میخورند.
۵.۶.۲ نیازمندیها
یکی از نیازمندیهای مثال audit ما این بود که نیازی به همبستگی (یا گروهبندی) رویدادها در رویدادهای فردی نیست. یعنی نگرانی ترتیب یا خواندن از پارتیشنهای مشخص نیست؛ هر consumer که هر پارتیشنی را بخواند خوب است. نیازمندی دیگر عدم از دست رفتن پیام بود. راه امن برای اطمینان از اجرای منطق برای هر رویداد audit، commit صریح offset برای هر رکورد پس از مصرف است. برای کنترل commit بهعنوان بخشی از کد، میتوانیم enable.auto.commit را روی false بگذاریم.
لیستینگ ۵-۹ مثالی از بهرهگیری از commit همگام پس از پردازش هر رکورد برای ویژگی audit است. جزئیات offset بعدی برای مصرف در رابطه با topic و پارتیشن offsetی که تازه مصرف شد بهعنوان بخشی از هر حلقه در رکوردها فرستاده میشود. یک «gotcha» که باید توجه کنید این است که ممکن است عجیب به نظر برسد ۱ به offset فعلی اضافه میکنیم، اما offset ارسالی به broker باید index آیندهٔ شما باشد. متد commitSync فراخوانی و map offset حاوی offset رکورد تازه پردازششده به آن پاس داده میشود [20].
لیستینگ ۵-۹. منطق audit consumer
java
...
kaProperties.put("enable.auto.commit", "false"); // Sets autocommit
// to false
try (KafkaConsumer<String, String> consumer =
new KafkaConsumer<>(kaProperties)) {
consumer.subscribe(List.of("kinaction_audit"));
while (keepConsuming) {
var records = consumer.poll(Duration.ofMillis(250));
for (ConsumerRecord<String, String> record : records) {
// audit record process ...
OffsetAndMetadata offsetMeta =
// Adding a record to the
new OffsetAndMetadata(++record.offset(), ""); // current offset determines
// the next offset to read.
Map<TopicPartition, OffsetAndMetadata> kaOffsetMap =
new HashMap<>();
kaOffsetMap.put(
new TopicPartition("kinaction_audit", // Allows for a topic
record.partition()), offsetMeta); // and partition key
// to be related to a
consumer.commitSync(kaOffsetMap); // Commits // specific offset
} // the offsets
}
}
...هدف دیگر طراحی کارخانهٔ e-bike، ثبت وضعیت alert و مانیتورینگ روند alert در طول زمان بود. اگرچه میدانیم رکوردها کلیدی دارند که stage ID است، نیازی به مصرف یک گروه در هر زمان یا نگرانی ترتیب نیست. لیستینگ ۵-۱۰ نحوهٔ تنظیم property key.deserializer را نشان میدهد تا consumer بداند چگونه با دادهٔ دودویی که هنگام تولید پیام در Kafka ذخیره شد برخورد کند. در این مثال AlertKeySerde برای deserialize کلید استفاده میشود. چون از دست رفتن پیام در سناریوی ما نگرانی بزرگی نیست، اجازهٔ autocommit پیامها در این موقعیت کافی است.
لیستینگ ۵-۱۰. Alert trending consumer
java
... // Uses autocommit as lost
kaProperties.put("enable.auto.commit", "true"); // messages are not an issue
kaProperties.put("key.deserializer",
AlertKeySerde.class.getName()); // AlertKeySerde
kaProperties.put("value.deserializer", // key deserializer
"org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<Alert, String> consumer =
new KafkaConsumer<Alert, String>(kaProperties);
consumer.subscribe(List.of("kinaction_alerttrend"));
while (true) {
ConsumerRecords<Alert, String> records =
consumer.poll(Duration.ofMillis(250));
for (ConsumerRecord<Alert, String> record : records) {
// ...
}
}
...نیازمندی بزرگ دیگر پردازش سریع alertها برای اطلاعرسانی به اپراتورها دربارهٔ مشکلات بحرانی است. چون producer در فصل ۴ از Partitioner سفارشی استفاده کرد، consumer را مستقیماً به همان پارتیشن اختصاص میدهیم تا به مشکلات بحرانی هشدار دهیم. چون تأخیر در مورد alertهای دیگر مطلوب نیست، commit برای هر offset بهصورت ناهمگام خواهد بود.
لیستینگ ۵-۱۱ منطق کلاینت consumer متمرکز روی alertهای بحرانی را نشان میدهد که خودشان را به topic و پارتیشن مشخصی که برای تولید alertها هنگام استفاده از کلاس partitioner سفارشی AlertLevelPartitioner استفاده میشود اختصاص میدهند. در این مورد پارتیشن ۰ و topic kinaction_alert است.
از اشیاء TopicPartition استفاده میکنیم تا به Kafka بگوییم به کدام پارتیشنهای مشخص topic علاقهمندیم. پاس دادن اشیاء TopicPartition به متد assign جایگزین اجازه دادن به consumer برای تخصیص group coordinator میشود [27].
برای لیستینگ ۵-۱۱، برای هر رکورد برگشتی از poll consumer، commit ناهمگام با callback استفاده میشود. commit offset بعدی برای مصرف به broker فرستاده میشود و نباید consumer را از پردازش رکورد بعدی block کند، طبق نیازمندیهای ما. گزینههای لیستینگ بعدی به نظر نیازمندیهای طراحی اصلی فصل ۳ را برآورده میکنند.
لیستینگ ۵-۱۱. Alert consumer
java
kaProperties.put("enable.auto.commit", "false");
KafkaConsumer<Alert, String> consumer =
new KafkaConsumer<Alert, String>(kaProperties);
TopicPartition partitionZero = // Uses TopicPartition
new TopicPartition("kinaction_alert", 0); // for critical messages
consumer.assign(List.of(partitionZero)); // Consumer assigns itself
// the partition rather than
while (true) { // subscribing to the topic
ConsumerRecords<Alert, String> records =
consumer.poll(Duration.ofMillis(250));
for (ConsumerRecord<Alert, String> record : records) {
// ...
commitOffset(record.offset(),
record.partition(), topicName, consumer); // Commits each record
} // asynchronously
}
...
public static void commitOffset(long offset,int part, String topic,
KafkaConsumer<Alert, String> consumer) {
OffsetAndMetadata offsetMeta = new OffsetAndMetadata(++offset, "");
Map<TopicPartition, OffsetAndMetadata> kaOffsetMap =
new HashMap<TopicPartition, OffsetAndMetadata>();
kaOffsetMap.put(new TopicPartition(topic, part), offsetMeta);
OffsetCommitCallback callback = new OffsetCommitCallback() {
...
};
consumer.commitAsync(kaOffsetMap, callback); // The asynchronous commit
} // uses the kaOffsetMap and
// callback arguments.بهطور کلی، consumer میتواند بخش پیچیدهای از تعاملات ما با Kafka باشد. برخی گزینهها فقط با پیکربندی property انجام میشوند، اما اگر نه، میتوانید با دانش topicها، پارتیشنها و offsetها به دادهای که نیاز دارید برسید.
خلاصه
- کلاینتهای consumer به توسعهدهندگان راهی برای بیرون آوردن داده از Kafka میدهند. مانند کلاینتهای producer، تعداد زیادی گزینهٔ پیکربندی قابل تنظیم بهجای کدنویسی سفارشی در دسترس است.
- Consumer groupها به بیش از یک کلاینت اجازه میدهند بهعنوان گروه برای پردازش رکوردها کار کنند. با گروهبندی، کلاینتها میتوانند داده را بهصورت موازی پردازش کنند.
- Offsetها موقعیت رکورد در commit log روی broker را نشان میدهند. با استفاده از offsetها، consumerها کنترل میکنند از کجا شروع به خواندن داده کنند.
- Offset میتواند offset قبلی باشد که consumerها قبلاً دیدهاند، که توانایی replay رکوردها را به ما میدهد.
- Consumerها میتوانند داده را بهصورت همگام یا ناهمگام بخوانند.
- اگر از متدهای ناهمگام استفاده شود، consumer میتواند کد در callbackها برای اجرای منطق پس از دریافت داده استفاده کند.
منابع
- S. Kozlovski. "Apache Kafka Data Access Semantics: Consumers and Membership." Confluent blog (n.d.). https://www.confluent.io/blog/apache-kafka-data-access-semantics-consumers-and-membership (accessed August 20, 2021).
- "Consumer Configurations." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html (accessed June 19, 2019).
- N. Narkhede. "Apache Kafka Hits 1.1 Trillion Messages Per Day – Joins the 4 Comma Club." Confluent blog (September 1, 2015). https://www.confluent.io/blog/apache-kafka-hits-1-1-trillion-messages-per-day-joins-the-4-comma-club/ (accessed October 20, 2019).
- "Class KafkaConsumer<K,V>." Kafka 2.7.0 API. Apache Software Foundation (n.d.). https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#poll-java.time.Duration- (accessed August 24, 2021).
- "Class WakeupException." Kafka 2.7.0 API. Apache Software Foundation (n.d.). https://kafka.apache.org/27/javadoc/org/apache/kafka/common/errors/WakeupException.html (accessed June 22, 2020).
- "Documentation: Topics and Logs." Confluent documentation (n.d.). https://docs.confluent.io/5.5.1/kafka/introduction.html#topics-and-logs (accessed October 20, 2021).
- "KIP-392: Allow consumers to fetch from closest replica." Wiki for Apache Kafka. Apache Software Foundation (November 05, 2019). https://cwiki.apache.org/confluence/display/KAFKA/KIP-392%3A+Allow+consumers+to+fetch+from+closest+replica (accessed December 10, 2019).
- J. Gustafson. "Introducing the Kafka Consumer: Getting Started with the New Apache Kafka 0.9 Consumer Client." Confluent blog (January 21, 2016). https://www.confluent.io/blog/tutorial-getting-started-with-the-new-apache-kafka-0-9-consumer-client/ (accessed June 01, 2020).
- J. Rao. "How to choose the number of topics/partitions in a Kafka cluster?" Confluent blog (March 12, 2015). https://www.confluent.io/blog/how-choose-number-topics-partitions-kafka-cluster/ (accessed May 19, 2019).
- "Committing and fetching consumer offsets in Kafka." Wiki for Apache Kafka. Apache Software Foundation (March 24, 2015). https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=48202031 (accessed December 15, 2019).
- "Consumer Configurations: group.id." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#consumerconfigs_group.id (accessed May 11, 2018).
- "Documentation: Consumers." Apache Software Foundation (n.d.). https://kafka.apache.org/23/documentation.html#intro_consumers (accessed December 11, 2019).
- "Consumer Configurations: heartbeat.interval.ms." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#consumerconfigs_heartbeat.interval.ms (accessed May 11, 2018).
- "Consumer Configurations: partition.assignment.strategy." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#consumerconfigs_partition.assignment.strategy (accessed December 22, 2020).
- S. Blee-Goldman. "From Eager to Smarter in Apache Kafka Consumer Rebalances." Confluent blog (n.d.). https://www.confluent.io/blog/cooperative-rebalancing-in-kafka-streams-consumer-ksqldb/ (accessed August 20, 2021).
- "RangeAssignor.java." Apache Kafka GitHub (n.d.). https://github.com/apache/kafka/blob/c9708387bb1dd1fd068d6d8cec2394098d5d6b9f/clients/src/main/java/org/apache/kafka/clients/consumer/RangeAssignor.java (accessed August 25, 2021).
- A. Li. "What I have learned from Kafka partition assignment strategy." Medium (December 1, 2017). https://medium.com/@anyili0928/what-i-have-learned-from-kafka-partition-assignment-strategy-799fdf15d3ab (accessed October 20, 2021).
- "Release Plan 0.11.0.0." Wiki for Apache Kafka. Apache Software Foundation (June 26, 2017). https://cwiki.apache.org/confluence/display/KAFKA/Release+Plan+0.11.0.0 (accessed December 14, 2019).
- "Consumer Configurations: enable.auto.commit." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#consumerconfigs_enable.auto.commit (accessed May 11, 2018).
- Synchronous Commits. Confluent documentation (n.d.). https://docs.confluent.io/3.0.0/clients/consumer.html#synchronous-commits (accessed August 24, 2021).
- Asynchronous Commits. Confluent documentation (n.d.). https://docs.confluent.io/3.0.0/clients/consumer.html#asynchronous-commits (accessed August 24, 2021).
- Kafka Design. Confluent documentation (n.d.). https://docs.confluent.io/platform/current/kafka/design.html (accessed August 24, 2021).
- Kafka Consumers. Confluent documentation (n.d.). https://docs.confluent.io/3.0.0/clients/consumer.html (accessed August 24, 2021).
- "Consumer Configurations: auto.offset.reset." Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#consumerconfigs_auto.offset.reset (accessed May 11, 2018).
- offsetsForTimes. Kafka 2.7.0 API. Apache Software Foundation (n.d.). https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/consumer/Consumer.html#offsetsForTimes-java.util.Map- (accessed June 22, 2020).
- seek. Kafka 2.7.0 API. Apache Software Foundation (n.d.). https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/consumer/Consumer.html#seek-org.apache.kafka.common.TopicPartition-long- (accessed June 22, 2020).
- assign. Kafka 2.7.0 API. Apache Software Foundation (n.d.). https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#assign-java.util.Collection- (accessed August 24, 2021).