Giới thiệu

Trong bài viết này, chúng ta sẽ khám phá một vấn đề phổ biến khi mở rộng dịch vụ trên Kubernetes, đặc biệt là trong việc tích hợp với Redis để xử lý các sự kiện. Qua ví dụ cụ thể về Service W được viết bằng Node.js, chúng ta sẽ tìm hiểu cách xử lý các yêu cầu đồng thời mà không gặp phải tình trạng gửi thông điệp trùng lặp tới Kafka.

1. Vấn đề gặp phải

Khi Service W được triển khai với nhiều Pod, mỗi Pod sẽ đăng ký lắng nghe các sự kiện từ Redis thông qua cơ chế pub-sub. Khi một key trong Redis hết hạn, tất cả N Pod sẽ nhận sự kiện giống nhau và cố gắng thực hiện các thao tác cập nhật dữ liệu, dẫn đến việc gửi nhiều thông điệp giống nhau vào Kafka.

Dưới đây là đoạn mã logic từ Service W để xử lý sự kiện từ Redis:

const movieName = 'Bleach thousand year blood war';

const availableSeat = await prisma.seat.findFirst({
  where: {
    movie: {
      name: movieName,
    },
    claimedBy: null,
  },
});

if (!availableSeat) {
  throw new Error(`Oh no! ${movieName} is all booked.`);
}

await prisma.seat.update({
  data: {
    claimedBy: userId,
  },
  where: {
    id: availableSeat.id,
  },
});

// Gửi thông điệp tới Kafka để thực hiện một hành động nào đó
await producer.send({ topic: 'topic-name', messages: [] });

Như bạn thấy, nếu có nhiều Pod cùng thời điểm lắng nghe sự kiện này, sẽ có những thông điệp trùng lặp được gửi đến Kafka, gây ra rắc rối trong việc xử lý dữ liệu.

2. Giải pháp xử lý: Kiểm soát đồng thời theo cách tối ưu (Optimistic Concurrency Control)

Để giải quyết vấn đề này, chúng ta có thể áp dụng mô hình Kiểm soát đồng thời theo cách tối ưu (Optimistic Concurrency Control - OCC). Mô hình này cho phép xử lý đồng thời mà không cần phải dựa vào việc khóa tài nguyên. OCC sử dụng các mã thông báo, chẳng hạn như trường version hoặc timestamp trong bảng, để phát hiện thay đổi trên bản ghi.

Dưới đây là cách chúng ta có thể điều chỉnh logic để đảm bảo chỉ một worker thực hiện gửi thông điệp tới Kafka, tránh việc gửi thông điệp trùng lặp:

const userEmail = '98savage@gmail.io';
const movieName = 'Bleach thousand year blood war';

const availableSeat = await client.seat.findFirst({
  where: {
    Movie: {
      name: movieName,
    },
    claimedBy: null,
  },
});

if (!availableSeat) {
  throw new Error(`Oh no! ${movieName} is all booked.`);
}

const seats = await client.seat.updateMany({
  data: {
    claimedBy: userEmail,
    version: {
      increment: 1,
    },
  },
  where: {
    id: availableSeat.id,
    version: 0,
  },
});

if (seats.count === 0) {
  throw new Error(`That seat is already booked! Please try again.`);
}

// Gửi thông điệp tới Kafka để thực hiện một hành động nào đó
await producer.send({ topic: 'topic-name', messages: [] });

Với thay đổi này, khi N Pod nhận cùng một sự kiện, chỉ một Pod sẽ thực hiện cập nhật và gửi thông điệp đến Kafka. Các Pod còn lại sẽ nhận thông báo lỗi khi không còn phù hợp với điều kiện cập nhật nữa.

Tài nguyên liên quan

Bạn có thể tham khảo thêm tại:

• Prisma ORM và kiểm soát đồng thời
• Hướng dẫn sử dụng Redis Keyspace

source: viblo