как идентифицировать и объединять сообщения разных очередей в кафке

Question

Справочная информация:
Ранее мы использовали поиск в спящем режиме, очередь Lucene и jboss hornetq для индексации.

Наше приложение является производителем и отправляет метаданные (уникальная информация о данных для идентификации записи в базе данных).в шершень.Потребитель получает эти метаданные и запрашивает базу данных, чтобы получить полную информацию о записи (включая дочерние объекты).Это гораздо более ориентированный на базу данных подход.

Теперь мы хотим исключить ориентированный на базу данных подход для индексации.Мы решили использовать kafka, а не hornetq.

Нет проблем при создании данных пользователем.

Мы видим, что существует потенциальная проблема, когда пользователь редактирует данные (скажем, родительский объект сдва дочерних объекта).Когда данные извлекаются из базы данных для отображения пользователем,
мы помещаем те же данные в кафку topic1.Когда пользователь изменяет данные (скажем, данные уровня parenet) и отправляет.Мы получаем только данные родительского уровня (не получаем данные дочерних объектов), мы помещаем измененные данные в topic2.Теперь мы должны объединить сообщение, присутствующее в topic1 (дочерние объекты), с соответствующим сообщением в topic2 (данные родительского уровня)

Примечание. Нам нужно выбрать этот маршрут, поскольку вы знаете, что в Indexing нет обновлений, а оноэто удалить, а затем вставить.

Вопросы:

1. Если я придерживаюсь вышеуказанного подхода, как я могу сопоставить конкретное сообщение, присутствующее в теме1, с конкретным сообщением в теме2,Можно ли указать одинаковые идентификаторы сообщений в topic1 и topic2?

2. Можно ли решить эту проблему, если я использую одну тему?

3. Есть ли лучший дизайн / подход для решения вышеуказанной проблемы?

Заранее спасибо.

Answer 1

1. Если я придерживаюсь вышеуказанного подхода, как я могу сопоставить конкретное сообщение, присутствующее в теме1, с конкретным сообщением в теме2.Есть ли способ предоставить одинаковые идентификаторы сообщений в topic1 и topic2?

Чтобы отобразить или объединить определенные сообщения между темами в одном кластере Kafka, возможно, Kafka Stream и KSQL - это хорошее направлениесделать.Можете ли вы найти ссылку здесь .

Существует множество способов сделать объект уникальным, и я предлагаю использовать идентификатор родительского объекта при отправке сообщений в topic1 и topic2.Пример кода Java выглядит следующим образом:

ProducerRecord<String, ParentEntity> record = new ProducerRecord<>(topic1, 
ParentEntity.getId(), ParentEntity);
ListenableFuture<SendResult<String, ParentEntity>> future = 
kafkaTemplate.send(record);
future.addCallback(new ListenableFutureCallback<SendResult<String, 
ParentEntity>>() {
    @Override
    public void onSuccess(SendResult<String, ParentEntity> result) {}

    @Override
    public void onFailure(Throwable ex) {
        //print out error log 
    }
});

ProducerRecord<String, ChildEntity> record = new ProducerRecord<>(topic2, 
ChildEntity.getParentEntityId(), ChildEntity);
ListenableFuture<SendResult<String, ChildEntity>> future = 
kafkaTemplate.send(record);
future.addCallback(new ListenableFutureCallback<SendResult<String, 
ChildEntity>>() {
    @Override
    public void onSuccess(SendResult<String, ChildEntity> result) {}

    @Override
    public void onFailure(Throwable ex) {
        //print out error log 
    }
});

Есть ли способ решить эту проблему, если я использую одну тему?

Вы можете создать новую таблицу (сказал A) в базе данных, чтобы сохранить полное сообщение для отправкидля индексации.Каждый раз, когда пользователь создает или обновляет данные, сообщение также должно быть вставлено / обновлено в таблицу A. Наконец, ваш клиент Kafka извлекает объекты сообщения из таблицы A и создает уникальную тему в кластере Kafka.

Есть ли лучший дизайн / подход для решения вышеуказанной проблемы?

Можете ли вы попробовать Kafka Stream и KSQL, как я упоминал выше.