MSSQL 是Nodejs用于连接Microsoft SQL Server 的插件。
在nodejs中，mssql模块支持sqlserver数据库操作。今天将mssql模块的某些功能封装为一个类，方便以后调用。封装的功能有执行存储过程，执行查询语句操作等。如果本篇文章对大家有帮助，那就再好不过了！
如果需要扩展事务和其他功能，请到官网https://www.npmjs.com/package/mssql中看文档，里面有示例。
安装方法
安装指令npm install mssql@4.2.
配置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;
版权声明：除特别声明外，本站所有文章皆是本站原创，转载请以超链接形式注明出处！
