DEV Community

loading...
Cover image for Publish/Subscribe pattern example (Redis, Kafka)

Publish/Subscribe pattern example (Redis, Kafka)

Ronny Medina
I like to learn new things every day and play games... https://www.youtube.com/channel/UCJMCTQnAbfMYt0PnBK3cdBg
Updated on ・4 min read

Hi, in tutorial I'm goin to show you how to implement this pattern. I won't explain details about Kafka or Redis. Then main idea here is to show you how to create simple implementation with these technologies.

To this tutorial we going to use Docker and docker-compose is necessary if you want to follow this tutorial.

Explain

The publish/subscribe pattern is very easy to understand. The main idea is to has central point where all the applications will put the information and then it will be consumed by other applications. You can see the imagen below.

publish and subscribe pattern

Depends on the technologies we're using there're some terms that will change. For example in Redis to classify our information they call it a channel. In Kafka they call it topic.

Topics or channels is the way to classify information. For example, we will save information about the login process. We can create a (topic|chennel) called login_info and another to save then transaction information of a payment called transaction_payment.

Before start we need to install some dependencies.

npm install express ioredis kafkajs
Enter fullscreen mode Exit fullscreen mode

docker-compose (Redis)

We need to create a docker-compose.yml and put this content.


version: "3.9"
services:
  redis-service:
    image: "redis:alpine"
    container_name: redis-service
    ports:
      - "6379:6379"

Enter fullscreen mode Exit fullscreen mode

Publisher (Redis)

publisher.js. You can create an endpoint to send the information at moment. For me this approach works.

const express = require('express')
const Redis = require('ioredis')

const app = express()
const port = 3000
const redis = new Redis()

app.listen(port, async () => {
  const message = { message: 'message from redis' }
  redis.publish('test', JSON.stringify(message))
  console.log(`Example app (publisher) at http://localhost:${port}`)
})
Enter fullscreen mode Exit fullscreen mode

Subscriber (Redis)

subscriber.js


const express = require('express')
const Redis = require('ioredis')

const app = express()
const port = 3500

const redis = new Redis()

app.listen(port, async () => {
  await redis.subscribe('test')

  redis.on('message', (channel, message) => {
    console.log(`Received ${message} from ${channel} (Redis)`)
  })

  console.log(`Example app (subscribe) at http://localhost:${port}`)
})
Enter fullscreen mode Exit fullscreen mode

Now need to execute the following commands.


$ docker-compose up -d

$ node subscriber.js

$ node publisher.js // open in another terminal
Enter fullscreen mode Exit fullscreen mode

You should now see a message in your terminal (subscriber). That was very easy =). Now we can do the same process to implement Kafka.

docker-compose (Kafka)

To implement Kafka we need to download a docker-compose.yml file from confluent page https://docs.confluent.io/platform/current/quickstart/ce-docker-quickstart.html.

We can use the entire docker-compose file with all services. But it is not necessary for this tutorial. The fundamental services to start working with Kafka are zookeeper and broker.


version: "3.9"
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:6.2.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-server:6.2.0
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'true'
      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'
Enter fullscreen mode Exit fullscreen mode

Publisher (Kafka)


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

const app = express()
const port = 3000


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

app.listen(port, () => {
  const producer = kafka.producer()
  await producer.connect()
  await producer.send({
    topic: 'test',
    messages: [{ value: 'Hello KafkaJS user!' }],
  })

  await producer.disconnect()

  console.log(`Example app (publisher) at http://localhost:${port}`)
})

Enter fullscreen mode Exit fullscreen mode

Subscriber (Kafka)

Before I begin, I need to clarify that this code could be anywhere you want.


  const admin = kafka.admin()

  try {
    await admin.connect()

    await admin.createTopics({
      waitForLeaders: true,
      topics: [{topic: 'test' }],
    })
  } catch(err) {
    console.error(err)
  } finally {
    await admin.disconnect()
  }
Enter fullscreen mode Exit fullscreen mode

Kafka needs setup before starting. For example we need to create the topics before. Kafka has more settings unlike Redis. I put this content inside subscriber.js if you run this code multiple times this throws an error because the topic already exists.


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

const app = express()
const port = 3500

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

app.listen(port, async () => {
  const admin = kafka.admin()

  try {
    await admin.connect()

    await admin.createTopics({
      waitForLeaders: true,
      topics: [{topic: 'test' }],
    })
  } catch(err) {
    console.error(err)
  } finally {
    await admin.disconnect()
  }

  // kafka
  const consumer = kafka.consumer({ groupId: 'test-group' })
  await consumer.connect()
  await consumer.subscribe({ topic: 'test', fromBeginning: true })
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log(`Received ${message.value.toString()} from channel ${topic} (Kafka)`)
    },
  })

  console.log(`Example app (subscribe) at http://localhost:${port}`)
})
Enter fullscreen mode Exit fullscreen mode

That's it to be able to work with Kafka. Now we can run the same commands we saw before.


$ docker-compose up -d

$ node subscriber.js

$ node publisher.js // open in another terminal
Enter fullscreen mode Exit fullscreen mode

I hope it was useful for you and that you can start working with these technologies. If you have any corrections, let me know.

Discussion (0)