در چشمانداز مدرن میکروسرویسها و سیستمهای توزیعشده، توانایی مدیریت کارآمد جریانهای داده بلادرنگ نه یک تجمل، بلکه یک ضرورت است. Apache Kafka به عنوان استاندارد پیشفرض برای ساخت معماریهای رویداد-محور ظهور کرده است و عملکرد بالا، تحمل خطا و مقیاسپذیری را ارائه میدهد. با این حال، برای توسعهدهندگانی که از مدلهای متمرکز بر پایگاه داده سنتی به پردازش جریان مهاجرت میکنند، منحنی یادگیری میتواند شیبدار باشد. این راهنما شما را با مفاهیم اصلی Kafka آشنا کرده و نحوه پیادهسازی تولیدکنندگان و مصرفکنندگان را با استفاده از کلاینت جاوای Confluent Kafka که به طور گسترده استفاده میشود، نشان میدهد.
درک انتزاعات اصلی
قبل از غرق شدن در کد، درک بلوکهای سازنده بنیادین Kafka حیاتی است. برخلاف یک صف سنتی که مصرف و دور ریخته میشود، Kafka پیامها را برای یک دوره قابل پیکربندی حفظ میکند و به چندین سرویس اجازه میدهد دادههای یکسان را با سرعت خودشان بخوانند.
انتزاعات اصلی عبارتند از:
- موضوع (Topic): نام یک دسته یا جریان که رکوردها به آن منتشر میشوند.
- پارتیشن (Partition): موضوعات برای امکان موازیسازی و مقیاسپذیری به پارتیشنها تقسیم میشوند. هر پارتیشن یک دنباله مرتب و غیرقابل تغییر از رکوردها است.
- تولیدکننده (Producer): برنامهای که رکوردها را به موضوعات Kafka منتشر (مینویسد) میکند.
- مصرفکننده (Consumer): برنامهای که به موضوعات مشترکالمنفع (اشتراک) میشود (میخواند) و رکوردها را پردازش میکند.
- گروه مصرفکننده (Consumer Group): گروهی از مصرفکنندگان که به طور جمعی تمام پارتیشنهای یک موضوع را میخوانند و امکان تعادل بار و پردازش موازی را فراهم میکنند.
راهاندازی محیط
برای دنبال کردن مثالهای عملی در این پست، شما به یک برودکر Kafka در حال اجرا و کتابخانه کلاینت Confluent Kafka نیاز دارید. اطمینان حاصل کنید که Java 11 یا بالاتر نصب شده است. اگر از Maven استفاده میکنید، وابستگی زیر را به
pom.xml خود اضافه کنید:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.4.0</version>
</dependency>
ساخت یک تولیدکننده
تولیدکننده مسئول ارسال دادهها به Kafka است. کلیدها و مقادیر را به بایتها سریالسازی میکند و تعیین میکند که به کدام پارتیشن بنویسد. در زیر یک مثال مینیمال از یک تولیدکننده Kafka وجود دارد که یک پیام متنی ساده را به موضوعی به نام
test-topic ارسال میکند.
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class KafkaProducerExample {
public static void main(String[] args) throws Exception {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
try (Producer<String, String> producer = new KafkaProducer<>(props)) {
ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "key1", "Hello Kafka!");
producer.send(record, (metadata, exception) -> {
if (exception != null) {
System.out.println("Error sending message: " + exception.getMessage());
} else {
System.out.println("Sent to partition: " + metadata.partition() +
", offset: " + metadata.offset());
}
});
producer.flush();
}
}
}
به استفاده از
flush() توجه کنید تا اطمینان حاصل شود که تمام پیامهای بافر شده قبل از خروج برنامه ارسال میشوند. در یک محیط تولید، شما معمولاً فراخوانیهای ناهمگام را به صورت قویتری مدیریت کرده و چرخه عمر تولیدکننده را به عنوان یک تکنمونه (singleton) مدیریت میکنید.
ساخت یک مصرفکننده
مصرفکنندگان کمی پیچیدهتر هستند زیرا باید با آفستها مدیریت کنند و خطاهای احتمالی پردازش پیام را مدیریت کنند. مصرفکننده به یک موضوع مشترکالمنفع میشود و در یک حلقه برای پیامهای جدید پوول (Poll) میکند.
import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class KafkaConsumerExample {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
try (Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Collections.singletonList("test-topic"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("Consumed record: key=%s, value=%s, partition=%d%n",
record.key(), record.value(), record.partition());
}
}
}
}
}
بهترین شیوهها برای محیط تولید
در حالی که مثالهای بالا کاربردی هستند، سیستمهای تولید به ملاحظات اضافی نیاز دارند. همیشه از
acks=all در پیکربندی تولیدکننده خود استفاده کنید تا اطمینان حاصل شود که پیامها به تمام_replicas_ همگامشده پایدار میشوند و دوام را تضمین میکنند. برای مصرفکنندگان، در نظر بگیرید که کامیت خودکار را غیرفعال کنید و فقط پس از پردازش موفقیتآمیز پیام، آفستها را به صورت دستی کامیت کنید تا به معنای
دقیقاً یک بار (exactly-once) دست یابید یا از از دست رفتن دادهها در طول خرابیها جلوگیری کنید. علاوه بر این، همیشه منطق تلاش مجدد (Retry) و شکستدهنده مدار (Circuit Breaker) را پیادهسازی کنید تا شکستهای موقت شبکه را به آرامی مدیریت کنید.
نتیجهگیری
Apache Kafka یک ابزار قدرتمند است که وقتی به درستی درک شود، میتواند نحوه مدیریت دادهها توسط برنامههای شما را متحول کند. با تسلط بر الگوی تولیدکننده-مصرفکننده و درک استراتژیهای پارتیشنبندی، میتوانید سیستمهای مقاوم، مقیاسپذیر و جدا از هم (Decoupled) بسازید. همانطور که به پیش میروید، ویژگیهای پیشرفتهتری مانند Kafka Streams برای تحلیلهای بلادرنگ و Schema Registry برای مدیریت قراردادهای داده را برای تثبیت بیشتر معماری رویداد-محور خود کاوش کنید.