Chiến lược hiệu quả để tránh các thư trùng lặp trong người tiêu dùng apache kafka

Question

Tôi đã nghiên cứu apache kafka trong một tháng nay. Tuy nhiên, tôi bị mắc kẹt tại một điểm ngay bây giờ. Trường hợp sử dụng của tôi là, tôi có hai hoặc nhiều quy trình tiêu thụ chạy trên các máy khác nhau. Tôi đã chạy một vài thử nghiệm trong đó tôi đã xuất bản 10.000 tin nhắn trong máy chủ kafka. Sau đó, trong khi xử lý các thông điệp này tôi đã giết chết một trong các quy trình tiêu dùng và khởi động lại nó. Người tiêu dùng đã viết tin nhắn được xử lý trong một tệp. Vì vậy, sau khi tiêu thụ hoàn tất, tệp đã hiển thị hơn 10k thư. Vì vậy, một số tin nhắn đã được nhân đôi.

Trong quy trình tiêu dùng, tôi đã vô hiệu hóa cam kết tự động. Người tiêu dùng tự cam kết offsets hàng loạt khôn ngoan. Vì vậy, ví dụ: nếu 100 thư được ghi vào tệp, người tiêu dùng cam kết bù đắp. Khi quá trình tiêu dùng duy nhất đang chạy và nó bị treo và khôi phục trùng lặp được tránh theo cách này. Nhưng khi nhiều hơn một người tiêu dùng đang chạy và một trong số họ bị treo và phục hồi, nó sẽ ghi các thư trùng lặp vào tệp.

Có chiến lược hiệu quả nào để tránh các thư trùng lặp này không?

Answer 1

Câu trả lời ngắn gọn là, không.

Điều bạn đang tìm kiếm chính xác là một lần xử lý. Trong khi nó thường có vẻ khả thi, nó không bao giờ nên dựa vào vì luôn luôn có sự cẩn thận.

Thậm chí để cố ngăn chặn các bản sao bạn cần sử dụng người tiêu dùng đơn giản. Cách thức hoạt động của phương pháp này là cho mỗi người tiêu dùng, khi một thông điệp được tiêu thụ từ một số phân vùng, hãy ghi phân vùng và độ lệch của thông điệp được tiêu thụ vào đĩa. Khi người tiêu dùng khởi động lại sau khi thất bại, hãy đọc mức bù trừ được tiêu thụ cuối cùng cho mỗi phân vùng từ đĩa.

Nhưng ngay cả với mẫu này người tiêu dùng không thể đảm bảo rằng nó sẽ không xử lý lại thông báo sau khi thất bại. Điều gì nếu người tiêu dùng tiêu thụ một tin nhắn và sau đó thất bại trước khi bù đắp được flushed vào đĩa? Nếu bạn ghi vào đĩa trước khi xử lý tin nhắn, bạn sẽ làm gì nếu bạn viết offset và sau đó thất bại trước khi thực sự xử lý tin nhắn? Cùng một vấn đề này sẽ tồn tại ngay cả khi bạn đã cam kết bù đắp cho ZooKeeper sau mỗi tin nhắn.

Có một số trường hợp, mặc dù, trong đó xử lý chính xác một lần có thể đạt được nhiều hơn, nhưng chỉ cho một số trường hợp sử dụng nhất định. Điều này chỉ đơn giản đòi hỏi rằng bù đắp của bạn được lưu trữ trong cùng một vị trí như đầu ra của ứng dụng đơn vị. Ví dụ, nếu bạn viết một người tiêu dùng đếm tin nhắn, bằng cách lưu trữ các tính toán bù đắp cuối cùng với mỗi số bạn có thể đảm bảo rằng bù đắp được lưu trữ cùng một lúc với trạng thái của người tiêu dùng. Tất nhiên, để đảm bảo xử lý chính xác một lần, điều này sẽ yêu cầu bạn tiêu thụ chính xác một thông điệp và cập nhật trạng thái chính xác một lần cho mỗi thư, và điều đó hoàn toàn không thực tế đối với hầu hết các ứng dụng khách hàng của Kafka. Theo bản chất của nó, Kafka tiêu thụ tin nhắn theo lô vì lý do hiệu suất.

