? ?? Spring EmbeddedKafka ??? ?????? ???? ???? ????? ???? ??? ?????? ?? ??? ???? ?? ??? ??. Kafka ???? ???? ?????? 'acks'? ???? ???? ???? ???? ?? ???? ????? ??? ??? ????. ??? ???? ??? ???? ???? ??? ???? ???? ??? ??? ?????. ? ????? ? ??? ???? ?? ????? ??? ?????.

Spring EmbeddedKafka ????? ???? ?? ???? ???? ???? ???? ????? ???? ?? ???? ?? ????, ?? ?? ?????? ?? ?????. ??? Kafka? ??? ??? ???? ???? ?????. ??? ?? ack ??? ??? ?? ?? ????? ??? ? ??? ??? ???? ???? ??? ?? ??? ?? ????. ??? ? ??? ????? ???? ????? ???? ???.

?? ????: ?? ?? ??? ???? ??

???? ???? ?????? ???? ???? ??? ? ????? ?? ? ?? ??? ?????. ???? ?? ??? ?? ?? ???? ???? ????. ?? ?? ??? ????.

• ?? ConcurrentHashMap: ?? JVM ??? ??? ???? ???? ??????.
• Redis ?? ?? ?? ????: ?? ??? ??? ???? ???? ?????.

?? ??:

1. ??? ?:

   • ???? ??? ?? ?? ID(?: UUID)? ?????.
   • ??? ?? ?? ??? ??? ??? ID? ????? ????.
   • ??? ???? ??? ID? ???? ?? ??? ?????.
   • ???? ?? ID? ?????? ????? ?????. ?? ?? ??? ???? ??? ??? ??? ??? ??? ?????.

2. ??? ?:

   • ???? ?? ? ???? ?? ID? ?????.
   • ???? ??? ? ?? ?? ????? ID? ?????.
   • Acknowledgement.acknowledge()? ???? ???? ?????? ?????.

?? ??(ConcurrentHashMap ??):

 org.springframework.kafka.core.KafkaTemplate ????;
org.springframework.kafka.support.Acknowledgement ????;
org.springframework.stereotype.Component ????;

java.util.UUID ????;
java.util.concurrent.ConcurrentHashMap ????;
import java.util.concurrent.TimeUnit;

@??
?? ??? MessageHandler {

    ?? ?? KafkaTemplate<string string> kafkaTemplate;
    private final ConcurrentHashMap<string boolean> receivedMessages = new ConcurrentHashMap();

    ?? MessageHandler(KafkaTemplate<string string> kafkaTemplate) {
        this.kafkaTemplate = ??????;
    }

    ?? ?? sendMessage(??? ??, ??? ???)? ??? ??????.
        String messageId = UUID.randomUUID().toString();
        ???Messages.put(messageId, false); // ??? ???? ??? ??

        kafkaTemplate.send(topic, messageId, message).get(); // ???? ????? ?????? ?????. // ??? ??? ???? ?? ??? ?????. waitForConfirmation(messageId, 5, TimeUnit.SECONDS);
    }

    ?? ?? ?? ???(??? messageId, ??? ???, ?? ??) {
        ???? {
            // ??? ??...
            System.out.println("??? ???: " message);
            ???Messages.remove(messageId); //ID? ???? ?????? ?????.ackdgment.acknowledge();
        } ??(?? e) {
            // ?? ??...
        }
    }

    private void waitForConfirmation(String messageId, ? ?? ??, TimeUnit ??)? ??? ??????.
        ? startTime = System.currentTimeMillis();
        while (processedMessages.containsKey(messageId)) {
            if (System.currentTimeMillis() - startTime > unit.toMillis(timeout)) {
                throw new Exception("??? ?? ?? ?? ??: " messageId);
            }
            Thread.sleep(100); //??? CPU ??? ??? ?? ?? ???}
    }
}</string></string></string>

KafkaListener ??:

 org.springframework.kafka.annotation.KafkaListener ????;
org.springframework.kafka.support.Acknowledgement ????;
org.springframework.stereotype.Component ????;

@??
?? ??? KafkaConsumer {

    ?? ?? MessageHandler messageHandler;

    ?? KafkaConsumer(MessageHandler messageHandler) {
        this.messageHandler = ??????;
    }

    @KafkaListener(topics = "your_topic", groupId = "your_group_id")
    public void listening(String messageId, String message, Acknowledgement acknowledgement) {
        messageHandler.consumeMessage(messageId, ???, ??);
    }
}

?? ??:

• ?? ?? ????: ??? ??? ????? ???? ?? ??? ???? ???.
• ?? ??: ??? ???? ? ?? ?? ?? ?? ?? ??? ?? ??? ?? ??? ???? ???? ???.
• ??? ID ???: ??? ID? ??? ???? ???? ?????.
• ??? ??: ?? ???? ??? ???? ??? ?? ?????? ??? ??? ?? ??? ???? ? ??? ??? ???? ???.

??:

Kafka ??? ???? ??? ??? ???? ????? ???? ??? ?? ?? ??? ????? ???? ? ??? ??? ? ????. ?? ?? ??? ConcurrentHashMap? ???? ? ?? ???? ?????. Redis? ?? ?? ????? ???? ? ?? ?? ??? ?? ?? ??? ???? ??? ? ????. ??? ???? ??? ?? ?? ??? ???? ??? ?? ??? ????? ????. ??? ???? ?? ???? ???? ????? ???? ???? ???? ????? ? ? ????.

? ??? ??? ??? ???? Spring EmbeddedKafka ??? ?? ??? ?? ?????. ??? ??? PHP ??? ????? ?? ?? ??? ?????!