Series Docker & Kafka cho Laravel #2: Nghệ thuật Phát/Hứng sự kiện và Chống rò rỉ bộ nhớ (Memory Leak)

1. The Producer: Máy phát tín hiệu siêu tốc

Khi máy trạm FAM/TVM gửi API báo cáo một lượt quẹt thẻ (Tap In), Controller của chúng ta sẽ đóng gói dữ liệu và bắn thẳng vào một "Chủ đề" (Topic) của Kafka mang tên metro.transactions.

Chuẩn Senior (Bảo đảm thứ tự - Ordering): Trong môi trường phân tán, nếu một hành khách Quẹt vào (Tap In) rồi Quẹt ra (Tap Out) quá nhanh, 2 sự kiện này có thể bị Kafka xử lý đan chéo nếu chúng ta ném bừa bãi. Để đảm bảo mọi sự kiện của cùng một thẻ phải được xử lý theo đúng thứ tự thời gian, chúng ta phải gán Kafka Key (chìa khóa) bằng chính mã thẻ (card_uid). Kafka sẽ tự động điều hướng tất cả sự kiện có cùng Key vào chung một Partition (vùng nhớ), đảm bảo thứ tự tuyệt đối.

Hãy viết logic bắn sự kiện trong Controller:

namespace App\Http\Controllers;

use Illuminate\Http\Request;
use Junges\Kafka\Facades\Kafka;
use Junges\Kafka\Message\Message;

class TransactionController extends Controller
{
    public function processTapIn(Request $request)
    {
        $payload = $request->validate([
            'card_uid' => 'required|string',
            'station_id' => 'required|string',
            'timestamp' => 'required|date',
        ]);

        // Đóng gói thông điệp
        $message = new Message(
            headers: ['event_type' => 'tap_in'],
            body: $payload,
            key: $payload['card_uid'] // CHÌA KHÓA QUAN TRỌNG: Đảm bảo thứ tự cho từng thẻ
        );

        // Bắn vào Topic 'metro.transactions'
        Kafka::publishOn('metro.transactions')
            ->withMessage($message)
            // acks = 'all': Yêu cầu Kafka xác nhận đã lưu an toàn ra ổ cứng mới báo thành công
            ->withConfigOption('acks', 'all') 
            ->send();

        // Lập tức trả về Response mở cổng (chưa tới 10ms)
        return response()->json(['status' => 'success', 'open_gate' => true]);
    }
}

2. The Consumer: Công nhân xử lý 24/7

Sau khi sự kiện đã nằm gọn trong Kafka, chúng ta cần một Consumer (hay Worker) để lấy nó ra xử lý. Khác với Redis Queue dùng câu lệnh php artisan queue:work mặc định, Consumer của Kafka là một tiến trình PHP vòng lặp vô tận (Infinite Loop) liên tục "lắng nghe" (poll) mạng lưới.

Đầu tiên, tạo một Class Handler để định nghĩa việc cần làm khi nhận được thông điệp:

php artisan make:class KafkaHandlers/TransactionSyncHandler

namespace App\KafkaHandlers;

use Junges\Kafka\Contracts\KafkaConsumerMessage;
use Illuminate\Support\Facades\Log;

class TransactionSyncHandler
{
    public function __invoke(KafkaConsumerMessage $message)
    {
        $payload = $message->getBody();
        $cardUid = $message->getKey();

        Log::info("Đang xử lý trừ tiền và đồng bộ Hitachi cho thẻ: " . $cardUid);

        try {
            // Xử lý logic trừ tiền vào Database
            // Gọi API đồng bộ sang hệ thống đối tác
            // ...
            
        } catch (\Exception $e) {
            Log::error("Lỗi đồng bộ thẻ {$cardUid}: " . $e->getMessage());
            // Ở bài sau chúng ta sẽ ném nó vào Dead Letter Queue
        }
    }
}

Tiếp theo, tạo một Command (Lệnh Artisan) để khởi chạy Consumer này. Lệnh này sẽ được cấu hình chạy ngầm bởi Supervisor trên Server Docker.

