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

egg.js ： 基于koa2的企业级框架

kafka：高吞吐量的分布式发布订阅消息系统

本文章将集成egg + kafka + mysql 的日志系统例子

系统要求：日志记录，通过kafka进行消息队列控制

思路图：

这里消费者和生产者都由日志系统提供

λ.1 环境准备

①kafka

官网下载kafka后，解压

启动zookeeper：

bin/zookeeper-server-start.sh config/zookeeper.properties

启动kafka server

这里config/server.properties中将num.partitions=5，我们设置5个partitions

bin/kafka-server-start.sh config/server.properties

② egg + mysql

根据脚手架搭建好egg，再多安装kafka-node，egg-mysql

mysql 用户名root 密码123456

λ.2 集成

1、根目录新建app.js，这个文件在每次项目加载时候都会运作

'use strict';
 
const kafka = require('kafka-node');
 
module.exports = app => {
 app.beforestart(async () => {
 const ctx = app.createanonymouscontext();
 
 const producer = kafka.producer;
 const client = new kafka.kafkaclient({ kafkahost: app.config.kafkahost });
 const producer = new producer(client, app.config.producerconfig);
 
 producer.on('error', function(err) {
  console.error('error: [producer] ' + err);
 });
 
 app.producer = producer;
 
 const consumer = new kafka.consumer(client, app.config.consumertopics, {
  autocommit: false,
 });
 
 consumer.on('message', async function(message) {
  try {
  await ctx.service.log.insert(json.parse(message.value));
  consumer.commit(true, (err, data) => {
   console.error('commit:', err, data);
  });
  } catch (error) {
  console.error('error: [getmessage] ', message, error);
  }
 });
 
 consumer.on('error', function(err) {
  console.error('error: [consumer] ' + err);
 });
 });
};

上述代码新建了生产者、消费者。

生产者新建后加载进app全局对象。我们将在请求时候生产消息。这里只是先新建实例

消费者获取消息将访问service层的insert方法（数据库插入数据）。

具体参数可以参考kafka-node官方api，往下看会有生产者和消费者的配置参数。

2、controller · log.js

这里获取到了producer，并传往service层

'use strict';
 
const controller = require('egg').controller;
 
class logcontroller extends controller {
 /**
 * @description kafka控制日志信息流
 * @host /log/notice
 * @method post
 * @param {log} log 日志信息
 */
 async notice() {
 const producer = this.ctx.app.producer;
 const response = new this.ctx.app.response();
 
 const requestbody = this.ctx.request.body;
 const backinfo = await this.ctx.service.log.send(producer, requestbody);
 this.ctx.body = response.success(backinfo);
 }
}
 
module.exports = logcontroller;

3、service · log.js

这里有一个send方法，这里调用了producer.send ，进行生产者生产

insert方法则是数据库插入数据

'use strict';
 
const service = require('egg').service;
const uuidv1 = require('uuid/v1');
 
class logservice extends service {
 async send(producer, params) {
 const payloads = [
  {
  topic: this.ctx.app.config.topic,
  messages: json.stringify(params),
  },
 ];
 
 producer.send(payloads, function(err, data) {
  console.log('send : ', data);
 });
 
 return 'success';
 }
 async insert(message) {
 try {
  const logdb = this.ctx.app.mysql.get('log');
  const ip = this.ctx.ip;
 
  const logs = this.ctx.model.log.build({
  id: uuidv1(),
  type: message.type || '',
  level: message.level || 0,
  operator: message.operator || '',
  content: message.content || '',
  ip,
  user_agent: message.user_agent || '',
  error_stack: message.error_stack || '',
  url: message.url || '',
  request: message.request || '',
  response: message.response || '',
  created_at: new date(),
  updated_at: new date(),
  });
 
  const result = await logdb.insert('logs', logs.datavalues);
 
  if (result.affectedrows === 1) {
  console.log(`suceess: [insert ${message.type}]`);
  } else console.error('error: [insert db] ', result);
 } catch (error) {
  console.error('error: [insert] ', message, error);
 }
 }
}
 
module.exports = logservice;

4、config · config.default.js

一些上述代码用到的配置参数具体在这里，注这里开了5个partition。

'use strict';
 
