हममें से कई लोग संदेश प्रकाशित करने के लिए काफ्का का उपयोग करते हैं, लेकिन हम उन्हें कैसे प्राप्त करते हैं? इस लेख में, हम काफ्का के संदेशों के उपभोग के लिए एक छोटा सा एप्लिकेशन लिखेंगे। और, ज़ाहिर है, e2e परीक्षण।  आइए सबसे पहले समझें कि काफ्का कैसे काम करता है और यह क्या है।  काफ्का एक संदेश दलाल है जिसमें कुछ सेवाएँ संदेश उत्पन्न करती हैं और अन्य उन्हें प्राप्त करती हैं। ब्रोकरों का उपयोग मुख्य रूप से सेवाओं के बीच संदेश भेजने के लिए माइक्रोसर्विस आर्किटेक्चर वाले सिस्टम में किया जाता है।  संदेशों को विषयों में संग्रहीत किया जाता है. संदेश भेजते समय, निर्माता विषय के नाम के साथ-साथ संदेश को भी इंगित करता है, जिसमें एक कुंजी और एक मान होता है। और बस; निर्माता का काम ख़त्म हो गया है.   फिर उपभोक्ता खेल में आते हैं, वे वांछित विषय की सदस्यता लेते हैं, और संदेश पढ़ना शुरू करते हैं। प्रत्येक एप्लिकेशन की अपनी कतार होती है, जिसे पढ़कर उपभोक्ता ऑफसेट पॉइंटर को स्थानांतरित करता है।   काफ्का की विशिष्ट विशेषताएं हैं:  गारंटी दें कि सभी संदेशों को ठीक उसी क्रम में क्रमबद्ध किया जाएगा जिस क्रम में वे विषय में आए थे  काफ्का स्टोर कुछ देर के लिए संदेश पढ़ते हैं  उच्च थ्रूपुट  अब, आइए NestJs ढांचे का उपयोग करके काफ्का के साथ काम करें। सबसे पहले, हमें एक नियंत्रक बनाना होगा जो संदेशों को संसाधित करेगा।   @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); } }    विशेषता पर ध्यान दें, जो इंगित करता है कि हमारा   फ़ंक्शन कॉन्फ़िगरेशन फ़ाइल   में निर्दिष्ट विषय से संदेश प्राप्त करेगा।   विशेषता विषय संदेश से मूल्य प्राप्त करने में मदद करती है। @EventPattern handleEvent() config.get('kafka.topics.exampleTopic') @Payload()  अपने एप्लिकेशन को काफ्का दलालों से जोड़ने के लिए, आपको दो काम करने होंगे। आरंभ करने के लिए, स्टार्टअप फ़ाइल में माइक्रोसर्विस कनेक्ट करें:   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'), }, }, });  और फिर माइक्रोसर्विसेज को 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();  एप्लिकेशन का परीक्षण करने के लिए, मैं @testcontainers/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, ]; }  कृपया ध्यान दें कि, इस फ़ाइल में, मैंने ब्रोकर पते को नव निर्मित कंटेनर में ओवरराइड कर दिया है।  परीक्षण फ़ाइल में ही,   फ़ंक्शन में, मैं एक काफ्का क्लाइंट बनाता हूँ। निर्माता के साथ, मैं एक विषय भी बनाता हूं और अपना एप्लिकेशन लॉन्च करता हूं। 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);  बेशक,   फ़ंक्शन में, आपको कंटेनरों को रोकने की आवश्यकता है: afterAll   afterAll(async () => { await app.close(); await Promise.all(kafkaContainers.map(c => c.stop())); }, 15 * 1000);  मैंने एक परीक्षण लिखा जो सत्यापित करता है कि जब किसी विषय पर कोई संदेश आता है, तो नियंत्रक से हमारा हैंडलर फ़ंक्शन आवश्यक सेवा फ़ंक्शन को कॉल करता है। ऐसा करने के लिए, मैं   फ़ंक्शन के कार्यान्वयन को ओवरराइड करता हूं और इसके कॉल होने की प्रतीक्षा करता हूं। 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(); }); });  बस इतना ही। यदि आप NestJs ढांचे का उपयोग करते हैं तो काफ्का के साथ काम करना अविश्वसनीय रूप से आसान है। मुझे आशा है कि मेरा अनुभव आपके काम आएगा। एक उदाहरण एप्लिकेशन   पर देखा जा सकता है 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

यह ऑडियो कहानी की मूल भाषा में निर्मित है!

बहुत लंबा; पढ़ने के लिए

Make resilience your competitive advantage

NestJS के साथ काफ्का संदेशों का उपभोग कैसे करें

About Author

टिप्पणियाँ

लेबल

इस लेख में चित्रित किया गया था

Related Stories

Paint us Red and Call us Santa! Our Devs Just Gifted You a Sleigh of New Features!

टेलीग्राम: क्रिप्टो द्वीप का मुख्य भूमि से पुल

क्रिप्टो ग्रोथ: प्रभावी उपयोगकर्ता व्यक्तित्व बनाना

जलयात्रा: डेटा झीलों के साथ उत्पादन-ग्रेड आरएजी अनुप्रयोगों का विकास

Paint us Red and Call us Santa! Our Devs Just Gifted You a Sleigh of New Features!

टेलीग्राम: क्रिप्टो द्वीप का मुख्य भूमि से पुल

क्रिप्टो ग्रोथ: प्रभावी उपयोगकर्ता व्यक्तित्व बनाना

जलयात्रा: डेटा झीलों के साथ उत्पादन-ग्रेड आरएजी अनुप्रयोगों का विकास

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps