Muitos de nós usamos o Kafka para publicar mensagens, mas como as recebemos? Neste artigo, escreveremos um pequeno aplicativo para consumir mensagens do Kafka. E, claro, testes e2e.  Vamos primeiro entender como funciona o Kafka e o que é.  Kafka é um corretor de mensagens no qual alguns serviços geram mensagens e outros as recebem. Os corretores são usados principalmente em sistemas com arquitetura de microsserviços para passar mensagens entre serviços.  As mensagens são armazenadas em tópicos. Ao enviar uma mensagem, o produtor indica o nome do tópico, bem como a própria mensagem, que consiste em uma chave e um valor. E é isso; o trabalho do produtor está concluído.   Aí os consumidores entram em cena, se inscrevem no tema desejado e começam a ler as mensagens. Cada aplicação possui sua própria fila, a partir da qual o consumidor move o ponteiro de deslocamento.   As características distintivas do Kafka são:  Garantir que todas as mensagens serão ordenadas exatamente na sequência em que chegaram ao tópico  Kafka armazena mensagens lidas por um tempo  Alto rendimento  Agora, vamos trabalhar com Kafka usando a estrutura NestJs. Primeiro, precisamos criar um controlador que irá processar mensagens.   @Controller() export class AppController{ constructor( private readonly appService: AppService, ) { } @EventPattern(config.get('kafka.topics.exampleTopic'), Transport.KAFKA) handleEvent( @Payload() payload: ExamplePayloadDto, ): Promise<void> { return this.appService.handleExampleEvent(payload.message); } }  Preste atenção ao atributo   , que indica que nossa função   receberá mensagens do tópico especificado no arquivo de configuração   . O atributo   ajuda a obter o valor da mensagem do tópico. @EventPattern handleEvent() config.get('kafka.topics.exampleTopic') @Payload()  Para conectar seu aplicativo aos corretores Kafka, você precisa fazer duas coisas. Para começar, conecte o microsserviço no arquivo de inicialização:   app.connectMicroservice({ transport: Transport.KAFKA, options: { client: { clientId: config.get('kafka.clientId'), brokers: config.get('kafka.brokers'), retry: { retries: config.get('kafka.retryCount'), }, }, consumer: { groupId: config.get('kafka.consumer.groupId'), }, }, });  E então execute os microsserviços no main.ts:   async function bootstrap() { const app = await NestFactory.create(AppModule, { bufferLogs: true, }); appStartup(app); await app.startAllMicroservices(); await app.listen(config.get('app.port')); }; void bootstrap();  Para testar o aplicativo, uso o pacote @testcontainers/kafka. Com a ajuda disso, criei um contêiner zooKeeper e, em seguida, um contêiner Kafka:   export async function kafkaSetup(): Promise<StartedTestContainer[]> { const network = await new Network().start(); const zooKeeperHost = "zookeeper"; const zooKeeperPort = 2181; const zookeeperContainer = await new GenericContainer("confluentinc/cp-zookeeper:7.3.2") .withNetwork(network) .withNetworkAliases(zooKeeperHost) .withEnvironment({ ZOOKEEPER_CLIENT_PORT: zooKeeperPort.toString() }) .withExposedPorts(zooKeeperPort) .start(); const kafkaPort = 9093; const kafkaContainer = await new KafkaContainer() .withNetwork(network) .withZooKeeper(zooKeeperHost, zooKeeperPort) .withExposedPorts(kafkaPort) .start(); const externalPort = kafkaContainer.getMappedPort(kafkaPort); config.set('kafka.brokers', [`localhost:${externalPort}`]); return [ zookeeperContainer, kafkaContainer, ]; }  Observe que, neste arquivo, substituí o endereço do corretor pelo contêiner recém-criado.  No próprio arquivo de teste, na função   , crio um cliente Kafka. Com o produtor também crio um tópico e lanço nosso aplicativo. beforeAll   beforeAll(async () => { kafkaContainers = await kafkaSetup(); kafka = new Kafka({ clientId: 'mock', brokers: config.get('kafka.brokers'), logLevel: logLevel.NOTHING, }); producer = kafka.producer(); await producer.connect(); const admin = kafka.admin(); await admin.connect(); await admin.createTopics({ topics: [{ topic: config.get('kafka.topics.exampleTopic') }], }); appService = mockDeep<AppService>(); const module: TestingModule = await Test.createTestingModule({ imports: [AppModule], }) .overrideProvider(AppService) .useValue(appService) .compile(); app = module.createNestApplication(); appStartup(app); await app.startAllMicroservices(); await app.init(); }, 30 * 1000);  Claro, na função   , você precisa parar os contêineres: afterAll   afterAll(async () => { await app.close(); await Promise.all(kafkaContainers.map(c => c.stop())); }, 15 * 1000);  Escrevi um teste que verifica se quando uma mensagem chega em um tópico, nossa função manipuladora do controlador chama a função de serviço necessária. Para fazer isso, substituo a implementação da função   e espero que ela seja chamada. handleExampleEvent   describe('handleEvent', () => { it('should call appService', async () => { let resolve: (value: unknown) => void; const promise = new Promise((res) => { resolve = res; }); appService.handleExampleEvent.mockImplementation(async () => { resolve(0); }); const event: ExamplePayloadDto = { message: 'Hello World!', }; await producer.send({ topic: config.get('kafka.topics.exampleTopic'), messages: [{ key: 'key', value: JSON.stringify(event), }] }); await promise; await kafka.producer().disconnect(); }); });  Isso é tudo. Trabalhar com Kafka é incrivelmente fácil se você usar a estrutura NestJs. Espero que minha experiência seja útil para você. Um exemplo de aplicativo pode ser visto em  https://github.com/waksund/kafka

Walkthroughs, tutorials, guides, and tips. This story will teach you how to do something new or how to do something better.

Read My Stories

Este áudio é produzido no idioma original da história!

Como consumir mensagens Kafka com NestJS

About Author

COMENTARIOS

Rótulos

ESTE ARTIGO FOI APRESENTADO EM

Related Stories

Guia do arquiteto para construir arquitetura de referência para um Datalake de IA/ML

Crescimento de Criptomoedas: Criando Personas de Usuários Eficazes

Telegram: a ponte da Crypto Island para o continente

Toque para ganhar: Telegram pode integrar os próximos 10 bilhões de usuários criptográficos antes de Solana

Guia do arquiteto para construir arquitetura de referência para um Datalake de IA/ML

Crescimento de Criptomoedas: Criando Personas de Usuários Eficazes

Telegram: a ponte da Crypto Island para o continente

Toque para ganhar: Telegram pode integrar os próximos 10 bilhões de usuários criptográficos antes de Solana

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps