Riky Fahri Hasibuan

Posted on Aug 9 • Originally published at codenoun.com

Implement Kafka and Node.js in Microservice Architecture

#javascript #eventdriven #webdev #microservices

When designing microservices architecture for event-driven applications, integrating Apache Kafka and Node.js can significantly enhance real-time data processing capabilities. In this article, we'll explore how to leverage Kafka Node.js integration to build robust and scalable microservices that handle streaming data efficiently.

Why Use Apache Kafka in a Microservices Architecture?

In a microservices architecture, services need to communicate with each other efficiently. Apache Kafka serves as a distributed event streaming platform that enables real-time data exchange between microservices. It decouples the services, allowing them to operate independently while processing large volumes of data.

Benefits of Kafka in Event-Driven Applications

Scalability: Kafka's distributed architecture supports horizontal scaling, making it ideal for real-time data processing in event-driven applications.
Fault Tolerance: Kafka ensures that data is reliably delivered, even in the event of failures.
High Throughput: Kafka can handle millions of events per second, providing high throughput for demanding microservices applications.

Setting Up Kafka Node.js Integration

To integrate Apache Kafka and Node.js in a microservices environment, you'll need to set up Kafka as a message broker and connect it with your Node.js services. Here's a step-by-step guide:

Install Kafka and Node.js

First, ensure that Apache Kafka and Node.js are installed on your system. You can install Kafka & Node.js by following the following articles:

Install Kafka Node.js Client Library

To connect Node.js with Kafka, you can use the kafkajs library, a popular Kafka client for Node.js.

npm install kafkajs

Create a Kafka Producer in Node.js

In a microservices architecture, a Kafka producer is responsible for sending messages to a Kafka topic. Below is a simple example of how to create a Kafka producer in Node.js:

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-producer',
  brokers: ['localhost:9092']
});

const producer = kafka.producer();

const sendMessage = async () => {
  await producer.connect();
  await producer.send({
    topic: 'my-topic',
    messages: [
      { value: 'Hello Kafka' },
    ],
  });
  await producer.disconnect();
};

sendMessage().catch(console.error);

Create a Kafka Consumer in Node.js

A Kafka consumer is used to read messages from a Kafka topic. Here’s how you can create a consumer:

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-consumer',
  brokers: ['localhost:9092']
});

const consumer = kafka.consumer({ groupId: 'my-group' });

const runConsumer = async () => {
  await consumer.connect();
  await consumer.subscribe({ topic: 'my-topic', fromBeginning: true });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log({
        partition,
        offset: message.offset,
        value: message.value.toString(),
      });
    },
  });
};

runConsumer().catch(console.error);

Case Study

To illustrate the integration of Kafka and Node.js in a microservice architecture, consider the following case study:

Scenario

We have two microservices:

Order Service: Handles customer orders.
Product Service: Manages product stocks.

Whenever a purchase or transaction occurs in the Order Service, it will to update the stock in the Product Service. Kafka facilitates this communication by acting as a message broker.

Implementation

Order Service: Publishes order events to the product-updates topic.
Inventory Service: Consumes messages from the product-updates topic and updates the inventory accordingly.

Order Service Producer Script

The Order Service is responsible for handling purchase orders and sending messages to the Product Service to update the stock. Here's how you can implement the Order Service as a Kafka producer:

// orderService.js
const express = require('express');
const { Kafka } = require('kafkajs');

// Kafka producer configuration
const kafka = new Kafka({
  clientId: 'order-service',
  brokers: ['localhost:9092'],
});

const producer = kafka.producer();

// Initialize Express app
const app = express();
app.use(express.json());

const placeOrder = async (orderId, productId, quantity) => {
  await producer.connect();
  const orderEvent = {
    orderId,
    productId,
    quantity,
    eventType: 'ORDER_PLACED',
    timestamp: Date.now(),
  };
  await producer.send({
    topic: 'product-updates',
    messages: [{ value: JSON.stringify(orderEvent) }],
  });
  await producer.disconnect();
  console.log(`Order placed: ${orderId} for product: ${productId}`);
};

// API endpoint to place an order
app.post('/order', async (req, res) => {
  const { orderId, productId, quantity } = req.body;

  if (!orderId || !productId || !quantity) {
    return res.status(400).json({ error: 'Missing orderId, productId, or quantity' });
  }

  try {
    await placeOrder(orderId, productId, quantity);
    res.status(200).json({ message: `Order ${orderId} placed successfully.` });
  } catch (error) {
    console.error('Error placing order:', error);
    res.status(500).json({ error: 'Failed to place order' });
  }
});

// Start the server
const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`Order Service API running on port ${PORT}`);
});

Product Service Consumer Script

The Product Service consumes messages from the product-updates Kafka topic and updates the product stock accordingly. Here's the implementation:

// productService.js
const express = require('express');
const { Kafka } = require('kafkajs');

// Kafka consumer configuration
const kafka = new Kafka({
  clientId: 'product-service',
  brokers: ['localhost:9092'],
});

const consumer = kafka.consumer({ groupId: 'product-group' });

// Initialize Express app
const app = express();
app.use(express.json());

const updateStock = async () => {
  await consumer.connect();
  await consumer.subscribe({ topic: 'product-updates', fromBeginning: true });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const orderEvent = JSON.parse(message.value.toString());
      console.log(`Received order: ${orderEvent.orderId}, Product: ${orderEvent.productId}, Quantity: ${orderEvent.quantity}`);

      // Simulate stock update
      console.log(`Updating stock for product: ${orderEvent.productId}`);
      // logic to update stock
    },
  });
};

// Start the Product Service to listen for messages
updateStock().catch(console.error);

// Start the server
const PORT = process.env.PORT || 3001;
app.listen(PORT, () => {
  console.log(`Product Service API running on port ${PORT}`);
});

Start the Product Service first, as it needs to listen for incoming messages:

node productService.js

The Product Service will start listening on port 3001 (or another port if specified).

Start the Order Service with this command:

node orderService.js

The Order Service will be available on port 3000 (or another port if specified).

You can place an order by sending a POST request to the Order Service API:

curl -X POST http://localhost:3000/order \
-H "Content-Type: application/json" \
-d '{
  "orderId": "order-789",
  "productId": "product-123",
  "quantity": 5
}'

When an order is placed, the Order Service will send a Kafka message, and the Product Service will consume that message to update the stock:

Received order: order-789, Product: product-123, Quantity: 5
Updating stock for product: product-123

Conclusion

Integrating Apache Kafka and Node.js in your microservices architecture allows you to build highly scalable and resilient event-driven applications.

By following best practices and leveraging Kafka’s powerful features, you can efficiently process real-time data and create a robust communication layer between your microservices.

DEV Community