php artisan make:command StartTransactionConsumer

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Junges\Kafka\Facades\Kafka;
use App\KafkaHandlers\TransactionSyncHandler;

class StartTransactionConsumer extends Command
{
    protected $signature = 'kafka:consume-transactions';
    protected $description = 'Chạy Worker hứng sự kiện giao dịch vé';

    public function handle()
    {
        $this->info("Bắt đầu lắng nghe Topic: metro.transactions...");

        $consumer = Kafka::createConsumer(['metro.transactions'])
            // CỰC KỲ QUAN TRỌNG: Group ID. Nếu bạn chạy 5 tiến trình Command này cùng lúc 
            // có chung Group ID, Kafka sẽ tự động chia đều tải (load balance) cho 5 tiến trình.
            ->withConsumerGroupId('hitachi_sync_group')
            ->withHandler(new TransactionSyncHandler())
            // TẮT Auto Commit: Chỉ xác nhận hoàn thành khi logic PHP không bị ném lỗi
            ->withAutoCommit(false) 
            ->withMaxMessages(1000) // Đòn bẩy chống rò rỉ bộ nhớ
            ->build();

        $consumer->consume();
    }
}

3. Bí kíp sinh tồn của Senior: Auto Commit và Memory Leak

Đoạn mã Command trên chứa 2 thiết lập sinh tử phân biệt ranh giới giữa một dự án chạy cho vui và một hệ thống Enterprise:

A. Kéo cần gạt withAutoCommit(false):

Mặc định Kafka sẽ bật tính năng Auto Commit. Nghĩa là cứ đưa Message cho PHP xong là Kafka đánh dấu "Đã xử lý" (Đã dời Offset đi tới tin nhắn tiếp theo). Hậu quả: Nếu PHP đang chạy nửa chừng bị lỗi Database sập, cái tin nhắn đó bị mất vĩnh viễn, Kafka đã quên nó rồi. Giải pháp: Tắt Auto Commit. Thư viện laravel-kafka sẽ tự động commit (xác nhận) CHỈ KHI hàm __invoke() trong Handler chạy thành công và không bị văng Exception. Nếu lỗi, nó sẽ chạy lại tin nhắn đó hoặc đưa vào hàng đợi lỗi.

B. Tuyệt chiêu chống tràn RAM (withMaxMessages(1000)):

PHP vốn sinh ra để chạy theo Request (chạy xong xóa bộ nhớ ngay). Nó không được thiết kế để chạy một vòng lặp while(true) hàng tuần liền. Các truy vấn Eloquent, Log, Array trong Laravel sẽ dần dần tích tụ rác (Memory Leak) cho đến khi RAM bị vắt kiệt và Container bị Docker "Kill" (OOM Killed).

Tham số withMaxMessages(1000) chỉ thị Consumer: Hãy xử lý đúng 1000 sự kiện, sau đó TỰ SÁT (thoát chương trình). Khi chương trình tắt, toàn bộ rác RAM của PHP được giải phóng sạch sẽ. Ngay lập tức, tiến trình Supervisor (chạy nền trong Docker) sẽ tự động gọi lại lệnh php artisan kafka:consume-transactions để hồi sinh một Worker hoàn toàn mới, sạch sẽ và tiếp tục cày cuốc. Đây là cách thanh lịch nhất để giữ hệ thống chạy mượt mà nhiều năm liền.

Bằng việc kết hợp Kafka Key để giữ đúng thứ tự dữ liệu, tắt Auto Commit để không mất mát giao dịch, và reset tiến trình để dọn dẹp RAM, bạn đã chế tạo được một đường ống dữ liệu (Data Pipeline) cực kỳ kiên cố cho hệ thống Metro.

Để kết thúc chuỗi kiến trúc xịn sò này ở Bài 3, bạn muốn chúng ta giải quyết bài toán: Khi đối tác Hitachi bị sập API 1 tiếng đồng hồ, làm sao để bắt các sự kiện bị lỗi nhét vào Dead Letter Queue (DLQ) để chờ xử lý lại sau, hay muốn học cách sử dụng Schema Registry để chuẩn hóa định dạng JSON (tránh việc Frontend đẩy thiếu trường card_uid làm chết hệ thống)?