FR
RU
EN
ES
ZH
DE
PT
AR
HI
Bibliothèque KafkaJS : travailler avec Apache Kafka en TypeScript
\n\nApache Kafka est une plateforme distribuée de traitement de flux de données, largement utilisée pour construire des systèmes d'échange de messages à forte charge. Dans l'écosystème TypeScript (et Node.js), la bibliothèque KafkaJS est devenue la norme de facto. Elle offre une API propre et moderne avec un support complet de TypeScript, permettant aux développeurs d'écrire du code typé sécurisé pour les producteurs et consommateurs Kafka.
\n\nDans cet article, nous allons détailler l'installation, la configuration et les scénarios d'utilisation principaux de KafkaJS. Vous apprendrez à envoyer et recevoir des messages, à gérer les erreurs et à configurer correctement le client pour un environnement de production.
\n\nInstallation et configuration de KafkaJS
\n\nPour commencer, installez la bibliothèque via npm ou yarn :
\n\nnpm install kafkajs\n# ou\nyarn add kafkajsKafkaJS est écrite en TypeScript, donc les types sont déjà intégrés — il n'est pas nécessaire d'installer @types/kafkajs supplémentaire.
Créons maintenant une instance du client Kafka. Il est recommandé de spécifier plusieurs courtiers pour la tolérance aux pannes :
\n\nimport { Kafka, logLevel } from 'kafkajs';\n\nconst kafka = new Kafka({\n clientId: 'my-app',\n brokers: ['localhost:9092', 'localhost:9093'],\n logLevel: logLevel.INFO, // Niveau de journalisation\n retry: {\n initialRetryTime: 100,\n retries: 8\n }\n});Producteur : envoi de messages à Kafka
\n\nLe producteur (producer) est responsable de l'écriture des messages dans les topics Kafka. KafkaJS fournit une API pratique pour envoyer des messages individuels ainsi que des lots.
\n\nExemple d'un producteur de base :
\n\nimport { Kafka, Producer } from 'kafkajs';\n\nasync function produceMessage(producer: Producer) {\n await producer.send({\n topic: 'orders',\n messages: [\n { \n key: 'order-123', \n value: JSON.stringify({ userId: 42, amount: 1500 }) \n },\n ],\n });\n console.log('Message envoyé');\n}\n\nasync function runProducer() {\n const producer = kafka.producer();\n await producer.connect();\n\n // Envoi de plusieurs messages\n await producer.send({\n topic: 'orders',\n messages: [\n { key: 'order-124', value: '{"item":"laptop"}' },\n { key: 'order-125', value: '{"item":"mouse"}' },\n ],\n });\n\n await produceMessage(producer);\n await producer.disconnect();\n}\n\nrunProducer().catch(console.error);Important : Appelez toujours disconnect() après avoir terminé pour libérer les ressources. Pour les applications de longue durée, le producteur peut rester connecté en permanence.
Consommateur : lecture des messages depuis Kafka
\n\nLe consommateur (consumer) s'abonne à un ou plusieurs topics et traite les messages. KafkaJS prend en charge les groupes de consommateurs pour le traitement parallèle.
\n\nExemple d'un consommateur avec traitement des messages :
\n\nimport { Kafka, Consumer, EachMessagePayload } from 'kafkajs';\n\nasync function runConsumer() {\n const consumer: Consumer = kafka.consumer({ \n groupId: 'order-processor-group' \n });\n\n await consumer.connect();\n await consumer.subscribe({ \n topic: 'orders', \n fromBeginning: true // Lire depuis le tout premier message\n });\n\n await consumer.run({\n eachMessage: async ({ topic, partition, message }: EachMessagePayload) => {\n const key = message.key?.toString();\n const value = message.value?.toString();\n console.log({\n partition,\n offset: message.offset,\n key,\n value: value ? JSON.parse(value) : null,\n });\n },\n });\n}\n\nrunConsumer().catch(console.error);Notez le paramètre fromBeginning : s'il est défini sur true, le consommateur lira tous les messages précédemment accumulés. Pour un environnement de production, on utilise généralement false (uniquement les nouveaux messages).
