Some of Our Sources
- BoingBoing
- Techcrunch
- Simplebits
- Joshua Blankenship
- Six Revisions
- Fuel Your Creativity
- Web Designer Depot
- Stylized Web
- Wal You
- TechPowerUp
Help Webnuz
Referal links:
Message Queue Service using Kafka
In this article, we are going to create a Message Queue Service using Kafka and KafkaJS, somewhere similar to SQS, and we will make sure that messages are processed exactly once, in the exact order that they are sent. We will also implement the redelivery mechanism to make sure that if something fails in our business logic, we can add it to the queue again.
Find the source code on Github:
https://github.com/icpsoni/kafka-message-queue
Prerequisites
1: Get Kafka
Download Kafka and extract it from here, and navigate to the directory.
$ tar -xzf <file_name>.tgz$ cd <file_name>tec2: Start the Kafka Environment
NOTE: Your local environment must have Java 8+ installed.
Run the following commands to start ZooKeeper:
# Start the ZooKeeper service# Note: Soon, ZooKeeper will no longer be required by Apache Kafka.$ bin/zookeeper-server-start.sh config/zookeeper.propertiesOpen another terminal session and run Kafka Broker service using:
# Start the Kafka broker service$ bin/kafka-server-start.sh config/server.propertiesOnce all services have successfully launched, you will have a basic Kafka environment running and ready to use.
3: Create a Topic to Store our Queue Messages
With the following command, we will create a topic called message-queue in our Kafka.
$ bin/kafka-topics.sh --create --topic message-queue --bootstrap-server localhost:9092To check the created topic use this command.
$ bin/kafka-topics.sh --describe --topic message-queue --bootstrap-server localhost:9092You can create a topic with any number of partitions, In here we are using 1 partition only.
Creating Services Using KafkaJS
We basically need 3 things to make our queue system work perfectly.
- Sending Events to Kafka Queue topic.
- Subscribing to the topic and reading the Queue Message.
- Handling the redelivery.
kafka-config.js
The following snippet contains the basic config of Kafka that we need for our queue system.
// Using KafkaJs nodejs libraryimport { Kafka } from 'kafkajs';// kafka broker running on localhost:9092 default portconst kafkaBroker = 'localhost:9092';// kafka topic used for queue messagesexport const kafkaTopic = 'message-queue';// kafka client with basic configexport const KafkaClient = new Kafka({ brokers: [kafkaBroker]});producer.js
This service will send messages to the Kafka topic we created earlier. We can use sendMessageToQueue function and pass the message object which needs to be sent to Kafka.
import { KafkaClient } from './index.js';import { kafkaTopic } from "./kafka-config";export const sendMessageToQueue = async (message) => { const producer = KafkaClient.producer(); await producer.connect(); await producer.send({ topic: kafkaTopic, messages: [ { value: message // Your message data goes here } ] }); // Disconnect producer once message sending is done. await producer.disconnect();};message-queue.js
This is our main service that takes care of receiving the messages from the Kafka queue doing the business logic and handling the redelivery if something goes wrong in business logic.
import { sendMessageToQueue } from "./producer.js";import { KafkaClient, kafkaTopic, kafkaGroupId } from "./kafka-config.js";export const consumeMessage = async () => { // Creating a Consumer Instance const consumer = KafkaClient.consumer({ groupId: kafkaGroupId, }); await consumer.connect(); // Subscribing to out Kafka topic await consumer.subscribe({ topic: kafkaTopic, fromBeginning: true}); await consumer.run({ autoCommit: false, // It won't commit message acknowledge to kafka until we don't do manually eachMessage: async ({ topic, partition, message}) => { const messageData = message.value.toString(); try { // Do the business Logic console.info('Received Message', messageData); } catch (error) { console.error(error); // Resending message to kafka queue for redelivery await sendMessageToQueue(messageData); } finally { const offset = +message.offset + 1; // Committing the message offset to Kafka await consumer.commitOffsets([{topic: kafkaTopic, partition, offset: offset.toString()}]); } } });};Download the source
Bonus: It contains test sample files too.
https://github.com/icpsoni/kafka-message-queue
References:
Original Link: https://dev.to/icpsoni/message-queue-service-using-kafka-2bli
Dev To
An online community for sharing and discovering great ideas, having debates, and making friendsMore About this Source Visit Dev To