module.exports = appinfo => {
 const config = (exports = {});
 
 const topic = 'logaction_p5';
 
 // add your config here
 config.middleware = [];
 
 config.security = {
 csrf: {
  enable: false,
 },
 };
 
 // mysql database configuration
 config.mysql = {
 clients: {
  basic: {
  host: 'localhost',
  port: '3306',
  user: 'root',
  password: '123456',
  database: 'merchants_basic',
  },
  log: {
  host: 'localhost',
  port: '3306',
  user: 'root',
  password: '123456',
  database: 'merchants_log',
  },
 },
 default: {},
 app: true,
 agent: false,
 };
 
 // sequelize config
 config.sequelize = {
 dialect: 'mysql',
 database: 'merchants_log',
 host: 'localhost',
 port: '3306',
 username: 'root',
 password: '123456',
 dialectoptions: {
  requesttimeout: 999999,
 },
 pool: {
  acquire: 999999,
 },
 };
 
 // kafka config
 config.kafkahost = 'localhost:9092';
 
 config.topic = topic;
 
 config.producerconfig = {
 // partitioner type (default = 0, random = 1, cyclic = 2, keyed = 3, custom = 4), default 0
 partitionertype: 1,
 };
 
 config.consumertopics = [
 { topic, partition: 0 },
 { topic, partition: 1 },
 { topic, partition: 2 },
 { topic, partition: 3 },
 { topic, partition: 4 },
 ];
 
 return config;
};

5、实体类：

mode · log.js

这里使用了 sequelize

'use strict';
 
module.exports = app => {
 const { string, integer, date, text } = app.sequelize;
 
 const log = app.model.define('log', {
 /**
  * uuid
  */
 id: { type: string(36), primarykey: true },
 /**
  * 日志类型
  */
 type: string(100),
 /**
  * 优先等级(数字越高，优先级越高）
  */
 level: integer,
 /**
  * 操作者
  */
 operator: string(50),
 /**
  * 日志内容
  */
 content: text,
 /**
  * ip
  */
 ip: string(36),
 /**
  * 当前用户代理信息
  */
 user_agent: string(150),
 /**
  * 错误堆栈
  */
 error_stack: text,
 /**
  * url
  */
 url: string(255),
 /**
  * 请求对象
  */
 request: text,
 /**
  * 响应对象
  */
 response: text,
 /**
  * 创建时间
  */
 created_at: date,
 /**
  * 更新时间
  */
 updated_at: date,
 });
 
 return log;
};

6、测试python脚本：

import requests
 
from multiprocessing import pool
from threading import thread
 
from multiprocessing import process
 
 
def loop():
 t = 1000
 while t:
  url = "http://localhost:7001/log/notice"
 
  payload = "{\n\t\"type\": \"error\",\n\t\"level\": 1,\n\t\"content\": \"url send error\",\n\t\"operator\": \"knove\"\n}"
  headers = {
  'content-type': "application/json",
  'cache-control': "no-cache"
  }
 
  response = requests.request("post", url, data=payload, headers=headers)
 
  print(response.text)
 
if __name__ == '__main__':
 for i in range(10):
  t = thread(target=loop)
  t.start()

7、建表语句：

set names utf8mb4;
set foreign_key_checks = 0;
 
-- ----------------------------
-- table structure for logs
-- ----------------------------
drop table if exists `logs`;
create table `logs` (
 `id` varchar(36) character set utf8mb4 collate utf8mb4_bin not null,
 `type` varchar(100) character set utf8mb4 collate utf8mb4_bin not null comment '日志类型',
 `level` int(11) null default null comment '优先等级(数字越高，优先级越高）',
 `operator` varchar(50) character set utf8mb4 collate utf8mb4_bin null default null comment '操作人',
 `content` text character set utf8mb4 collate utf8mb4_bin null comment '日志信息',
 `ip` varchar(36) character set utf8mb4 collate utf8mb4_bin null default null comment 'ip\r\nip',
 `user_agent` varchar(150) character set utf8mb4 collate utf8mb4_bin null default null comment '当前用户代理信息',
 `error_stack` text character set utf8mb4 collate utf8mb4_bin null comment '错误堆栈',
 `url` varchar(255) character set utf8mb4 collate utf8mb4_bin null default null comment '当前url',
 `request` text character set utf8mb4 collate utf8mb4_bin null comment '请求对象',
 `response` text character set utf8mb4 collate utf8mb4_bin null comment '响应对象',
 `created_at` datetime(0) null default null comment '创建时间',
 `updated_at` datetime(0) null default null comment '更新时间',
 primary key (`id`) using btree
) engine = innodb character set = utf8mb4 collate = utf8mb4_bin row_format = dynamic;
 
set foreign_key_checks = 1;

λ.3 后话

网上类似资料甚少，啃各种文档，探寻技术实现方式

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。