一种优雅的用于Node.js的MySQL包的异步事务函数封装方式

2024-07-20 06:58:52

引言

众所周知，Node.js中的 mysql 包提供的接口均为异步接口，这样虽然有效提升了性能，但在大多数业务场景中，我们需要将其转为同步方式使用。
对于常规的单次查询，可以使用一个简单的Promise完成如下封装：

const query = async function (sql, values) {
	return new Promise((resolve, reject) => {
		pool.getConnection(function (err, connection) {
			if (err) {
				reject(err);
			} else {
				connection.query(sql, values, (err, rows) => {
					if (err) {
						reject(err);
					} else {
						resolve(rows);
					}
					connection.release();
				});
			}
		});
	});
};

在上述代码中，我们将 getConnection 方法和 connection.query 方法封装为一个异步的查询方法 query，调用该函数，并传入SQL语句模板和值即可完成查询，并获得查询结果。
但是在复杂的多次事务查询，我们可能需要连续执行多条SQL语句，在执行SQL语句的间隙，可能还需要进行一系列的逻辑判断，或者执行其他可能抛出异常的操作。
因此，事务查询相比较于单次查询，封装起来就不太容易。如果不进行封装，在业务代码中就会出现大量的回调，降低代码的可读性和稳定性。
本文拟提出一种新的封装方式，将多次事务查询功能封装为一个简单的异步方法。通过传入一个简单的异步函数来完成上述功能，该函数使用起来是这样的：

await execTransaction(async (query, commit, rollback)=>{
    try{
         await query(sql1, values1)
         doSomethingElse();
         let something = await query(sql2, values2)
         await doSomethingAsyncElse();
         await query(sql3, values3)
         if(condition) {
         	commit();
         } else {
			reject();
         }
     } catch(e){
         console.error(e);
         rollback();
         throw e;
     }
 })

可以看见，在上述代码中，在我们调用execTransaction方法的过程中，整个写法非常符合直觉，回调函数中提供了query、commit、rollback三个方法，分别提供执行单次查询、提交事务和回滚事务三种操作。可以灵活随意地使用。
接下来，笔者将带你一起完成这个封装方法。

先上代码

废话不多说，首先先上代码，如果你只是想直接使用，而不想研究原理，可以直接把代码拿走：

const execTransaction = (callback) => {
   return new Promise((resolve, reject) => {
       pool.getConnection(function (err, connection) {
           if (err) {
               return reject(err)
           }
           connection.beginTransaction(err => {
               if (err) {
                   return reject('开启事务失败')
               }
               let _resolve = resolve;
               let _reject = reject;
               let lastResult = undefined;
               let query = (sql, values) => {
                   return new Promise((resolve, reject) => {
                       connection.query(sql, values, (e, rows, fields) => {
                           if(e) {
                               setTimeout(()=>{
                                   reject(e);
                               })
                               _reject(e);
                           } else {
                               lastResult = rows;
                               resolve(rows);
                           }
                       })
                   })
               }
               let rollback = () => {
                   connection.rollback(() => {
                       console.log('数据操作回滚')
                       connection.release()
                       reject(err);
                   })
               }
               let commit = () => {
                   connection.commit((error) => {
                       if (error) {
                           console.log('事务提交失败')
                           connection.release()
                           reject(error)
                       } else {
                           connection.release()  // 释放链接
                           resolve(lastResult)
                       }
                   })
               }
               callback(query, commit, rollback);
           })
       });
   })
}

接下来，笔者将逐一分析上述代码的实现逻辑。

返回一个Promise

由于我们要实现的 execTransaction 方法是异步的，因此必然要返回一个Promise来实现异步。该Promise在resolve时表示事务执行成功，在reject时则表示事务执行失败。
在Promise中，我们首先创建一个connection，然后再调用connection.beginTransaction方法，这些方法都是mysql库提供给我们的，它们都采用了回调函数的写法。
接下来，在该方法中做文章。

Callback函数