Thông thường thời gian của bạn sẽ được chi tiêu tốt hơn và ứng dụng của bạn sẽ đáng tin cậy hơn nhiều nếu bạn chỉ đơn giản là thiết kế nó là không đáng kể.

Answer 2

Đây là những gì Kafka FAQ có nói về vấn đề chính xác-once:

Làm thế nào để chính xác một lần gửi tin nhắn từ Kafka?

Các ngữ nghĩa chính xác một lần có hai phần: tránh trùng lặp trong quá trình tạo dữ liệu và tránh trùng lặp trong khi sử dụng dữ liệu.

Có hai phương pháp để nhận được đúng một lần ngữ nghĩa trong sản xuất dữ liệu:

• Sử dụng một đơn nhà văn mỗi phân vùng và mỗi khi bạn nhận được một lỗi mạng kiểm tra các tin nhắn cuối cùng trong phân vùng đó để xem nếu cuối cùng của bạn viết thành công
• Bao gồm khóa chính (UUID hoặc nội dung nào đó) trong thư và trùng lặp trên người tiêu dùng.

Nếu bạn thực hiện một trong những điều này, nhật ký mà máy chủ Kafka sẽ không bị trùng lặp. Tuy nhiên, việc đọc mà không có bản sao phụ thuộc vào một số sự hợp tác từ người tiêu dùng. Nếu người tiêu dùng định kỳ kiểm tra vị trí của nó sau đó nếu nó không thành công và khởi động lại nó sẽ khởi động lại từ vị trí được kiểm tra. Vì vậy, nếu đầu ra dữ liệu và điểm kiểm tra không được viết bằng nguyên tử thì cũng có thể nhận được các bản sao ở đây. Vấn đề này đặc biệt đối với hệ thống lưu trữ của bạn. Ví dụ, nếu bạn đang sử dụng một cơ sở dữ liệu, bạn có thể kết hợp chúng lại với nhau trong một giao dịch. Bộ tải HDFS Camus mà LinkedIn viết đã thực hiện một cái gì đó như thế này cho tải Hadoop. Cách thay thế khác không yêu cầu giao dịch là lưu trữ bù trừ với dữ liệu được nạp và loại bỏ trùng lặp bằng cách sử dụng kết hợp chủ đề/phân vùng/bù đắp.

Tôi nghĩ có hai cải tiến mà có thể làm dễ dàng hơn này rất nhiều:

• Nhà sản xuất idempotence có thể được thực hiện tự động và nhiều hơn với giá rẻ hơn bằng cách tùy chọn tích hợp hỗ trợ cho việc này trên máy chủ.
• Người tiêu dùng cao cấp hiện tại không thể hiện nhiều kiểm soát hạt mịn hơn (ví dụ: để đặt lại vị trí của bạn). Chúng tôi sẽ làm việc trên đó sớm

Answer 3

Tôi đồng ý với deduplicate Rage về phía người tiêu dùng. Và chúng tôi sử dụng Redis để loại bỏ thông điệp Kafka.

Giả sử lớp Thông báo có thành viên được gọi là 'uniqId', được điền bởi phía nhà sản xuất và được đảm bảo là duy nhất. Chúng tôi sử dụng một chuỗi 12 chiều ngẫu nhiên. (regexp là '^[A-Za-z0-9]{12}$')

Phía người tiêu dùng sử dụng SETNX của Redis để trùng lặp và EXPIRE để tự động xóa các khóa đã hết hạn. Mã mẫu:

Message msg = ... // eg. ConsumerIterator.next().message().fromJson(); 
Jedis jedis = ... // eg. JedisPool.getResource(); 
String key = "SPOUT:" + msg.uniqId; // prefix name at will 
String val = Long.toString(System.currentTimeMillis()); 
long rsps = jedis.setnx(key, val); 
if (rsps <= 0) { 
    log.warn("kafka dup: {}", msg.toJson()); // and other logic 
} else { 
    jedis.expire(key, 7200); // 2 hours is ok for production environment; 
} 

Mã trên đã phát hiện thư trùng lặp nhiều lần khi Kafka (phiên bản 0.8.x) gặp tình huống. Với nhật ký kiểm tra số dư đầu vào/đầu ra của chúng tôi, không có thông báo bị mất hoặc bị lỗi xảy ra.