Nous sommes nombreux à utiliser Kafka pour publier des messages, mais comment les recevons-nous ? Dans cet article, nous allons écrire une petite application pour consommer les messages de Kafka. Et bien sûr, les tests e2e.  Comprenons d'abord comment fonctionne Kafka et ce que c'est.  Kafka est un courtier de messages dans lequel certains services génèrent des messages et d'autres les reçoivent. Les courtiers sont principalement utilisés dans les systèmes dotés d'une architecture de microservices pour transmettre des messages entre les services.  Les messages sont stockés dans des sujets. Lors de l'envoi d'un message, le producteur indique le nom du sujet, ainsi que le message lui-même, qui est constitué d'une clé et d'une valeur. Et c'est tout; le travail du producteur est terminé.   Ensuite, les consommateurs entrent en jeu, s’abonnent au sujet souhaité et commencent à lire les messages. Chaque application possède sa propre file d'attente de lecture à partir de laquelle le consommateur déplace le pointeur de décalage.   Les caractéristiques distinctives de Kafka sont :  Garantir que tous les messages seront classés exactement dans l'ordre dans lequel ils sont arrivés dans le sujet  Kafka stocke les messages lus pendant un certain temps  Débit élevé  Travaillons maintenant avec Kafka en utilisant le framework NestJs. Tout d’abord, nous devons créer un contrôleur qui traitera les messages.   @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); } }  Faites attention à l'attribut   , qui indique que notre fonction   recevra des messages du sujet spécifié dans le fichier de configuration   . L'attribut   permet d'obtenir la valeur du message du sujet. @EventPattern handleEvent() config.get('kafka.topics.exampleTopic') @Payload()  Pour connecter votre application aux courtiers Kafka, vous devez faire deux choses. Pour commencer, connectez le microservice dans le fichier de démarrage :   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'), }, }, });  Et puis exécutez les microservices dans le fichier 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();  Pour tester l'application, j'utilise le package @testcontainers/kafka. Avec l'aide de cela, j'ai créé un conteneur zooKeeper, puis un conteneur 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, ]; }  Veuillez noter que, dans ce fichier, j'ai remplacé l'adresse du courtier par le conteneur nouvellement créé.  Dans le fichier de test lui-même, dans la fonction   , je crée un client Kafka. Avec le producteur, je crée également un sujet et lance notre application. 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);  Bien entendu, dans la fonction   , vous devez arrêter les conteneurs : afterAll   afterAll(async () => { await app.close(); await Promise.all(kafkaContainers.map(c => c.stop())); }, 15 * 1000);  J'ai écrit un test qui vérifie que lorsqu'un message arrive dans un sujet, notre fonction de gestionnaire depuis le contrôleur appelle la fonction de service nécessaire. Pour ce faire, je remplace l'implémentation de la fonction   et j'attends qu'elle soit appelée. 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(); }); });  C'est tout. Travailler avec Kafka est incroyablement simple si vous utilisez le framework NestJs. J'espère que mon expérience vous sera utile. Un exemple d'application peut être consulté sur  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

Cet audio est produit dans la langue originale de l'histoire !

Comment consommer des messages Kafka avec NestJS

About Author

COMMENTAIRES

ÉTIQUETTES

CET ARTICLE A ÉTÉ PARU DANS

Related Stories

Croissance des crypto-monnaies : créer des profils d'utilisateurs efficaces

Nomades numériques, écoutez : ce que vous devez savoir sur le nouveau visa DTV de la Thaïlande

Télégramme : le pont de Crypto Island vers le continent

Libérer la puissance de l’IA. Une revue systématique des techniques de pointe : résumé et introduction

Croissance des crypto-monnaies : créer des profils d'utilisateurs efficaces

Nomades numériques, écoutez : ce que vous devez savoir sur le nouveau visa DTV de la Thaïlande

Télégramme : le pont de Crypto Island vers le continent

Libérer la puissance de l’IA. Une revue systématique des techniques de pointe : résumé et introduction

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps