پیاده‌سازی معنای دقیقاً یک‌بار در آپاچی کافکا: یک راهنمای عملی

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

درک مسئله

در سیستم‌های توزیع‌شده، تکه‌تکه شدن شبکه (Network Partitions)، کرش مصرف‌کنندگان و تلاش مجدد تولیدکنندگان می‌تواند منجر به پیام‌های تکراری یا از دست رفتن داده شود. بدون محافظت‌های مناسب، یک مصرف‌کننده ممکن است یک رکورد را دو بار پردازش کند که منجر به ناسازگاری وضعیت می‌شود. برای مثال، اگر یک برنامه بانکی دو پیام یکسان «کسر ۱۰۰ دلار» دریافت کند، موجودی حساب کاربر به‌درستی ۲۰۰ دلار کاهش خواهد یافت. معنای دقیقاً یک‌بار این ابهام را با اطمینان از اینکه هر رکورد دقیقاً یک‌بار به وضعیت پایین‌دست اعمال می‌شود، برطرف می‌کند.

اجزای کلیدی: تولیدکنندگان ایده‌مپوتنت و تراکنش‌ها

برای دستیابی به معنای دقیقاً یک‌بار، شما به دو مکانیزم اصلی نیاز دارید که با هم کار می‌کنند:

1. تولیدکننده ایده‌مپوتنت: اطمینان حاصل می‌کند که چندین نمونه از یک پیام تولید نمی‌شوند. بروکرهای کافکا شناسه‌های تولیدکننده و شماره‌های ترتیبی را ردیابی می‌کنند تا پیام‌های ارسالی از یک جلسه تولیدکننده را منحصربه‌فرد کنند.
2. تراکنش‌ها: به شما امکان می‌دهند چندین عملیات تولید و مصرف را در یک واحد اتمیک کار گروه‌بندی کنید. اگر هر بخش از تراکنش شکست بخورد، کل تراکنش بازگردانی (Rollback) می‌شود.

پیاده‌سازی یک تولیدکننده ایده‌مپوتنت

اولین کار، فعال‌سازی ایده‌مپوتنت در پیکربندی تولیدکننده کافکای شماست. این کار با تنظیم enable.idempotence=true انجام می‌شود. توجه داشته باشید که وقتی ایده‌مپوتنت فعال است، acks باید روی all تنظیم شود تا اطمینان حاصل شود که پیام‌ها قبل از تأیید، به همه کپی‌های همگام (In-sync Replicas) تعهد می‌شوند.

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class IdempotentProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        
        // Enable idempotence
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        KafkaProducer producer = new KafkaProducer<>(props);
        // Send messages...
        producer.close();
    }
}

استفاده از تراکنش‌ها برای پردازش دقیقاً یک‌بار

برای پردازش دقیقاً یک‌بار از مبدأ تا مقصد (End-to-End)، باید یک تولیدکننده ایده‌مپوتنت را با یک مصرف‌کننده که در همان تراکنش مشارکت دارد، ترکیب کنید. نکته کلیدی در اینجا استفاده از پیکربندی transactional.id برای پیوند دادن گروه مصرف‌کننده به یک جلسه تراکنشی خاص است.

گام ۱: پیکربندی تولیدکننده با شناسه تراکنشی

props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id");
producer.initTransactions();

گام ۲: شروع تراکنش، پردازش داده و تعهد (Commit)

producer.beginTransaction();

// Consume messages
ConsumerRecords records = consumer.poll(Duration.ofMillis(100));

for (ConsumerRecord record : records) {
    // Process the record (e.g., update external system)
    process(record);
    
    // Produce the result to a downstream topic
    producer.send(new ProducerRecord<>("output-topic", record.key(), record.value()));
}

// Commit the transaction atomically
producer.commitTransaction();

// Offset commit is optional here as Kafka manages offsets via transaction log
consumer.commitSync(); 

ملاحظات عملی

• تایم‌اوت‌ها: مقادیر مناسبی را برای transaction.timeout.ms و session.timeout.ms تنظیم کنید تا از انقضای زودهنگام تراکنش جلوگیری شود.
• هماهنگی گروه مصرف‌کننده: در سناریوهای مقیاس‌پذیری (Scale-out)، باید از همان transactional.id در تمام نمونه‌های جفت تولیدکننده-مصرف‌کننده استفاده شود. کافکا اطمینان حاصل می‌کند که در هر زمان فقط یک نمونه برای هر شناسه تراکنشی فعال است.
• سیستم‌های پایین‌دست: معنای دقیقاً یک‌بار در کافکا فقط تضمین می‌کند که پیام یک‌بار در کافکا نوشته می‌شود. اگر سیستم پایین‌دست شما (مثلاً یک پایگاه داده) ایده‌مپوتنت نباشد، باید منطق منحصربه‌فردسازی (Deduplication) را در سطح مصرف‌کننده نیز پیاده‌سازی کنید.

نتیجه‌گیری

پیاده‌سازی معنای دقیقاً یک‌بار در آپاچی کافکا نیازمند ترکیب دقیق تولیدکنندگان ایده‌مپوتنت، APIهای تراکنشی و پیکربندی مناسب است. با استفاده از پشتیبانی داخلی تراکنشی کافکا، می‌توانید برنامه‌های استریمینگ مقاوم و تحمل‌کننده خطا بسازید که حتی در محیط‌های بسیار توزیع‌شده، یکپارچگی داده را حفظ می‌کنند. اگرچه راه‌اندازی این سیستم پیچیده‌تر از تحویل ساده حداقل یک‌بار است، اما اطمینان از یکپارچگی داده آن را به یک سرمایه‌گذاری ارزشمند برای سیستم‌های حیاتی تبدیل می‌کند.