Skip to content

Z
zhichan
Commit c3318ec5 by 宋毅
Options

tj

parent 48c50875
Showing with 172 additions and 51 deletions
brg-queue-center/app/base/api/api.base.js
const system = require("../system"); const system = require("../system");
const moment = require('moment'); const moment = require('moment');
const settings = require("../../../app/config/settings") const settings = require("../../../app/config/settings");
const axios = require('axios'); const axios = require('axios');
class APIBase { class APIBase {
...@@ -8,6 +8,7 @@ class APIBase { ...@@ -8,6 +8,7 @@ class APIBase {
this.redisClient = system.getObject("util.redisClient"); this.redisClient = system.getObject("util.redisClient");
this.execClient = system.getObject("util.execClient"); this.execClient = system.getObject("util.execClient");
this.esUtils = system.getObject("util.esUtils"); this.esUtils = system.getObject("util.esUtils");
this.exTime = 6 * 3600;//缓存过期时间，6小时
} }
//-----------------------新的模式------------------开始 //-----------------------新的模式------------------开始
...@@ -17,17 +18,17 @@ class APIBase { ...@@ -17,17 +18,17 @@ class APIBase {
if ("LOGS-SYTXPUBLIC-MSGQ" != settings.queuedName) { if ("LOGS-SYTXPUBLIC-MSGQ" != settings.queuedName) {
pobj.actionBody.resultInfo = result; pobj.actionBody.resultInfo = result;
pobj.actionBody.requestId = result.requestId; pobj.actionBody.requestId = result.requestId;
pobj.actionBody.opTitle = "执行" + methodname + "方法"; pobj.actionBody.opTitle = "reqPath:" + req.path;
this.esUtils.addEsLogs(settings.queuedName + "api-doexecmethod-info", pobj.actionBody); this.esUtils.addEsLogs(settings.queuedName + "api-req", pobj.actionBody);
return result;
} }
return result;
} catch (e) { } catch (e) {
console.log(e.stack, "api.base调用出现异常，请联系管理员.........."); console.log(e.stack, "api.base调用出现异常，请联系管理员..........");
var rtnerror = system.getResultFail(-200, "出现异常，error：" + e.stack); var rtnerror = system.getResultFail(-200, "出现异常，error：" + e.stack);
rtnerror.requestId = await this.getBusUid("err"); rtnerror.requestId = await this.getBusUid("err");
pobj.actionBody.requestId = rtnerror.requestId; pobj.actionBody.requestId = rtnerror.requestId;
pobj.actionBody.errorInfo = e.stack; pobj.actionBody.errorInfo = e.stack;
pobj.actionBody.opTitle = "执行" + methodname + "方法"; pobj.actionBody.opTitle = "，reqPath:" + req.path;
this.esUtils.addEsLogs(settings.queuedName + "apidoexec-error", pobj.actionBody); this.esUtils.addEsLogs(settings.queuedName + "apidoexec-error", pobj.actionBody);
return rtnerror; return rtnerror;
} }
......
brg-queue-center/app/base/db/consumer.base.js
...@@ -35,11 +35,11 @@ class ConsumerBase { ...@@ -35,11 +35,11 @@ class ConsumerBase {
}); });
} catch (error) { } catch (error) {
this.errorLogDao.addOpErrorLogs(queuedName + "队列执行doConsumer存在异常", null, null, error.stack, 3); this.errorLogDao.addOpErrorLogs(queuedName + "队列执行doConsumer存在异常", null, null, error.stack, 3);
//日志记录 //日志
console.log(JSON.stringify({ console.log(JSON.stringify({
optitle: this.serviceName + "，队列执行doConsumer存在异常", optitle: this.serviceName + "，队列执行doConsumer存在异常",
op: "base/db/consumer.base.js", op: "base/db/consumer.base.js",
message: error.stack message: ""
})); }));
} }
} }
...@@ -81,7 +81,7 @@ class ConsumerBase { ...@@ -81,7 +81,7 @@ class ConsumerBase {
console.log(JSON.stringify({ console.log(JSON.stringify({
optitle: this.serviceName + "，队列执行execSubDoConsumer存在异常", optitle: this.serviceName + "，队列执行execSubDoConsumer存在异常",
op: "base/db/consumer.base.js", op: "base/db/consumer.base.js",
message: "" message: error.stack
})); }));
} }
} }
......
brg-queue-center/app/base/db/impl/common/connection.js
...@@ -18,10 +18,13 @@ class DbFactory{ ...@@ -18,10 +18,13 @@ class DbFactory{
async initModels(){ async initModels(){
var self=this; var self=this;
var modelpath=path.normalize(path.join(__dirname, '../..'))+"/models/"; var modelpath=path.normalize(path.join(__dirname, '../..'))+"/models/";
console.log(modelpath);
var models=glob.sync(modelpath+"/**/*.js"); var models=glob.sync(modelpath+"/**/*.js");
console.log(models.length);
models.forEach(function(m){ models.forEach(function(m){
self.db.import(m); self.db.import(m);
}); });
console.log("init models....");
} }
async initRelations(){ async initRelations(){
...@@ -53,4 +56,49 @@ class DbFactory{ ...@@ -53,4 +56,49 @@ class DbFactory{
return this.db; return this.db;
} }
} }
module.exports=DbFactory; module.exports=DbFactory;
\ No newline at end of file // const dbf=new DbFactory();
// dbf.getCon().then((db)=>{
// //console.log(db);
// // db.models.user.create({nickName:"jy","description":"cccc",openId:"xxyy",unionId:"zz"})
// // .then(function(user){
// // var acc=db.models.account.build({unionId:"zz",nickName:"jy"});
// // acc.save().then(a=>{
// // user.setAccount(a);
// // });
// // console.log(user);
// // });
// // db.models.user.findAll().then(function(rs){
// // console.log("xxxxyyyyyyyyyyyyyyyyy");
// // console.log(rs);
// // })
// });
// const User = db.define('user', {
// firstName: {
// type: Sequelize.STRING
// },
// lastName: {
// type: Sequelize.STRING
// }
// });
// db
// .authenticate()
// .then(() => {
// console.log('Co+nnection has been established successfully.');
//
// User.sync(/*{force: true}*/).then(() => {
// // Table created
// return User.create({
// firstName: 'John',
// lastName: 'Hancock'
// });
// });
//
// })
// .catch(err => {
// console.error('Unable to connect to the database:', err);
// });
//
// User.findAll().then((rows)=>{
// console.log(rows[0].firstName);
// });
brg-queue-center/app/base/db/impl/opLogs/errorLogDao.js
const system = require("../../../system"); const system = require("../../../system");
const Dao = require("../../dao.base"); const Dao = require("../../dao.base");
const settings = require("../../../../config/settings");
class ErrorLogDao extends Dao { class ErrorLogDao extends Dao {
constructor() { constructor() {
super(Dao.getModelName(ErrorLogDao)); super(Dao.getModelName(ErrorLogDao));
...@@ -12,9 +14,10 @@ class ErrorLogDao extends Dao { ...@@ -12,9 +14,10 @@ class ErrorLogDao extends Dao {
* @param {*} error 错误信息 * @param {*} error 错误信息
* @param {*} logLevel 日志级别信息,debug: 0, info: 1, warn: 2, error: 3, fatal: 4 * @param {*} logLevel 日志级别信息,debug: 0, info: 1, warn: 2, error: 3, fatal: 4
*/ */
async addOpErrorLogs(opTitle, actionBody, execResult, error,logLevel) { async addOpErrorLogs(opTitle, actionBody, execResult, error, logLevel) {
error = typeof error === 'object' ? error : { error_info: error }; error = typeof error === 'object' ? error : { error_info: error };
var params = { var params = {
queued_name: settings.queuedName,
identify_code: actionBody.identifyCode, identify_code: actionBody.identifyCode,
op_title: opTitle, op_title: opTitle,
push_content: actionBody || null, push_content: actionBody || null,
......
brg-queue-center/app/base/db/models/opLogs/errorLog.js
const system = require("../../../system"); const system = require("../../../system");
module.exports = (db, DataTypes) => { module.exports = (db, DataTypes) => {
return db.define("errorLog", { return db.define("errorLog", {
queued_name: DataTypes.STRING(512), //队列名称
identify_code: DataTypes.STRING(100), //标识code identify_code: DataTypes.STRING(100), //标识code
op_title: DataTypes.STRING(100), // 操作标题 op_title: DataTypes.STRING(100), // 操作标题
push_content: DataTypes.JSON, //推送的内容 push_content: DataTypes.JSON, //推送的内容
......
brg-queue-center/app/base/service/app.base.js
const system = require("../system"); const system = require("../system");
const moment = require('moment') const moment = require('moment');
const axios = require('axios');
const settings = require("../../config/settings"); const settings = require("../../config/settings");
class AppServiceBase { class AppServiceBase {
...@@ -8,13 +9,44 @@ class AppServiceBase { ...@@ -8,13 +9,44 @@ class AppServiceBase {
this.redisClient = system.getObject("util.redisClient"); this.redisClient = system.getObject("util.redisClient");
} }
/** /**
* 带超时时间的post请求 * 带超时时间的post请求
* @param {*} params 请求数据-json格式 * @param {*} params 请求数据-json格式
* @param {*} url 请求地址 * @param {*} url 请求地址
* @param {*} ContentType 请求头类型，默认application/json * @param {*} ContentType 请求头类型，默认application/json
* @param {*} headData 请求头内容-json格式，如：请求头中传递token,格式：{token:"9098902q849q0434q09439"} * @param {*} headData 请求头内容-json格式，如：请求头中传递token,格式：{token:"9098902q849q0434q09439"}
*/ */
async execPostByTimeOut(params, url, ContentType, headData, timeOut = 60) { async execPostByTimeOut(params, url, ContentType, headData, timeOut = 60) {
if (settings.env === "dev") {
var headers = {
'Content-type': 'application/json'
}
if (headData) {
var headDataKeys = Object.keys(headData);
if (headDataKeys.length > 0) {
for (let index = 0; index < headDataKeys.length; index++) {
const indexKey = headDataKeys[index];
var headValue = headData[indexKey];
if (indexKey && headValue) {
headers[indexKey] = headValue;
}
}
}
}
let result = await axios({
// headers: {'Content-Type': 'application/x-www-form-urlencoded'},
headers: headers,
method: 'POST',
url: url,
data: JSON.stringify(params),
timeout: timeOut
});
if (result.status == 200 && result.data) {
return result.data;
}
this.errorLogDao.addOpErrorLogs(queuedName + "执行execPostEs存在错误", params, result, null, 3);
return system.getResult(null, "执行execPostEs存在错误");
}
//方式二
var rtn = await this.execClient.execPostTimeOut(params, url, ContentType, headData, timeOut); var rtn = await this.execClient.execPostTimeOut(params, url, ContentType, headData, timeOut);
if (!rtn || !rtn.stdout) { if (!rtn || !rtn.stdout) {
return system.getResult(null, "execPostTimeOut data is empty"); return system.getResult(null, "execPostTimeOut data is empty");
...@@ -41,7 +73,7 @@ class AppServiceBase { ...@@ -41,7 +73,7 @@ class AppServiceBase {
返回20位业务订单号 返回20位业务订单号
prefix：业务前缀 prefix：业务前缀
*/ */
async getBusUid(prefix) { async getBusUid(prefix) {
prefix = (prefix || ""); prefix = (prefix || "");
if (prefix) { if (prefix) {
prefix = prefix.toUpperCase(); prefix = prefix.toUpperCase();
...@@ -59,7 +91,7 @@ class AppServiceBase { ...@@ -59,7 +91,7 @@ class AppServiceBase {
len：返回长度 len：返回长度
radix：参与计算的长度，最大为62 radix：参与计算的长度，最大为62
*/ */
async getUidInfo(len, radix) { async getUidInfo(len, radix) {
var chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'.split('');//长度62，到yz长度为长36 var chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'.split('');//长度62，到yz长度为长36
var uuid = [], i; var uuid = [], i;
radix = radix || chars.length; radix = radix || chars.length;
......
brg-queue-center/app/base/service/sve.base.js
const system = require("../system"); const system = require("../system");
const axios = require('axios');
const moment = require('moment') const moment = require('moment')
// const settings = require("../../config/settings"); // const settings = require("../../config/settings");
class ServiceBase { class ServiceBase {
...@@ -15,13 +16,44 @@ class ServiceBase { ...@@ -15,13 +16,44 @@ class ServiceBase {
return resultStr; return resultStr;
} }
/** /**
* 带超时时间的post请求 * 带超时时间的post请求
* @param {*} params 请求数据-json格式 * @param {*} params 请求数据-json格式
* @param {*} url 请求地址 * @param {*} url 请求地址
* @param {*} ContentType 请求头类型，默认application/json * @param {*} ContentType 请求头类型，默认application/json
* @param {*} headData 请求头内容-json格式，如：请求头中传递token,格式：{token:"9098902q849q0434q09439"} * @param {*} headData 请求头内容-json格式，如：请求头中传递token,格式：{token:"9098902q849q0434q09439"}
*/ */
async execPostByTimeOut(params, url, ContentType, headData, timeOut = 60) { async execPostByTimeOut(params, url, ContentType, headData, timeOut = 60) {
if (settings.env === "dev") {
var headers = {
'Content-type': 'application/json'
}
if (headData) {
var headDataKeys = Object.keys(headData);
if (headDataKeys.length > 0) {
for (let index = 0; index < headDataKeys.length; index++) {
const indexKey = headDataKeys[index];
var headValue = headData[indexKey];
if (indexKey && headValue) {
headers[indexKey] = headValue;
}
}
}
}
let result = await axios({
// headers: {'Content-Type': 'application/x-www-form-urlencoded'},
headers: headers,
method: 'POST',
url: url,
data: JSON.stringify(params),
timeout: timeOut
});
if (result.status == 200 && result.data) {
return result.data;
}
this.errorLogDao.addOpErrorLogs(queuedName + "执行execPostEs存在错误", params, result, null, 3);
return system.getResult(null, "执行execPostEs存在错误");
}
//方式二
var rtn = await this.execClient.execPostTimeOut(params, url, ContentType, headData, timeOut); var rtn = await this.execClient.execPostTimeOut(params, url, ContentType, headData, timeOut);
if (!rtn || !rtn.stdout) { if (!rtn || !rtn.stdout) {
return system.getResult(null, "execPostTimeOut data is empty"); return system.getResult(null, "execPostTimeOut data is empty");
...@@ -39,7 +71,7 @@ class ServiceBase { ...@@ -39,7 +71,7 @@ class ServiceBase {
返回20位业务订单号 返回20位业务订单号
prefix：业务前缀 prefix：业务前缀
*/ */
async getBusUid(prefix) { async getBusUid(prefix) {
prefix = (prefix || ""); prefix = (prefix || "");
if (prefix) { if (prefix) {
prefix = prefix.toUpperCase(); prefix = prefix.toUpperCase();
...@@ -57,7 +89,7 @@ class ServiceBase { ...@@ -57,7 +89,7 @@ class ServiceBase {
len：返回长度 len：返回长度
radix：参与计算的长度，最大为62 radix：参与计算的长度，最大为62
*/ */
async getUidInfo(len, radix) { async getUidInfo(len, radix) {
var chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'.split('');//长度62，到yz长度为长36 var chars = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'.split('');//长度62，到yz长度为长36
var uuid = [], i; var uuid = [], i;
radix = radix || chars.length; radix = radix || chars.length;
......
brg-queue-center/app/base/utils/esUtils.js
...@@ -33,7 +33,7 @@ class EsUtils { ...@@ -33,7 +33,7 @@ class EsUtils {
actionBody.resultInfo = typeof actionBody.resultInfo === 'object' ? JSON.stringify(actionBody.resultInfo) : actionBody.resultInfo; actionBody.resultInfo = typeof actionBody.resultInfo === 'object' ? JSON.stringify(actionBody.resultInfo) : actionBody.resultInfo;
actionBody.errorInfo = typeof actionBody.errorInfo === 'object' ? JSON.stringify(actionBody.errorInfo) : actionBody.errorInfo; actionBody.errorInfo = typeof actionBody.errorInfo === 'object' ? JSON.stringify(actionBody.errorInfo) : actionBody.errorInfo;
var params = { var params = {
opTitle: actionBody.opTitle || "", opTitle: moment().format("YYYY-MM-DD HH:mm:ss:SSS") + "，" + actionBody.opTitle || "",
identifyCode: actionBody.identifyCode || "", identifyCode: actionBody.identifyCode || "",
messageBody: actionBody.messageBody || "", messageBody: actionBody.messageBody || "",
resultInfo: actionBody.resultInfo || "", resultInfo: actionBody.resultInfo || "",
...@@ -51,7 +51,6 @@ class EsUtils { ...@@ -51,7 +51,6 @@ class EsUtils {
*/ */
async execPostEs(queuedName, params, reqUrl, esName, esPwd) { async execPostEs(queuedName, params, reqUrl, esName, esPwd) {
try { try {
params.opTitle = moment().format("YYYY-MM-DD HH:mm:ss:SSS") + params.opTitle || "";
if (settings.env === "dev") { if (settings.env === "dev") {
let result = await axios({ let result = await axios({
// headers: {'Content-Type': 'application/x-www-form-urlencoded'}, // headers: {'Content-Type': 'application/x-www-form-urlencoded'},
...@@ -77,9 +76,14 @@ class EsUtils { ...@@ -77,9 +76,14 @@ class EsUtils {
return system.getResult(null, "执行execPostEs存在错误"); return system.getResult(null, "执行execPostEs存在错误");
return system.getResult(null, "execPostTimeOut data is empty"); return system.getResult(null, "execPostTimeOut data is empty");
} }
var stdoutInfo=JSON.parse(result.stdout);
if(stdoutInfo.error){
this.errorLogDao.addOpErrorLogs(queuedName + "执行execPostEs存在错误", params, result, null, 3);
return system.getResult(null, "执行execPostEs存在错误");
}
return system.getResultSuccess(); return system.getResultSuccess();
} catch (error) { } catch (error) {
console.log(error.stack, "......execPostEs......error..."); console.log(error.stack, "......execPostEs....error.....");
this.errorLogDao.addOpErrorLogs(queuedName + "执行execPostEs存在异常", params, null, error.stack, 3); this.errorLogDao.addOpErrorLogs(queuedName + "执行execPostEs存在异常", params, null, error.stack, 3);
} }
} }
......
brg-queue-center/app/base/utils/execClient.js
...@@ -15,6 +15,7 @@ class ExecClient { ...@@ -15,6 +15,7 @@ class ExecClient {
*/ */
async execGet(params, url, headData) { async execGet(params, url, headData) {
let cmd = this.FetchGetCmd(params, url, headData); let cmd = this.FetchGetCmd(params, url, headData);
console.log(cmd);
var result = await this.exec(cmd); var result = await this.exec(cmd);
return result; return result;
} }
...@@ -90,6 +91,7 @@ class ExecClient { ...@@ -90,6 +91,7 @@ class ExecClient {
FetchGetCmd(params, url, headData) { FetchGetCmd(params, url, headData) {
var cmd = this.cmdGetPattern.replace( var cmd = this.cmdGetPattern.replace(
/\{data\}/g, params).replace(/\{url\}/g, url); /\{data\}/g, params).replace(/\{url\}/g, url);
console.log(cmd);
return cmd; return cmd;
} }
FetchPostCmd(params, url, ContentType, headData) { FetchPostCmd(params, url, ContentType, headData) {
......
brg-queue-center/app/base/utils/redisClient.js
...@@ -10,14 +10,8 @@ class RedisClient { ...@@ -10,14 +10,8 @@ class RedisClient {
this.subclient = this.client.duplicate(); this.subclient = this.client.duplicate();
this.client.on("error", function (err) { this.client.on("error", function (err) {
//TODO：日志处理
console.log(err, "..redisClient........error"); console.log(err, "..redisClient........error");
//TODO:日志记录
// logCtl.error({
// optitle:"redis this.client.on异常：",
// op:"base/utils/redisClient/this.client.on",
// content:err,
// clientIp:""
// });
}); });
const self = this; const self = this;
// 监听回调 // 监听回调
...@@ -90,11 +84,9 @@ class RedisClient { ...@@ -90,11 +84,9 @@ class RedisClient {
messageBody = typeof messageBody === 'object' ? JSON.stringify(messageBody) : messageBody; messageBody = typeof messageBody === 'object' ? JSON.stringify(messageBody) : messageBody;
this.client.lpush(queuedName, messageBody, function (err, reply) { this.client.lpush(queuedName, messageBody, function (err, reply) {
if (err) { if (err) {
//TODO:日志记录 return new Error("lpush message error :" + err + "，queuedName：" + queuedName + "，messageBody：" + messageBody);
console.log('lpush message error :' + err + "，queuedName：" + queuedName + "，messageBody：" + messageBody);
return new Error('lpush message error :' + err);
} else { } else {
console.log('lpush message success'); console.log("lpush message success");
} }
}); });
} }
...@@ -120,10 +112,9 @@ class RedisClient { ...@@ -120,10 +112,9 @@ class RedisClient {
param = typeof param === 'object' ? JSON.stringify(param) : param; param = typeof param === 'object' ? JSON.stringify(param) : param;
await this.client.zadd([collectionName + "-SORT", score, param], function (err, data) { await this.client.zadd([collectionName + "-SORT", score, param], function (err, data) {
if (err) { if (err) {
// TDO:日志记录 return new Error("zaddSortedSet data error :" + err + "，collectionName：" + collectionName + "，param：" + param);
return new Error('zadd data error :' + err);
} else { } else {
console.log(data, 'zadd data success'); console.log("zaddSortedSet data success");
} }
}); });
} }
......
brg-queue-center/app/config/environment.js
...@@ -66,8 +66,16 @@ module.exports = function (app) { ...@@ -66,8 +66,16 @@ module.exports = function (app) {
app.use(errorHandler()); app.use(errorHandler());
} else { } else {
app.use(function (err, req, res) { app.use(function (err, req, res) {
//TODO:日志记录 console.log("prod error handler...........................................");
console.log(err, "...environment........................error................"); console.log(err);
logCtl.error({
optitle: "environment 调用异常error:",
op: req.url,
content: e.toString(),
clientIp: system.get_client_ip(req),
agent: req.headers["user-agent"],
});
//logerApp.error("prod error handler",err);
res.send("link index"); res.send("link index");
}); });
} }
......
brg-queue-center/app/config/settings.js
...@@ -18,10 +18,9 @@ var settings = { ...@@ -18,10 +18,9 @@ var settings = {
consumerName: ENVINPUT.CONSUMER_NAME, consumerName: ENVINPUT.CONSUMER_NAME,
queuedName: ENVINPUT.QUEUED_NAME, queuedName: ENVINPUT.QUEUED_NAME,
basepath: path.normalize(path.join(__dirname, '../..')), basepath: path.normalize(path.join(__dirname, '../..')),
port: process.env.NODE_PORT || 8080, port: process.env.NODE_PORT || 8086,
redis: function () { redis: function () {
if (this.env == "dev") { if (this.env == "dev" || this.env == "test") {
console.log("dev.........................................................");
var localsettings = require("./localsettings"); var localsettings = require("./localsettings");
return localsettings.redis; return localsettings.redis;
} else { } else {
...@@ -34,7 +33,7 @@ var settings = { ...@@ -34,7 +33,7 @@ var settings = {
} }
}, },
database: function () { database: function () {
if (this.env == "dev") { if (this.env == "dev" || this.env == "test") {
var localsettings = require("./localsettings"); var localsettings = require("./localsettings");
return localsettings.database; return localsettings.database;
} else { } else {
......
brg-queue-center/main.js
...@@ -9,6 +9,7 @@ const app = express(); ...@@ -9,6 +9,7 @@ const app = express();
// var dbf = system.getObject("db.common.connection"); // var dbf = system.getObject("db.common.connection");
// con = dbf.getCon(); // con = dbf.getCon();
console.log(settings.consumerName, "--consumerName-----start-----");
if (settings.consumerName) { if (settings.consumerName) {
var consumer = system.getObject("consumer." + settings.consumerName); var consumer = system.getObject("consumer." + settings.consumerName);
(async () => { (async () => {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment