Nodejs MSSQL详细解读

MSSQL 是Nodejs用于连接Microsoft SQL Server 的插件。

在nodejs中，mssql模块支持sqlserver数据库操作。今天将mssql模块的某些功能封装为一个类，方便以后调用。封装的功能有执行存储过程，执行查询语句操作等。如果本篇文章对大家有帮助，那就再好不过了！

如果需要扩展事务和其他功能，请到官网https://www.npmjs.com/package/mssql中看文档，里面有示例。

安装方法

安装指令npm install [email protected]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 [email protected]
// 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 [email protected]
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;

//默认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;
    pool = null;
}

let pool = null;
async function initPool()
{
    if(pool == null)
    {
        pool = await mssql.connect(config);
    }
};

(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.execute=async function(procedure,params){
    try {
        //创建数据库连接池
        // var pool = new mssql.ConnectionPool(config);
        // await pool.connect();//连接数据库
        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);
        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);
        return {state:false,data:err};
    }
};

/**
 * 执行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){
        await doReconnection(err.message,sqlText);
        return {state:false,data:err};
    }
};

/**
 * 执行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));
        return {state:false,data:err};
    }
};

/**
 * 如果需要处理大批量的数据行，通常应该使用流
 * @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.log("代码执行异常poolFunc，error");
        await doReconnection(err.message,sqlText);
        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 doReconnection(message,sqlText) {
    try{
        let time=new Date().Format("HH:mm:ss");
        common.writeLog("mssql_connection",time  + common.partition +sqlText+ common.partition + message);
        console.log(message);
        //释放连接，重新连接mssql
        if(message.search("Failed to connect")>=0 || message.search("Connection is closed")>=0)
        {
            pool = null;//重新初始化mssql连接
            await mssql.close();
        }
    }
    catch (err) {
        console.log(err);
    }
}

module.exports=sql;