整段代码的核心实质是一个callback函数的调用。
在调用该方法时，用户提交了一个回调函数，因此在这个函数中，回调函数一定会被调用，并传入三个方法query、commit、rollback，接下来的工作就是实现这三个方法。

实现三个方法

query方法

此函数内部创建并返回一个新的Promise对象，用于执行 SQL 语句。
使用 connection.query 方法执行 SQL 查询。查询结果或错误通过回调函数返回。
如果查询出错，首先通过 setTimeout 延迟执行 reject 函数，然后通过 _reject 直接拒绝外层的 Promise。
如果查询成功，将结果存储在 lastResult 中，并通过 resolve 解决内部 Promise。

rollback方法

当需要回滚事务时调用，例如在查询过程中出错。
调用 connection.rollback 方法回滚所有更改，并释放数据库连接。
通过外部的 reject 函数拒绝外层的 Promise。

commit 方法

用于在所有操作成功完成后提交事务。
调用 connection.commit 提交事务。如果提交时出错，打印错误消息，并通过 reject 函数拒绝外层 Promise。
如果提交成功，释放连接并通过 resolve 解决外层 Promise，返回 lastResult 作为操作结果。

一个使用该方法的实际例子（由ChatGPT提出）

这是一个使用 execTransaction 方法的示例，假设我们需要在一个电商系统中添加订单数据。这个例子包括两个步骤：首先插入订单主信息，然后插入订单的详细商品信息。如果其中任何一个步骤失败，我们需要回滚整个事务以保持数据的一致性。

// 引入已定义的 execTransaction 方法
const { execTransaction } = require('./path_to_your_transaction_file');

// 使用 execTransaction 方法执行事务
execTransaction(async (query, commit, rollback) => {
    try {
        // 第一步：插入订单主信息
        const orderSql = 'INSERT INTO orders (customer_id, order_date, total) VALUES (?, ?, ?)';
        const orderValues = [123, '2023-07-19', 299.99];
        const orderResult = await query(orderSql, orderValues);
        console.log('订单主信息已添加', orderResult);

        // 第二步：插入订单详情
        const detailSql = 'INSERT INTO order_details (order_id, product_id, quantity, price) VALUES (?, ?, ?, ?)';
        const orderDetails = [
            [orderResult.insertId, 1, 2, 59.99],  // 商品1
            [orderResult.insertId, 2, 1, 180.00]  // 商品2
        ];
        for (let detail of orderDetails) {
            const detailResult = await query(detailSql, detail);
            console.log('订单详细信息已添加', detailResult);
        }

        // 如果所有操作成功，提交事务
        await commit();
        console.log('事务已成功提交');
    } catch (error) {
        console.log('事务处理中出错，进行回滚', error);
        // 如果出现错误，回滚事务
        await rollback();
    }
}).then(() => {
    console.log('所有操作完成');
}).catch((error) => {
    console.log('事务处理失败', error);
});

在这个例子中：

我们首先尝试插入订单的主信息。如果主信息插入成功，我们接着插入订单详细的商品信息。对于每一步操作，我们都使用 query 函数进行数据库查询，它会在成功时返回结果，在失败时抛出异常。如果所有数据库操作都成功，我们调用 commit 函数提交事务。
如果任何一个数据库操作失败，我们捕获异常，并调用 rollback 函数回滚整个事务，以确保数据库状态的一致性和准确性。
execTransaction本身返回的是一个 Promise，我们可以用 .then 和 .catch来处理正常完成或发生错误的情况。

这样，整个事务操作的逻辑就非常清晰，并且能够有效地处理可能出现的错误。

总结

本文提出了一种优雅的基于Node.js的MySQL包的异步事务函数封装方式，该函数接收一个回调函数 callback 作为参数，并返回一个 Promise 对象，以支持异步操作和错误处理。
使用该函数可以使得进行复杂的数据库操作时，在一个封闭的事务中顺序执行多个查询，并且只有在所有操作成功完成后才提交事务，否则回滚所有更改。这有助于保证数据的一致性和完整性。

码农公寓

引言