Apache Kafka ด้วย Node.js

// producer.js
const { Kafka } = require('kafkajs');
const fs = require('fs');

// สร้าง Kafka producer
const kafka = new Kafka({
    clientId: 'my-producer',
    brokers: ['kafka:9092'],
});

// สร้าง producer
const producer = kafka.producer();

// ฟังก์ชันส่งข้อความ
async function sendMessage() {
    const message = {
        topic: 'my-topic',
        messages: [
            { key: 'user-123', value: JSON.stringify({ name: 'John', action: 'login' }) },
            { key: 'user-456', value: JSON.stringify({ name: 'Jane', action: 'purchase' }) },
        ],
    };

    try {
        await producer.send(message);
        console.log('Message sent:', message);
    } catch (error) {
        console.error('Error sending message:', error);
    }
}

// ส่งข้อความ 100 ครั้ง
setInterval(() => {
    sendMessage();
}, 1000);

// จัดการ shutdown
process.on('SIGINT', async () => {
    await producer.flush();
    process.exit(0);
});

// consumer.js
const { Kafka, Consumer } = require('kafkajs');

// สร้าง Kafka consumer
const kafka = new Kafka({
    clientId: 'my-consumer',
    brokers: ['kafka:9092'],
    groupId: 'my-group',
});

// สร้าง consumer
const consumer = new Consumer({
    groupId: 'my-group',
    topic: 'my-topic',
    fromBeginning: false,
});

// รัน consumer
await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
        console.log({
            topic,
            partition,
            offset: message.offset,
            key: message.key.toString(),
            value: message.value.toString(),
        });

        // ประมวลผม
        console.log('Processing message...');
        
        // Commit offset เมื่อประมวลผมเสร็จ
        if (shouldCommit) {
            consumer.commitOffsets();
        }
    },
});

// จัดการ shutdown
process.on('SIGINT', async () => {
    await consumer.disconnect();
    process.exit(0);
});