Oto co Kafka FAQ ma do powiedzenia na temat dokładnie-raz:

Jak dostać dokładnie-raz wiadomości z Kafka?

Dokładnie raz semantyka ma dwie części: unikanie duplikatów podczas produkcji danych i unikanie duplikatów podczas zużycia danych.

Istnieją dwa podejścia do uzyskiwania semantyki dokładnie raz podczas produkcji danych:

• użyj jednego edytora na partycję i za każdym razem, gdy otrzymasz sprawdzenie błędu sieci ostatnia wiadomość w tej partycji, aby sprawdzić, czy twój ostatni zapis się powiódł
• Dołącz klucz główny (UUID lub coś takiego) w wiadomości i deduplikuj na konsumencie.

Jeśli zrobisz jedną z tych rzeczy, dziennik, który hostuje Kafka, będzie wolny od duplikatów. Jednak czytanie bez duplikatów zależy również od pewnej współpracy ze strony konsumenta. Jeśli konsument okresowo sprawdza swoją pozycję, to jeśli zawiedzie i uruchomi się ponownie, uruchomi się ponownie ze wskazanego pozycji. Tak więc, jeśli Dane wyjściowe i punkt kontrolny nie są zapisywane atomicznie, będzie można uzyskać duplikaty również tutaj. Ten problem dotyczy zwłaszcza systemu pamięci masowej. Na przykład, jeśli używasz bazy danych, możesz zatwierdzić je razem w transakcji. Ładowarka HDFS Camus, którą napisał LinkedIn, robi coś takiego dla obciążeń Hadoop. Inną alternatywą, która nie wymaga transakcji, jest przechowywanie offsetu z danymi załadowanymi i deduplikacją za pomocą połączenie tematu/partycji/przesunięcia.

Myślę, że są dwie ulepszenia, które znacznie by to ułatwiły:

• Producer idempotence moĹźe byÄ ‡ wykonane automatycznie i znacznie taniej poprzez opcjonalnÄ ... integracjÄ ™ wsparcia dla tego na serwerze.
• istniejący konsument wysokiego szczebla nie naraża dużo bardziej drobnoziarnistej kontroli przesunięć (np. aby zresetować swoją pozycję). Wkrótce będziemy nad tym pracować

Zgadzam się z deduplikacją RaGe ' a po stronie konsumenta. I używamy Redis do deduplikacji wiadomości Kafka.

Załóżmy, że Klasa Message ma element o nazwie 'uniqId', który jest wypełniany przez stronę producenta i jest gwarantowany jako unikalny. Używamy losowego ciągu o długości 12. (regexp to '^[A-Za-z0-9]{12}$')

Strona konsumencka używa SETNX Redis do deduplikacji i wygaśnięcia, aby automatycznie wyczyścić wygasłe klucze. Przykładowy kod:

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;
}

Powyższy kod wykrył duplikaty wiadomości kilka razy kiedy Kafka (Wersja 0.8.x) miały sytuacje. Dzięki Naszemu Dziennikowi audytu balansu wejścia/wyjścia nie doszło do utraty wiadomości ani dup.