Egg.js: фреймворк корпоративного уровня на основе KOA2.
Kafka: высокопроизводительная распределенная система обмена сообщениями с публикацией и подпиской.
В этой статье будет интегрирован пример лог-системы egg + kafka + mysql.
Системные требования: логирование, управление очередью сообщений через kafka
Карта идей:
λ.1 Подготовка окружающей среды
① Кафка
Скачав kafka с официального сайта, распакуйте его
Запуск зоопарка:
bin/zookeeper-server-start.sh config/zookeeper.properties
Запустить сервер Кафка
Здесь в config/server.properties будет num.partitions=5, разделов ставим 5
bin/kafka-server-start.sh config/server.properties
② яйцо + mysql
Соберите яйцо согласно лесам, а затем установите больше kafka-node, egg-mysql
имя пользователя mysql root пароль 123456
λ.2 Интеграция
- Создайте новый 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);
});
});
};
Приведенный выше код создает новые производители и потребителей.
После создания производителя он загружается в глобальный объект приложения. Мы будем производить сообщения по запросу. Вот просто новый экземпляр
Когда потребители получают сообщения, они получают доступ к методу вставки сервисного уровня (вставка данных из базы данных).
Конкретные параметры см. в официальном API kafka-node, и там будут параметры конфигурации для производителей и потребителей.
- controller · log.js
Производитель получается здесь и передается на сервисный уровень.
'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;
- service · log.js
Здесь есть метод отправки, который вызывает производителя. отправить для производства производителя
Метод вставки данных вставки - это база данных
'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;
- config · config.default.js
Некоторые из параметров конфигурации, используемые в вышеуказанном коде, здесь подробно описаны, обратите внимание, что здесь открываются 5 разделов.
'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;
};
- Класс сущности:
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;
};
- Тестовый скрипт 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()
- Оператор создания таблицы:
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 Позже
В интернете очень мало подобной информации, грызя разные документы, и изучая технические способы реализации
Если вам это нравится, пожалуйста, дайте ему палец вверх.Если у вас есть какие-либо вопросы, вы можете оставить комментарий.Я очень рад помочь решить проблему.