Các ứng dụng cần đọc dữ liệu từ Kafka sử dụng KafkaConsumer để subcribe tới các topics và nhận các messages về từ những topics này. Việc đọc dữ liệu từ Kafka có một chút khác biệt so với việc đọc dữ liệu từ các hệ thống message khác, và có một số khái niệm và ý tưởng độc đáo liên quan đến quá trình này. Rất khó để có thể hiểu cách sử dụng Consumer API mà chưa hiểu được những khái niệm này trước. Mình sẽ bắt đầu giải thích một số khái niệm quan trọng, sau đó sẽ đi qua một số ví dụ thể hiện các cách khác nhau mà Consumer API có thể được sử dụng để triển khai các ứng dụng với các yêu cầu khác nhau.
Các khái niệm về Kafka Consumer
Để hiểu cách đọc dữ liệu từ Kafka, đầu tiên bạn cần hiểu consumer và consumer groups là gì trước đã. Các phần mình viết dưới đây sẽ giải thích rõ các khái niệm này.
#1. Consumer và Consumer Group
Giả sử bạn có một ứng dụng cần đọc các message từ một topic, xử lý validation chúng và sau đó ghi kết quả tới một nguồn dữ liệu khác như Elasticsearch chẳng hạn. Trong trường hợp này, ứng dụng của bạn sẽ tạo ra một consumer object, nó subcribe tới topic tương ứng, và bắt đầu nhận về các message, validate các message này và ghi ra các kết qủa. Quá trình này có thể hoạt động tốt trong một thời gian, tuy nhiên điều gì sẽ xảy ra nếu như tỷ lệ các producer ghi message tới topic vượt quá tốc độ mà ứng dụng consumer của bạn có thể validate? Nếu bạn bị giới hạn ở tỷ lệ một consumer đọc và xử lý dữ liệu, ứng dụng của bạn có thể sẽ bị hãm phanh lại, giống như kiểu ngày càng chậm chạp hơn, không thể đuổi kịp các message đến. Rõ ràng là bạn sẽ cần phải mở rộng việc đọc message từ topic này rồi. Vì có nhiều producer có thể ghi message tới cùng một topic tại một thời điểm, chúng ta cũng sẽ cần cho phép nhiều consumer đọc message từ cùng một topic này, phân tách dữ liệu ra cho nhiều consumer cùng đọc (hay còn gọi là scaling).
Kafka Consumer thường là một phần của Consumer Group. Khi nhiều consumer subcribe tới một topic và đều thuộc về một Consumer Group này, mỗi consumer trong consumer group sẽ nhận về các message từ một tập các partition khác nhau trong topic.
Giả sử mình sẽ có một topic gọi là T1 với bốn partition. Bây giờ, mình sẽ tạo ra một consumer gọi là C1 – là consumer duy nhất trong group G1, và dùng nó để subcribe tới topic T1. Consumer C1 sẽ nhận tất cả các message từ bốn partition trong topic T1.
Nếu như mình thêm một consumer khác – C2 tới group G1, mỗi consumer sẽ chỉ nhận các message từ hai partition. Có thể các message từ partition 0 và 2 sẽ đi tới C1 và các message từ partition 1 và 3 sẽ tới consumer C2.
Nếu group C1 có bốn consumer, khi đó mỗi consumer sẽ đọc các message từ chỉ một partition như hình bên dưới đây mà thôi.
Nếu chúng ta thêm nhiều consumer hơn nữa vào group G1 này nhiều hơn số partition mà chúng ta có trong một topic, khi đó một số consumer sẽ không nhận được message nào.
Cách để scale việc tiêu thụ các message từ một topic chính là thêm nhiều consumer hơn vào một consumer group. Thường thì Kafka Consumer sẽ thực hiện các hoạt động trên các message sẽ dẫn đến độ trễ cao cho consumer như là ghi kết quả dữ liệu tới database hay tốn thời gian tính toán trên dữ liệu nhận được. Trong những trường hợp như vậy, chỉ một consumer duy nhất sẽ không thể nào giúp đảm bảo tốc độ xử lý dữ liệu trong một topic được, và việc thêm nhiều consumer vào consumer group sẽ giúp chia tải cho hệ thống – và đây chính là phương thức scaling chính trong Kafka.
Có một lý do mà chúng ta nên tạo các topics với số partition lớn là – nó cho phép thêm nhiều consumer khi tải hệ thống tăng. Nên nhớ, không nên để tình trạng số consumer lớn hơn số partition mà bạn có trong một topic – một số consumer sẽ rơi vào trạng thái rảnh rỗi.
Ngoài việc thêm các consumer để mở rộng cho một ứng dụng, thường khá phổ biến trên các hệ thống chính là: có nhiều ứng dụng sẽ đọc dữ liệu từ cùng một topic. Trong thực tế, một trong những mục tiêu thiết kế chính trong Kafka chính là đảm bảo dữ liệu được tạo ra tới các topics luôn có sẵn cho nhiều trường hợp sử dụng cho toàn bộ tổ chức. Trong những trường hợp như vậy, chúng ta luôn muốn mỗi ứng dụng sẽ nhận được tất cả các message trong một topic, đảm bảo ứng dụng có consumer group của riêng nó. Không giống như các hệ thống message truyền thống, Kafka mở rộng tới một số lượng lớn các consumer và consumer groups mà không hề làm giảm performance.
Trong ví dụ trước đó, nếu chúng ta thêm một consumer groups mới G2 với chỉ hai consumer, consumer này sẽ nhận tất cả các message trong topic T1, độc lập với những gì G1 đang thực hiện. G2 có thể có nhiều consumer, trong trường hợp này, mỗi consumer sẽ nhận được một tập các partition, giống như cách mình đã chỉ ra bên trên cho G1, nhưng G2 vẫn sẽ nhận được toàn bộ các message, bất chấp các consumer group khác thế nào đi nữa.
Tóm lại, bạn tạo ra một consumer group cho mỗi ứng dụng cần tất cả các message từ một hoặc nhiều topic. Bạn thêm nhiều consumer tới một consumer group đang tồn tại để mở rộng việc đọc và xử lý các message từ topics, vì vậy mỗi consumer bổ sung trong một group sẽ chỉ nhận được một tập của các message đó.
#2. Consumer Groups và Partition Rebalance
Như mình đã nói ở phần trước đó, các consumer trong một consumer group chia sẽ quyền sở hữu các partition trong một topic mà chúng subcribe tới. Khi chúng ta thêm một consumer mới tới group, nó bắt đầu tiêu thụ các message từ các partitions trước đó mà đã được tiêu thụ bởi consumer khác. Điều tương tự cũng xảy ra khi một consumer bị shutdown hay crash. Nó rời khỏi group, và các partitions nó đã sử dụng sẽ được tiêu tiêu thụ bởi một trong những consumer còn lại trong consumer group. Quá trình sắp xếp lại các partition (Reasssignment) tới các consumers cũng xảy ra khi các topics mà consumer group đang tiêu thụ được chỉnh sửa (nếu người quản trị thêm các partition mới vào topic).
Quá trình di chuyển các partitions được liên kết tới từ consumer này tới consumer khác được gọi là quá trình rebalance. Rebalance rất quan trọng vì chúng cung cấp cho consumer group khả năng sẵn sàng và tính mở rộng cao (cho phép chúng ta thêm và xóa consumer dễ dàng và an toàn), nhưng trong quá trình sử dụng bình thường, quá trình này không nên xảy ra. Trong quá trình rebalance, consumer không thể tiêu thụ các message được, do đó, việc rebalance về cơ bản sẽ làm cho toàn bộ consumer trong consumer group không thể tiêu thụ message được. Ngoài ra, khi các partitions được di chuyển từ consumer này tới consumer khác, consumer sẽ mất đi trạng thái hiện tại của nó. Nếu như nó đã cache dữ liệu sẵn rồi, nó sẽ cần phải làm mới lại cache – dẫn đến làm chậm ứng dụng cho đến khi consumer thiết lập lại trạng thái của nó.
Cách mà consumer duy thì tư cách thành viên trong một consumer group và sở hữu các partitions được gán tới chúng được thực hiện bằng cách gửi heartsbeats tới Kafka Broker – được chỉ định làm Kafka Coordinator hay điều phối viên (broker này có thể khác consumer groups khác). Miễn là consumer gửi heartbeats đều đặn, nó sẽ được coi là alive (đang còn khỏe) và đang xử lý các message từ các partitions của nó. Heartbeats được gửi khi một consumer polls (nhận các record về) diễn ra và khi nó commit records mà nó đã tiêu thụ.
Nếu như consumer ngừng gửi heartbeats trong một khoảng thời gian được quy định, khi đó phiên làm việc của nó sẽ hết hạn và group coordinator sẽ coi như nó đã chết và kích hoạt quá trình rebalance. Nếu một consumer gặp sự cố và ngừng xử lý các message, sẽ mất vài giây để group coordinator quyết định rằng nó đã chết hay chưa và kích hoạt quá trình rebalance. Trong những giây này, không có message nào được xử lý từ các partitions thuộc sở hữu bởi consumer đã chết. Khi đóng một consumer, consumer sẽ thông báo cho group coodinator rằng nó sẽ đang đi, và group coordinator sẽ kích hoạt quá trình rebalance ngay lập tức, giảm thiểu khoảng cách trong xử lý.
Xem thêm:
