KOA + egg.js 集成 kafka 消息隊列

時間  2019-11-24
標籤 koa egg.js egg 集成 kafka 消息 隊列 欄目 JavaScript 简体版
原文   原文鏈接

Egg.js ： 基於KOA2的企業級框架javascript

Kafka：高吞吐量的分佈式發佈訂閱消息系統java

本文章將集成egg + kafka + mysql 的日誌系統例子node

系統要求：日誌記錄，經過kafka進行消息隊列控制python

思路圖：mysql

這裏消費者和生產者都由日誌系統提供

λ.1 環境準備

①Kafka

官網下載kafka後，解壓sql

啓動zookeeper：數據庫

bin/zookeeper-server-start.sh config/zookeeper.properties
複製代碼

啓動Kafka serverjson

這裏config/server.properties中將num.partitions=5，咱們設置5個partitionsbash

bin/kafka-server-start.sh config/server.properties
複製代碼

② egg + mysql

根據腳手架搭建好egg，再多安裝kafka-node，egg-mysqlapp

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，往下看會有生產者和消費者的配置參數。

  1. 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;
複製代碼
  1. 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;
複製代碼
  1. 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;
};
複製代碼
  1. 實體類：

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;
};

複製代碼
  1. 測試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()
複製代碼
  1. 建表語句：
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 後話

網上相似資料甚少，啃各類文檔，探尋技術實現方式

喜歡請點個贊，如有疑問能夠評論，很是樂意幫助解決問題

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程