KOA+egg.js集成kafka消息队列的示例

下面是关于KOA+egg.js集成kafka消息队列的完整攻略。

一、什么是Kafka

Kafka是一个高吞吐量的分布式队列系统，被广泛应用于大规模数据处理和处理高并发请求的场景。

二、集成kafka消息队列方案

KOA+egg.js集成kafka消息队列，需要用到kafka-node和egg-kafkanode插件。

其中，kafka-node是kafka的js实现，提供操作kafka的API，而egg-kafkanode是用于egg框架的kafka插件，主要提供config配置和消费者和生产者的接入封装。

下面介绍集成kafka消息队列的步骤：

1.安装kafka-node和egg-kafkanode

npm install kafka-node egg-kafkanode --save

2.配置kafka

在egg.js的config.default.js中增加以下配置：

config.kafka = {
    client: {
        kafkaHost: 'localhost:9092' // kafka服务器地址
    },
    topics: [
        {
            topic: 'test', // 创建一个名为test的topic
            partitions: 1, // 设置该topic分区数为1
            // 用于设置分区的key和value
            config: {
                'cleanup.policy': 'compact',
                'compression.type': 'gzip',
                'segment.ms': 100
            }
        }
    ]
};

3.创建kafka客户端

在config.default.js下增加以下配置：

const kafka = require('kafka-node');
const client = new kafka.KafkaClient(config.kafka.client);
config.kafka.client.clientId = client.clientId; // 将clientId记录到配置对象中

4.配置和启动kafka消费者

在app.js中增加消费者的配置和启动代码：

const ConsumerGroup = require('egg-kafkanode').ConsumerGroup;
const topics = config.kafka.topics.map(item => item.topic);
const consumerGroupOptions = {
    kafkaHost: config.kafka.client.kafkaHost,
    groupId: 'test-consumer', // 消费者组名
    sessionTimeout: 15000,
    protocol: ['roundrobin'],
    encoding: 'utf8',
    fromOffset: 'latest'
};
const consumerGroup = new ConsumerGroup(Object.assign({}, consumerGroupOptions, { topics }), topics);
consumerGroup.on('message', message => {
    console.log("receive kafka message:\n", message);
});

5.配置和发送kafka消息

在KOA中编写代码，完成发送消息的操作：

const Producer = require('kafka-node').Producer;
const payloads = [
    {
        topic: 'test',
        messages: 'test kafka message'
    }
];
const producer = new Producer(client);
producer.on('ready', _ => {
    producer.send(payloads, (err, data) => {
        console.log('kafka message gots sent:', data);
    });
});

以上就是集成kafka消息队列的完整攻略，下面提供两条示例说明。

三、示例说明

1.简单的kafka消息发布订阅

1.1 增加发送代码

const payloads = [
    {
        topic: 'test',
        messages: 'test kafka message'
    }
];
const producer = new Producer(client);
producer.on('ready', _ => {
    producer.send(payloads, (err, data) => {
        console.log('kafka message gots sent:', data);
    });
});

1.2 增加消费代码

const ConsumerGroup = require('egg-kafkanode').ConsumerGroup;
const topics = config.kafka.topics.map(item => item.topic);
const consumerGroupOptions = {
    kafkaHost: config.kafka.client.kafkaHost,
    groupId: 'test-consumer', // 消费者组名
    sessionTimeout: 15000,
    protocol: ['roundrobin'],
    encoding: 'utf8',
    fromOffset: 'latest'
};
const consumerGroup = new ConsumerGroup(Object.assign({}, consumerGroupOptions, { topics }), topics);
consumerGroup.on('message', message => {
    console.log("receive kafka message:\n", message);
});

2.使用kafka实现KOA和其他系统的异步通信

2.1 KOA中发送消息代码

在controller文件中加入以下代码：

ctx.app.kafkaProducer.send([
    {
        topic: 'test',
        messages: JSON.stringify({
            type: 'test',
            payload: 'message from koa'
        })
    }
], (err, data) => {
    console.log('kafka message gots sent:', data);
});

2.2 其他系统中接收消息代码

在其他系统中编写以下代码：

const ConsumerGroup = require('egg-kafkanode').ConsumerGroup;
const topics = config.kafka.topics.map(item => item.topic);
const consumerGroupOptions = {
    kafkaHost: config.kafka.client.kafkaHost,
    groupId: 'test-consumer', // 消费者组名
    sessionTimeout: 15000,
    protocol: ['roundrobin'],
    encoding: 'utf8',
    fromOffset: 'latest'
};
const consumerGroup = new ConsumerGroup(Object.assign({}, consumerGroupOptions, { topics }), topics);
consumerGroup.on('message', message => {
    const msg = JSON.parse(message.value);
    console.log('received message:\n', msg);
});

以上就是两条示例说明，第一个示例演示了简单的kafka消息发布订阅，第二个示例演示了如何使用kafka实现KOA和其他系统的异步通信。