Kafkajs & nestjs с упрощенным примером TypeScript

Этот пример кода попытался упростить интеграцию Kafka с NESTJS.

GitHub местоположение

https://github.com/rajeshkumarbehura/ts-nestjs-kafka

Как интегрировать-

Модуль Kafka создается как глобальный модуль в этом проекте. Чтобы интегрировать свой проект, скопируйте приложение/Common/Kakfa модуль и введите Kafkamodule в качестве модуля приложения.

    KafkaModule.register({
      clientId: 'test-app-client',
      brokers: ['localhost:9092'],
      groupId: 'test-app-group',
    }

В моем проекте он вводится на высший уровень в app.module.ts

Создайте полезную нагрузку кафки

Перед отправкой сообщения создайте пользовательскую полезную нагрузку для отправки сообщения

    const payload: KafkaPayload = {
      messageId: '' + new Date().valueOf(), // uuid
      body: message,
      messageType: 'Say.Hello',  
      topicName: 'hello.topic', 
    };

Отправить сообщение в Kafkatopic

Введите KafkaService в свой сервис или контроллер и вызовите SendMessage.

const value = await this.kafkaService.sendMessage('hello.topic', payload);

Потребительская реализация

Распространение класса AbstractKafkAconsumer и реализовать метод регистрации. Внутри RegisterTopic вам нужно только добавить имена тем. Не нужно вводить какую -либо услугу, как это реализовано во всем мире.

@Injectable()
export class ConsumerService extends AbstractKafkaConsumer {
    protected registerTopic() {
        this.addTopic('hello.topic');  
        this.addTopic('hello.fixed.topic');
    }
}

Подпишитесь на тему, когда GroupId не зафиксирована

Добавьте аннотацию в метод и определите имя темы как его параметр

 @SubscribeTo('hello.topic')
 helloSubscriber(payload: KafkaPayload) {
        console.log('Print message after receiving', payload);
 }

Подпишитесь на тему Когда GroupId исправлен

Когда несколько контейнеров или приложений работают во время горизонтального масштабирования, а ваш только один контейнер/приложение, необходимое для прослушивания темы.

 @SubscribeToFixedGroup('hello.fixed.topic')
 helloSubscriber(payload: KafkaPayload) {
        console.log('Print message after receiving', payload);
 }

Оригинал: “https://dev.to/rajeshkumarbehura/kafkajs-nestjs-with-typescript-simplified-example-35ep”