MSSQL 是Nodejs用於連接Microsoft SQL Server 的插件。
在nodejs中，mssql模塊支持sqlserver數據庫操作。今天將mssql模塊的某些功能封裝為一個類，方便以后調用。封裝的功能有執行存儲過程，執行查詢語句操作等。如果本篇文章對大家有幫助，那就再好不過了！
如果需要擴展事務和其他功能，請到官網https://www.npmjs.com/package/mssql中看文檔，里面有示例。
安裝方法
安裝指令npm install mssql@4.2.1
配置Config
const config = {
user: '...',
password: '...',
server: 'localhost',
database: '...',
options: {
encrypt: true //使用windows azure，需要設置次配置。
}
}
user：SQL Server 的登錄名
password: SQL Server的登錄密碼
server:SQL Server的地址
port：端口號，默認為1433
domain：設置domain后，可通過domain連接數據庫
database：數據庫名稱
connectionTimeout：連接timeout，單位ms 默認 15000
requestTimeout：請求timeout，單位ms默認15000
parseJSON：將json數據集轉化成json obj
pool.max：連接池最大連接數，默認10
pool.min：連接池最小連接數，默認0
pool.idleTimeoutMillis：設置關閉未使用連接的時間，單位ms默認30000
封裝的代碼如下
1.連接池的寫法：
//導入mssql模塊 //基於版本@4.2.1 安裝指令npm install mssql@4.2.1
// pool.close()非常重要，只創建，不關閉會造成非常嚴重的內存泄漏。 關閉池中的所有活動連接。
let mssql=require("mssql");
//引用配置參數模塊
let configFile = require("./config");
//數據庫異常
mssql.on('error', err => {
console.log("mssql異常原因："+err.message);
});
let sql={};
//sql參數的類型
sql.direction={
//輸入參數
Input:"input",
//輸出參數
Output:"output",
//返回參數
Return:"return"
};
//配置存儲過程是的輸出輸入
sql.sqlserver=mssql;
//默認config對象
let config=configFile.mssql_config;
/**
* 初始化連接參數
* @param {string} user 用戶名
* @param {string} password 密碼
* @param {string} server 服務器地址
* @param {string} database 數據庫名稱
* @param {string} port 數據庫端口
*/
sql.initConfig = function(user,password,server,database,port){
config.user = user;
config.password =password;
config.server =server;
config.database= database;
config.port = port;
}
/**
* 執行存儲過程
* @param {string} procedure 存儲過程名稱
* @param {JSON} params 存儲過程參數
* params的定義格式如：
let params={
//ID是存儲過程的第一個參數，要去掉@符號
ID:{
//sqlType是該ID參數在sqlserver中的類型
sqlType:sql.sqlserver.Int,
//direction是表明ID參數是輸入還是輸出(output)參數
direction:sql.direction.Input,
//該ID參數的值
inputValue:1
},
//Name是存儲過程的第二個參數，要去掉@符號
Name:{
sqlType:sqlHelper.sqlserver.Int,
direction:sqlHelper.direction.Output,
outputValue:null
}
};
* @param {function} func 回調函數 共有四個參數 error:錯誤信息 recordsets:查詢的表結果 returnValue:存儲過程的返回值 affected:影響的行數
*/
sql.execute=async function(procedure,params){
try {
//創建數據庫連接池
var pool = new mssql.ConnectionPool(config);
await pool.connect();//連接數據庫
let request = pool.request();
if (params != null) {
for (let index in params) {
if (params[index].direction == sql.direction.Output) {
request.output(index, params[index].sqlType);
}
else {
request.input(index, params[index].sqlType, params[index].inputValue);
}
}
}
// result 成功返回該結構
// {
// recordsets,
// recordset: recordsets && recordsets[0],
// output,
// rowsAffected,
// returnValue
// }
let result =await request.execute(procedure);
for (let index in params) {
if (params[index].direction == sql.direction.Output) {
params[index].outputValue = request.parameters[index].value;
}
}
return {state:true,data:result};
}catch(err){
return {state:false,data:err};
}finally {
await doRelease(pool);
}
};
/**
* 執行sql文本(帶params參數)
* @param {string} sqlText 執行的sql語句
* @param {JSON} params sql語句中的參數
* @param {function} func 回調函數 共有兩個個參數 error:錯誤消息 recordsets:查詢的結果
*/
sql.queryWithParams=async function(sqlText,params){
try {
//創建數據庫連接池
var pool = new mssql.ConnectionPool(config);
await pool.connect();
let request = pool.request();
request.multiple=true;
if (params != null) {
for(let index in params){
request.input(index,params[index].sqlType,params[index].inputValue);
}
}
let result = await request.query(sqlText);
return {state:true,data:result};
}catch(err){
return {state:false,data:err};
}finally {
await doRelease(pool);
}
};
/**
* 執行sql文本
* @param {string} sqlText 執行的sql語句
* @param {function} func 回調函數 共有兩個個參數 error:錯誤消息 recordsets:查詢的結果
*/
sql.query=function(sqlText){
return sql.queryWithParams(sqlText,null);
}
/**
* 執行大批量數據的插入
* @param {sqlserver.Table} table 需要插入的數據表
* 數據表的定義如下：
let table=new sql.sqlserver.Table('UserInfoTest');
table.create=true;
table.columns.add('name',sqlHelper.sqlserver.NVarChar(50),{nullable:true});
table.columns.add('pwd',sqlHelper.sqlserver.VarChar(200),{nullable:true});
table.rows.add('張1','jjasdfienf');
table.rows.add('張2','jjasdfienf');
table.rows.add('張3','jjasdfienf');
* @param {function} func 回調函數 共有兩個參數 error:錯誤信息 rowcount:插入數據的行數
*/
sql.bulkInsert=async function(tableObj){
try {
if(tableObj) {
//創建數據庫連接池
var pool = new mssql.ConnectionPool(config);
await pool.connect();
let request = pool.request()
let result = await request.bulk(tableObj);
return {state:true,data:result};
}
else
{
return {state:false,data:"table parameter undefined!"};
}
}catch(err){
return {state:false,data:err};
}finally {
await doRelease(pool);
}
};
/**
* 如果需要處理大批量的數據行，通常應該使用流
* @param {string} sqlText 需要執行的sql文本
* @param {JSON} params 輸入參數
* @param {JSON} func 表示一個回調函數的JSON對象，如下所示：
* {
error:function(err){
console.log(err);
},
columns:function(columns){
console.log(columns);
},
row:function(row){
console.log(row);
},
done:function(affected){
console.log(affected);
}
*/
sql.queryViaStreamWithParams= async function(sqlText,params,func){
try {
//創建數據庫連接池
var pool = new mssql.ConnectionPool(config);
await pool.connect();
let request = pool.request();
request.stream =true;
if(params){
for(let index in params){
request.input(index,params[index].sqlType,params[index].inputValue);
}
}
request.query(sqlText);
request.on('recordset', function(columns){
//columns是一個JSON對象，表示 返回數據表的整個結構，包括每個字段名稱以及每個字段的相關屬性
//如下所示
/*
{ id:
{ index: 0,
name: 'id',
length: undefined,
type: [sql.Int],
scale: undefined,
precision: undefined,
nullable: false,
caseSensitive: false,
identity: true,
readOnly: true },
name:
{ index: 1,
name: 'name',
length: 100,
type: [sql.NVarChar],
scale: undefined,
precision: undefined,
nullable: true,
caseSensitive: false,
identity: false,
readOnly: false },
Pwd:
{ index: 2,
name: 'Pwd',
length: 200,
type: [sql.VarChar],
scale: undefined,
precision: undefined,
nullable: true,
caseSensitive: false,
identity: false,
readOnly: false } }
*/
func.columns(columns);
});
request.on('row', function(row) {
//row是一個JSON對象，表示 每一行的數據，包括字段名和字段值
//如 { id: 1004, name: 'jsw', Pwd: '12345678' }
//如果行數較多，會多次進入該方法，每次只返回一行
func.row(row);
});
request.on('error',async function(err) {
//err是一個JSON對象，表示 錯誤信息
//如下所示：
/*
{ [RequestError: Incorrect syntax near the keyword 'from'.]
name: 'RequestError',
message: 'Incorrect syntax near the keyword \'from\'.',
code: 'EREQUEST',
number: 156,
lineNumber: 1,
state: 1,
class: 15,
serverName: '06-PC',
procName: '' }
*/
await doRelease(pool);
func.error(err);
});
request.on('done',async function(affected) {
//affected是一個數值，表示 影響的行數
//如 0
//該方法是最后一個執行
await doRelease(pool);
await func.done(affected);
});
}catch(err){
console.log("代碼執行異常poolFunc，error");
await doRelease(pool);
func.error(err);
}
};
/**
* 如果需要處理大批量的數據行，通常應該使用流
* @param {string} sqlText 需要執行的sql文本
* @param {JSON} func 表示一個回調函數的JSON對象，如下所示：
* {
error:function(err){
console.log(err);
},
columns:function(columns){
console.log(columns);
},
row:function(row){
console.log(row);
},
done:function(affected){
console.log(affected);
}
*/
sql.queryViaStream=async function(sqlText,func){
await sql.queryViaStreamWithParams(sqlText,null,func);
};
//釋放數據庫連接到數據
async function doRelease(connection) {
try {
if(connection)
{
await connection.close();//釋放連接，將連接放回池中
}
}
catch (err) {
console.error(err.message);
}
}
module.exports=sql;
2.長連接寫法
加載文件的時候，自動先連接數據庫
//導入mssql模塊 //基於版本@4.2.1 安裝指令npm install mssql@4.2.1
// pool.close()非常重要，只創建，不關閉會造成非常嚴重的內存泄漏。 關閉池中的所有活動連接。
let mssql = require("mssql");
//引用配置參數模塊
// let configFile = require("./config");
//引用通用方法模塊
// let common = require("./commonHelper");
//數據庫異常
mssql.on('error', err => {
console.log("mssql異常原因：" + err.message);
});
let sql = {};
//sql參數的類型
sql.direction = {
//輸入參數
Input: "input",
//輸出參數
Output: "output",
//返回參數
Return: "return"
};
//配置存儲過程是的輸出輸入
sql.sqlserver = mssql;
let config = null;
/**
* 初始化配置
*/
function init_config() {
if (config == null) {
//默認config對象
config = require("./config").mssql_config;
}
}
/**
* @description 這個好像沒有用到,是直接通過initPool進行初始化
* 初始化連接參數
* @param {string} user 用戶名
* @param {string} password 密碼
* @param {string} server 服務器地址
* @param {string} database 數據庫名稱
* @param {string} port 數據庫端口
*/
sql.initConfig = async function (user, password, server, database, port) {
config.user = user;
config.password = password;
config.server = server;
config.database = database;
config.port = port;
try {
if (pool != null) {
await mssql.close();//關閉原有的數據庫連接
}
} catch (err) {
// common.consoleLog(err);
console.error(err);
}
pool = null;
await initPool();
}
let pool = null;
async function initPool() {
init_config();
if (pool == null) {
//實例化一個knex對象
//不能直接初始化,出現多個連接對象,會報錯
// global.knex = require('knex')({
// client: 'mssql',
// connection: {
// host: config.server,
// user: config.user,
// password: config.password,
// port: config.port,
// database: config.database
// }
// });
// sql.knex = global.knex;
// return;
// return new Promise((resolve, reject) => {
pool = await (new mssql.ConnectionPool(config)).connect();
// pool = await mssql.connect(config, function (err) {
// if (err) {
// throw err;
// }
// common.consoleLog("初始化mssql數據庫連接");
// resolve();
// });
// });
}
return pool;
};
// (async () => {
// //是否初始化數據庫連接
// if (configFile.init_mssql) {
// try {
// common.consoleLog("初始化mssql數據庫連接");
// await initPool();
// } catch (err) {
// console.log(err);
// }
// }
// })();
/**
* 執行存儲過程
* @param {string} procedure 存儲過程名稱
* @param {JSON} params 存儲過程參數
* params的定義格式如：
let params={
//ID是存儲過程的第一個參數，要去掉@符號
ID:{
//sqlType是該ID參數在sqlserver中的類型
sqlType:sql.sqlserver.Int,
//direction是表明ID參數是輸入還是輸出(output)參數
direction:sql.direction.Input,
//該ID參數的值
inputValue:1
},
//Name是存儲過程的第二個參數，要去掉@符號
Name:{
sqlType:sqlHelper.sqlserver.Int,
direction:sqlHelper.direction.Output,
outputValue:null
}
};
* @param {function} func 回調函數 共有四個參數 error:錯誤信息 recordsets:查詢的表結果 returnValue:存儲過程的返回值 affected:影響的行數
*/
sql.executePro = async function (procedure, params) {
try {
//創建數據庫連接池
// var pool = new mssql.ConnectionPool(config);
// await pool.connect();//連接數據庫
pool = await initPool();//連接數據庫
let request = pool.request();
if (params != null) {
for (let index in params) {
if (params[index].direction == sql.direction.Output) {
request.output(index, params[index].sqlType);
} else {
request.input(index, params[index].sqlType, params[index].inputValue);
}
}
}
// result 成功返回該結構
// {
// recordsets,
// recordset: recordsets && recordsets[0],
// output,
// rowsAffected,
// returnValue
// }
let result = await request.execute(procedure);
//獲取存儲過程Output返回的數據
if(params != null)
{
for (let index in params) {
if (params[index].direction == sql.direction.Output) {
params[index].outputValue = request.parameters[index].value;
}
}
}
return {state: true, data: result};
} catch (err) {
await doReconnection(err.message, procedure);
//判斷是否數據庫連接出問題，是的重新連接數據庫
if (VerifyConnection(err.message)) {
let result = await sql.execute(procedure, params);
return result;
}
return {state: false, data: err.message};
}
};
/**
* 執行sql文本(帶params參數)
* @param {string} sqlText 執行的sql語句
* @param {JSON} params sql語句中的參數
* let params={
//ID是存儲過程的第一個參數，要去掉@符號
ID:{
//sqlType是該ID參數在sqlserver中的類型
sqlType:sql.sqlserver.Int,
//該ID參數的值
inputValue:1
}
}
* @param {function} func 回調函數 共有兩個個參數 error:錯誤消息 recordsets:查詢的結果
*/
sql.queryWithParams = async function (sqlText, params) {
try {
//創建數據庫連接池
// var pool = new mssql.ConnectionPool(config);
// await pool.connect();
await initPool();//連接數據庫
let request = pool.request();
request.multiple = true;
if (params != null) {
for (let index in params) {
request.input(index, params[index].sqlType, params[index].inputValue);
}
}
let result = await request.query(sqlText);
return {state: true, data: result};
} catch (err) {
console.error(sqlText, err);
await doReconnection(err.message, sqlText + common.JSON_stringify(params));
//判斷是否數據庫連接出問題，是的重新連接數據庫
if (VerifyConnection(err.message)) {
let result = await sql.queryWithParams(sqlText, params);
return result;
}
return {state: false, data: err.message};
}
};
/**
* 執行sql文本
* @param {string} sqlText 執行的sql語句
* @param {function} func 回調函數 共有兩個個參數 error:錯誤消息 recordsets:查詢的結果
*/
sql.query = function (sqlText) {
return sql.queryWithParams(sqlText, null);
}
/**
* 執行大批量數據的插入
* @param {sqlserver.Table} table 需要插入的數據表
* 數據表的定義如下：
let table=new sql.sqlserver.Table('UserInfoTest');
table.create=true;
table.columns.add('name',sqlHelper.sqlserver.NVarChar(50),{nullable:true});
table.columns.add('pwd',sqlHelper.sqlserver.VarChar(200),{nullable:true});
table.rows.add('張1','jjasdfienf');
table.rows.add('張2','jjasdfienf');
table.rows.add('張3','jjasdfienf');
* @param {function} func 回調函數 共有兩個參數 error:錯誤信息 rowcount:插入數據的行數
*/
sql.bulkInsert = async function (tableObj) {
try {
if (tableObj) {
//創建數據庫連接池
// var pool = new mssql.ConnectionPool(config);
// await pool.connect();
await initPool();//連接數據庫
let request = pool.request()
let result = await request.bulk(tableObj);
return {state: true, data: result};
} else {
return {state: false, data: "table parameter undefined!"};
}
} catch (err) {
await doReconnection(err.message, common.JSON_stringify(tableObj));
//判斷是否數據庫連接出問題，是的重新連接數據庫
if (VerifyConnection(err.message)) {
let result = await sql.bulkInsert(tableObj);
return result;
}
return {state: false, data: err.message};
}
};
/**
* 如果需要處理大批量的數據行，通常應該使用流
* @param {string} sqlText 需要執行的sql文本
* @param {JSON} params 輸入參數
* @param {JSON} func 表示一個回調函數的JSON對象，如下所示：
* {
error:function(err){
console.log(err);
},
columns:function(columns){
console.log(columns);
},
row:function(row){
console.log(row);
},
done:function(affected){
console.log(affected);
}
*/
sql.queryViaStreamWithParams = async function (sqlText, params, func) {
try {
//創建數據庫連接池
// var pool = new mssql.ConnectionPool(config);
// await pool.connect();
await initPool();//連接數據庫
let request = pool.request();
request.stream = true;
if (params) {
for (let index in params) {
request.input(index, params[index].sqlType, params[index].inputValue);
}
}
request.query(sqlText);
request.on('recordset', function (columns) {
//columns是一個JSON對象，表示 返回數據表的整個結構，包括每個字段名稱以及每個字段的相關屬性
//如下所示
/*
{ id:
{ index: 0,
name: 'id',
length: undefined,
type: [sql.Int],
scale: undefined,
precision: undefined,
nullable: false,
caseSensitive: false,
identity: true,
readOnly: true },
name:
{ index: 1,
name: 'name',
length: 100,
type: [sql.NVarChar],
scale: undefined,
precision: undefined,
nullable: true,
caseSensitive: false,
identity: false,
readOnly: false },
Pwd:
{ index: 2,
name: 'Pwd',
length: 200,
type: [sql.VarChar],
scale: undefined,
precision: undefined,
nullable: true,
caseSensitive: false,
identity: false,
readOnly: false } }
*/
func.columns(columns);
});
request.on('row', function (row) {
//row是一個JSON對象，表示 每一行的數據，包括字段名和字段值
//如 { id: 1004, name: 'jsw', Pwd: '12345678' }
//如果行數較多，會多次進入該方法，每次只返回一行
func.row(row);
});
request.on('error', async function (err) {
//err是一個JSON對象，表示 錯誤信息
//如下所示：
/*
{ [RequestError: Incorrect syntax near the keyword 'from'.]
name: 'RequestError',
message: 'Incorrect syntax near the keyword \'from\'.',
code: 'EREQUEST',
number: 156,
lineNumber: 1,
state: 1,
class: 15,
serverName: '06-PC',
procName: '' }
*/
func.error(err);
});
request.on('done', async function (affected) {
//affected是一個數值，表示 影響的行數
//如 0
//該方法是最后一個執行
await func.done(affected);
});
} catch (err) {
console.error("代碼執行異常poolFunc，error");
await doReconnection(err.message, sqlText);
//判斷是否數據庫連接出問題，是的重新連接數據庫
if (VerifyConnection(err.message)) {
await sql.queryViaStreamWithParams(sqlText, params, func);
} else {
func.error(err);
}
}
};
/**
* 如果需要處理大批量的數據行，通常應該使用流
* @param {string} sqlText 需要執行的sql文本
* @param {JSON} func 表示一個回調函數的JSON對象，如下所示：
* {
error:function(err){
console.log(err);
},
columns:function(columns){
console.log(columns);
},
row:function(row){
console.log(row);
},
done:function(affected){
console.log(affected);
}
*/
sql.queryViaStream = async function (sqlText, func) {
await sql.queryViaStreamWithParams(sqlText, null, func);
};
/**
* 驗證數據庫表是否存在
* 返回boolean類型，true或者false
* */
sql.verifyHaveTable = async function (tableName) {
let select_sql = "select * from sysobjects where name =@name";
let param = {
name: {sqlType: mssql.VarChar(128), inputValue: tableName},
}
let result = await sql.queryWithParams(select_sql, param)
if (result.state) {
if (result.data.recordset.length > 0) {
return true;
}
}
return false;
}
//重新連接數據庫
async function doReconnection(message, sqlText) {
try {
let time = new Date().Format("HH:mm:ss");
common.writeLog("mssql_connection", time + common.partition + sqlText + common.partition + message);
//釋放連接，重新連接mssql
if (VerifyConnection(message)) {
if (common.isNotEmpty(pool))
await pool.close();
pool = null;//重新初始化mssql連接
await mssql.close();
}
} catch (err) {
common.consoleLog(err);
}
}
//驗證數據庫連接是否正常
function VerifyConnection(message) {
common.consoleLog("VerifyConnection：" + message);//輸出錯誤信息
//釋放連接，重新連接mssql
if (message.search("Failed to connect") >= 0
|| message.search("Connection is closed") >= 0
|| message.search("Connection lost") >= 0
|| message.search("sql.close()") >= 0) {
return true;
}
return false;
}
module.exports = sql;