node.js的mysql模块本身没有提供返回promise的函数，即是说都是用的回调函数，那么对于我们使用async函数是很不方便的一件事。node.js有一个mysql封装库叫mysql-promise，这个库提供使用函数拼凑sql语句，但我比较想用原生的sql语句，原因在于mysql对于query函数封装得比较完美，能对应select,delete,update,insert返回不同的结果，比如update和delete返回affectRows，select返回查询结果这样，再加上使用参数化的sql语句能防止sql注入，所以封装了一下mysql的npm包。
首先是package.json，对于Promise，建议使用bluebird.Promise，然后自然是mysql
{
"name": "async-mysql",
"version": "1.0.0",
"main": "index.js",
"author": {
"name": "kazetotori/fxqn",
"email": "kakkouto98045@live.com"
},
"files": [
"lib",
"index.js"
],
"dependencies": {
"mysql": "2.12.0",
"bluebird": "3.4.6"
}
}
这个库入口文件为index.js，这里仅仅作为一个导出文件使用，没有任何代码
module.exports.Connection = require('./lib/Connection').Connection;
module.exports.Pool = require('./lib/Pool').Pool;
首先来实现一下单个连接的各个函数，使用es6的class关键字和Symbol封装，比传统的构造函数更加直观
const mysql = require('mysql');
const bluebird = require('bluebird');
const Promise = bluebird.Promise;
var $originConn = Symbol('originConn');
var $isPoolConn = Symbol('isPoolConn');
var $isAlive = Symbol('isAlive');
/**
* This function is the factory of the standard promise callback.
* @param {Function} resolve
* @param {Function} reject
* @return {Function} The standard promise callback.
*/
function promiseFn(resolve, reject) {
return (err, rst) => {
if (err) reject(err);
else resolve(rst);
}
}
/**
* Connection is the class that contains functions that each returns promise.
* These functions are just converted from a Mysql.Connection Object.
*/
class Connection {
/**
* Constructor, initialize the connection object.
* @param {Object} config The configuration of the connection.
* @param {Boolean} isPoolConn Orders the connection is in a pool or not.
*/
constructor(config, isPoolConn = false) {
if (config.query)
this[$originConn] = config;
else
this[$originConn] = mysql.createConnection(config);
this[$isPoolConn] = isPoolConn;
this[$isAlive] = true;
}
/**
* Connection config
*/
get config() { return this[$originConn].config; }
/**
* Orders the connection is in a pool or not.
*/
get isPoolConnection() { return this[$isPoolConn]; }
/**
* Orders the connection is destroyed or not.
*/
get isAlive() { return this[$isAlive]; }
/**
* Orders the threadId of the connection.
*/
get threadId() { return this[$originConn].threadId; }
/**
* Add listener of this connection.
*/
get on() { return this[$originConn].on; };
/**
* Ternimate the connection immediately whether there's any query in quene or not.
*/
destroy() {
return new Promise((resolve, reject) => {
this[$originConn].destroy();
this[$isAlive] = false;
resolve();
});
}
/**
* Ternimate the connection. This function will ternimate the connection after any query being complete.
*/
end() {
return new Promise((resolve, reject) => {
this[$originConn].end(promiseFn(resolve, reject))
})
.then(() => {
this[$isAlive] = false;
})
}
/**
* Execute sql command with parameters.
* @param {String} cmd The sql command would be executed.
* @param {Array} params Parameters.
* @return {Promise<any>} The sql result.
*/
query(cmd, params) {
return new Promise((resolve, reject) => {
let conn = this[$originConn];
let args = [cmd];
let callback = promiseFn(resolve, reject);
if (params)
args.push(params);
args.push(callback);
conn.query(...args);
});
}
/**
* Begin transaction of the connection. Following queries would not be useful until the function commit or rollback called.
* @return {Promise<undefined>}
*/
beginTran() {
return new Promise((resolve, reject) => {
let conn = this[$originConn];
conn.beginTransaction(promiseFn(resolve, reject));
});
}
/**
* Commit a transaction.
* @return {Promise<undefined>}
*/
commit() {
return new Promise((resolve, reject) => {
let conn = this[$originConn];
conn.commit((err) => {
if (err) this.rollback().then(() => reject(err));
else resolve();
})
});
}
/**
* Rollback a transaction
* @return {Promise<undefined>}
*/
rollback() {
return new Promise((resolve, reject) => {
let conn = this[$originConn];
conn.rollback(() => resolve());
});
}
}
/**
* PoolConnection is the class extending from Connection.
* Any object of this class is the connection in a connection pool.
*/
class PoolConnection extends Connection {
constructor(originConn) {
super(originConn, true);
}
/**
* Release the connection and put it back to the pool.
* @return {Promise<undefined>}
*/
release() {
return new Promise((resolve, reject) => {
this[$originConn].release();
resolve();
});
}
}
module.exports.Connection = Connection;
module.exports.PoolConnection = PoolConnection;
然后是连接池的部分
const Promise = require('bluebird').Promise;
const mysql = require('mysql');
const PoolConnection = require('./Connection').PoolConnection;
var $originPool = Symbol('originPool');
var $isAlive = Symbol('isAlive');
/**
* Pool is the class that contains functions each returns promise.
* These functions are just converted from the Mysql.Pool object.
*/
class Pool {
/**
* Constructor, initialize the pool.
* @param {Object} config The pool config.
*/
constructor(config) {
this[$originPool] = mysql.createPool(config);
this[$isAlive] = true;
}
/**
* Orders the pool config.
*/
get config() { return this[$originPool].config; }
/**
* Orders the pool is destroyed or not.
*/
get isAlive() { return this[$isAlive]; }
/**
* Add listener to the pool.
*/
get on() { return this[$originPool].on; }
/**
* Get a connection object from the pool.
* @return {Promise<PoolConnection>}
*/
getConn() {
return new Promise((resolve, reject) => {
this[$originPool].getConnection((err, originConn) => {
if (err)
return reject(err);
let conn = new PoolConnection(originConn);
resolve(conn);
});
});
}
/**
* Ternimate the pool. This function would ternimate the pool after any query being complete.
*/
end() {
return new Promise((resolve, reject) => {
this[$originPool].end((err) => {
if (err)
return reject(err);
this[$isAlive] = false;
resolve();
})
});
}
/**
* Use a connection to query a sql command with parameters.
* @param {String} cmd The sql command would be executed.
* @param {Array} params Parameters.
* @return {Promise<any>}
*/
query(cmd, params) {
return new Promise((resolve, reject) => {
let args = [cmd];
let callback = (err, rst) => {
if (err) reject(err);
else resolve(rst);
}
if (params)
args.push(params);
args.push(callback);
this[$originPool].query(...args);
});
}
}
module.exports.Pool = Pool;
最后加一个config，便于智能提示
var $host = Symbol('host');
var $port = Symbol('port');
var $localAddr = Symbol('localAddr');
var $socketPath = Symbol('socketPath');
var $user = Symbol('user');
var $pwd = Symbol('pwd');
var $db = Symbol('db');
var $charset = Symbol('charset');
var $timezone = Symbol('timezone');
var $connTimeout = Symbol('connTimeout');
var $stringifyObjs = Symbol('stringifyObjs');
var $typeCast = Symbol('typeCast');
var $queryFormat = Symbol('queryFormat');
var $supportBigNumbers = Symbol('supportBigNumbers');
var $bigNumberStrings = Symbol('bigNumberStrings');
var $dateStrings = Symbol('dateStrings');
var $debug = Symbol('debug');
var $trace = Symbol('trace');
var $multiStmts = Symbol('multipleStatements');
var $flags = Symbol('flags');
var $ssl = Symbol('ssl');
class MysqlConfig {
constructor(config) {
for (let k in config)
this[k] = config[k];
}
get host() { return this[$host] }
set host(val) { this[$host] = val }
get port() { return this[$port] }
set port(val) { this[$port] = val }
get localAddress() { return this[$localAddr] }
set localAddress(val) { this[$localAddr] = val }
get socketPath() { return this[$socketPath] }
set socketPath(val) { this[$socketPath] = val }
get user() { return this[$user] }
set user(val) { this[$user] = val }
get password() { return this[$pwd] }
set password(val) { this[$pwd] = val }
get database() { return this[$db] }
set database(val) { this[$db] = val }
get charset() { return this[$charset] }
set charset(val) { this[$charset] = val }
get timezone() { return this[$timezone] }
set timezone(val) { this[$timezone] = val }
get connectTimeout() { return this[$connTimeout] }
set connectTimeout(val) { this[$connTimeout] = val }
get stringifyObjects() { return this[$stringifyObjs] }
set stringifyObjects(val) { this[$stringifyObjs] = val }
get typeCast() { return this[$typeCast] }
set typeCast() { this[$typeCast] = val }
get queryFormat() { return this[$queryFormat] }
set queryFormat(val) { this[$queryFormat] = val }
get supportBigNumbers() { return this[$supportBigNumbers] }
set supportBigNumbers(val) { this[$supportBigNumbers] = val }
get bigNumberStrings() { return this[$bigNumberStrings] }
set bigNumberStrings(val) { this[$bigNumberStrings] = val }
get dateStrings() { return this[$dateStrings] }
set dateStrings(val) { this[$dateStrings] = val }
get debug() { return this[$debug] }
set debug(val) { this[$debug] = val }
get trace() { return this[$trace] }
set trace(val) { this[$trace] = val }
get multipleStatements() { return this[$multiStmts] }
set multipleStatements(val) { this[$multiStmts] = val }
get flags() { return this[$flags] }
set flags(val) { this[$flags] = val }
get ssl() { return this[$ssl] }
set ssl(val) { this[$ssl] = val }
}
module.exports.MysqlConfig = MysqlConfig;
测试代码
//Use this test.js need node version is higher than 7.0.0 .
//And need the node arg "--harmony". const config = {
"host": "localhost",
"port": 3306,
"user": "root",
"database": "testDB",
"charset": "UTF8_GENERAL_CI",
"timezone": "local",
"connectTimeout": 10000,
"connectionLimit": 10
};
const Pool = require('./lib/Pool').Pool;
const Connection = require('./lib/Connection').Connection;
var pool = new Pool(config);
var conn = new Connection(config); async function poolTest() { //pool.query()
let result = await pool.query('SELECT * FROM tbltest WHERE name=?', ['wr']);
console.log(result); //pool.getConn();
let poolConn = await pool.getConn();
console.log(poolConn.isPoolConnection);
result = await poolConn.query('SELECT * FROM tbltest WHERE name=?', ['zs']);
console.log(result); await pool.end();
console.log(pool.isAlive);
} async function connTest() {
let rst = await conn.query('SELECT * FROM tbltest WHERE name=?', ['ls']);
console.log(rst);
await conn.beginTran();
let count = (await conn.query('SELECT COUNT(*) FROM tbltest WHERE name=?', ['??']))[0]['COUNT(*)'];
console.log(count);
await conn.query('INSERT INTO tbltest(name) VALUES(?)', ['zhangsan']);
if (count > 0) {
await conn.commit();
console.log('commit');
}
else {
await conn.rollback();
console.log('rollback');
} rst = await conn.query('SELECT * FROM tbltest');
console.log(rst);
} poolTest();
connTest();