Muchos usamos Kafka para publicar mensajes, pero ¿cómo los recibimos? En este artículo, escribiremos una pequeña aplicación para consumir mensajes de Kafka. Y, por supuesto, pruebas e2e. Primero comprendamos cómo funciona Kafka y qué es. Kafka es un intermediario de mensajes en el que algunos servicios generan mensajes y otros los reciben. Los corredores se utilizan principalmente en sistemas con arquitectura de microservicios para pasar mensajes entre servicios. Los mensajes se almacenan en temas. Al enviar un mensaje, el productor indica el nombre del tema, así como el mensaje en sí, que consta de una clave y un valor. Y eso es; El trabajo del productor está terminado. Entonces entran en juego los consumidores, se suscriben al tema deseado y empiezan a leer los mensajes. Cada aplicación tiene su propia cola, desde la cual el consumidor mueve el puntero de desplazamiento. Las características distintivas de Kafka son: Garantizar que todos los mensajes se ordenarán exactamente en la secuencia en la que llegaron al tema. Las tiendas Kafka leen mensajes por un tiempo Alto rendimiento Ahora, trabajemos con Kafka usando el marco NestJs. Primero, necesitamos crear un controlador que procese los mensajes. @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 atención al atributo , que indica que nuestra función recibirá mensajes del tema especificado en el archivo de configuración . El atributo ayuda a obtener el valor del mensaje del tema. @EventPattern handleEvent() config.get('kafka.topics.exampleTopic') @Payload() Para conectar su aplicación a los corredores de Kafka, debe hacer dos cosas. Para comenzar, conecte el microservicio en el archivo de inicio: 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'), }, }, }); Y luego ejecute los microservicios en 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 probar la aplicación, utilizo el paquete @testcontainers/kafka. Con la ayuda de esto, creé un contenedor zooKeeper y luego un contenedor 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, ]; } Tenga en cuenta que, en este archivo, anulé la dirección del corredor al contenedor recién creado. En el archivo de prueba, en la función , creo un cliente Kafka. Con el productor, también creo un tema y lanzo nuestra aplicación. 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); Por supuesto, en la función , debes detener los contenedores: afterAll afterAll(async () => { await app.close(); await Promise.all(kafkaContainers.map(c => c.stop())); }, 15 * 1000); Escribí una prueba que verifica que cuando llega un mensaje a un tema, nuestra función de controlador del controlador llama a la función de servicio necesaria. Para hacer esto, anulo la implementación de la función y espero a que se llame. 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(); }); }); Eso es todo. Trabajar con Kafka es increíblemente fácil si utiliza el marco NestJs. Espero que mi experiencia te sea de utilidad. Se puede ver una aplicación de ejemplo en 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.

Este audio es producido en el idioma original de la historia!

Demasiado Largo; Para Leer

Making security fun one episode at a time

Cómo consumir mensajes de Kafka con NestJS

Demasiado Largo; Para Leer

Viktoria Dolzhenko

STORY’S CREDIBILITY

Guide

About Author

ETIQUETAS

Languages

ESTE ARTÍCULO FUE PRESENTADO EN...

Cómo consumir mensajes de Kafka con NestJS

Demasiado Largo; Para Leer

Viktoria Dolzhenko

STORY’S CREDIBILITY

Guide

About Author

ETIQUETAS

Languages

ESTE ARTÍCULO FUE PRESENTADO EN...

HISTORIAS RELACIONADAS