nodejs事务

nodejs事务插入数据：
function execTrans(sqlparamsEntities, callback) {
    pool.getConnection(function (err, connection) {
 if (err) {
 return callback(err, null);
 }
 connection.beginTransaction(function (err) {
 if (err) {
 return callback(err, null);
 }
 var funcAry = [];
 sqlparamsEntities.forEach(function (sql_param) {
 var temp = function (cb) {
 var sql = sql_param.sql;
 var param = sql_param.params;
 connection.query(sql, param, function (tErr, rows, fields) {
 if (tErr) {
 connection.rollback(function () {
 console.log("事务失败，" + sql_param + "，ERROR：" + tErr);
 throw tErr;
 });
 } else {
 return cb(null, 'ok');
 }
 })
 };
 funcAry.push(temp);
 });

 async.series(funcAry, function (err, result) {
 console.log("transaction error: " + err);
 if (err) {
 connection.rollback(function (err) {
 console.log("transaction error: " + err);
 connection.release();
 return callback(err, null);
 });
 } else {
 connection.commit(function (err, info) {
 console.log("transaction info: " + JSON.stringify(info));
 if (err) {
 console.log("执行事务失败，" + err);
 connection.rollback(function (err) {
 console.log("transaction error: " + err);
 connection.release();
 return callback(err, null);
 });
 } else {
 connection.release();
 return callback(null, info);
 }
 })
 }
 })
 });
 });
}



备注：sqlparamsEntities数组必须使用_getNewSqlParamEntity把sql和参数封装

function _getNewSqlParamEntity(sql, params, callback) {
    if (callback) {
 return callback(null, {
 sql: sql,
 params: params
 });
 }
 return {
 sql: sql,
 params: params
 };
}