搭建node服務(wù)（二）：操作MySQL

mysql"> 搭建node服務(wù)（二）：操作MySQL

為了進(jìn)行復(fù)雜信息的存儲和查詢，服務(wù)端系統(tǒng)往往需要數(shù)據(jù)庫操作。數(shù)據(jù)庫分為關(guān)系型數(shù)據(jù)庫和非關(guān)系型數(shù)據(jù)庫，關(guān)系型數(shù)據(jù)庫有MySQL、Oracle、SQL Server等，非關(guān)系型數(shù)據(jù)庫有Redis（常用來做緩存）、MongoDB等。MySQL是目前很流行的數(shù)據(jù)庫，本文將要介紹如何在node服務(wù)中進(jìn)行MySQL數(shù)據(jù)庫操作。

一、 安裝依賴

npm install mysql --save

或者

yarn add mysql

二、建立連接

要想進(jìn)行數(shù)據(jù)庫操作就需要和數(shù)據(jù)庫建立連接，然后通過連接進(jìn)行數(shù)據(jù)庫的操作。MySQL的數(shù)據(jù)庫連接方式有以下幾種：

• mysql.createConnection() 每次請求建立一個連接
• mysql.createPool() 創(chuàng)建連接池，從連接池中獲取連接
• mysql.createPoolCluster() 創(chuàng)建連接池集群，連接池集群可以提供多個主機(jī)連接

mysqljs文檔中推薦使用第一種方式：每次請求建立一個連接，但是由于頻繁的建立、關(guān)閉數(shù)據(jù)庫連接，會極大的降低系統(tǒng)的性能，所以我選擇了使用連接池的方式，如果對性能有更高的要求，安裝了MySQL 集群，可以選擇使用連接池集群。

1. 數(shù)據(jù)庫配置

將數(shù)據(jù)庫相關(guān)的配置添加到公用的配置文件中，方便項目的初始化。

• config.js

module.exports = {
    …
    // mysql數(shù)據(jù)庫配置
    mysql: {
        // 主機(jī)
        host: 'localhost',
        // 端口
        port: 3306,
        // 用戶名
        user: 'root',
        // 密碼
        password: '123456',
        // 數(shù)據(jù)庫名
        database: 'server-demo',
        // 連接池允許創(chuàng)建的最大連接數(shù)，默認(rèn)值為10
        connectionLimit: 50,
        // 允許掛起的最大連接數(shù),默認(rèn)值為0,代表掛起的連接數(shù)無限制
        queueLimit: 0
    }
};

connectionLimit 和 queueLimit 是數(shù)據(jù)連接池特有的配置項。

• connectionLimit 是指連接池允許創(chuàng)建的最大連接數(shù)，默認(rèn)值為10。當(dāng)獲取連接時，如果連接池中有空閑的連接則直接返回一個空閑連接。如果所有連接都被占用，則判斷連接池中的連接數(shù)是否達(dá)到了允許的最大數(shù)，如果未達(dá)到則創(chuàng)建新的連接，如果已達(dá)到則獲取連接的請求掛起，等待其他請求完成操作后釋放的連接。
• queueLimit 是指允許掛起的最大連接數(shù)，默認(rèn)值為0，代表掛起的連接數(shù)無限制。當(dāng)連接池中允許創(chuàng)建的所有連接都被占用時，獲取連接的請求掛起，等待可用的連接，所有掛起的請求形成一個隊列，queueLimit則是指這個隊列的最大長度。需要注意的是，當(dāng)queueLimit為0時并不表示不允許掛起，而是表示對掛起的數(shù)目沒有限制。

2. 創(chuàng)建連接池

• db/pool.js

/**
 * 數(shù)據(jù)庫連接池
 */
const mysql = require('mysql');
const config = require('../config');
// 創(chuàng)建數(shù)據(jù)庫連接池
const pool = mysql.createPool(config.mysql);
pool.on('acquire', function (connection) {
    console.log(`獲取數(shù)據(jù)庫連接 [${connection.threadId}]`);
});
pool.on('connection', function (connection) {
    console.log(`創(chuàng)建數(shù)據(jù)庫連接 [${connection.threadId}]`);
});
pool.on('enqueue', function () {
    console.log('正在等待可用數(shù)據(jù)庫連接');
});
pool.on('release', function (connection) {
    console.log(`數(shù)據(jù)庫連接 [${connection.threadId}] 已釋放`);
});
module.exports = pool;

創(chuàng)建數(shù)據(jù)庫連接池pool后，就可以通過pool獲取數(shù)據(jù)庫連接了，另外通過監(jiān)聽連接池的事件可以了解連接池中連接的使用情況。
如果將connectionLimit 設(shè)為2，queueLimit 設(shè)為0，當(dāng)同時有5個請求獲取數(shù)據(jù)庫連接時，線程池的事件日志如下：

正在等待可用數(shù)據(jù)庫連接
正在等待可用數(shù)據(jù)庫連接
正在等待可用數(shù)據(jù)庫連接
創(chuàng)建數(shù)據(jù)庫連接 [1011]
獲取數(shù)據(jù)庫連接 [1011]
數(shù)據(jù)庫連接 [1011] 已釋放
獲取數(shù)據(jù)庫連接 [1011]
創(chuàng)建數(shù)據(jù)庫連接 [1012]
獲取數(shù)據(jù)庫連接 [1012]
數(shù)據(jù)庫連接 [1011] 已釋放
獲取數(shù)據(jù)庫連接 [1011]
數(shù)據(jù)庫連接 [1012] 已釋放
獲取數(shù)據(jù)庫連接 [1012]
數(shù)據(jù)庫連接 [1011] 已釋放
數(shù)據(jù)庫連接 [1012] 已釋放

由于線程池允許的最大連接數(shù)是2，5個請求中會有2個請求能夠得到連接，另外3個請求掛起等待可用連接。由于創(chuàng)建數(shù)據(jù)庫連接的代價比較大，線程池在創(chuàng)建連接時采用懶漢式，也就是，用到時才創(chuàng)建。先得到連接的請求在完成操作后釋放連接，放回到連接池，然后掛起的請求從線程池取出空閑的連接進(jìn)行操作。

三、執(zhí)行操作

由于mysql 模塊的接口都為回調(diào)方式的，為了操作方便簡單地將接口封裝為Promise，相關(guān)方法封裝如下：

const pool = require('./pool');
// 獲取連接
function getConnection () {
    return new Promise((resolve, reject) => {
        pool.getConnection((err, connection) => {
            if (err) {
                console.error('獲取數(shù)據(jù)庫連接失?。?, err)
                reject(err);
            } else {
                resolve(connection);
            }
        });
    });
}
// 開始數(shù)據(jù)庫事務(wù)
function beginTransaction (connection) {
    return new Promise((resolve, reject) => {
        connection.beginTransaction(err => {
            if (err) {
                reject(err);
            } else {
                resolve();
            }
        });
    });
}
// 提交數(shù)據(jù)庫操作
function commit (connection) {
    return new Promise((resolve, reject) => {
        connection.commit(err => {
            if (err) {
                reject(err);
            } else {
                resolve();
            }
        });
    })
}
// 回滾數(shù)據(jù)庫操作
function rollback (connection) {
    return new Promise((resolve, reject) => {
        connection.rollback(err => {
            if (err) {
                reject(err);
            } else {
                resolve();
            }
        });
    })
}

1. 執(zhí)行普通操作

對于不需要使用事務(wù)的普通操作，獲取數(shù)據(jù)庫連接connection后，使用connection進(jìn)行數(shù)據(jù)庫操作，完成后釋放連接到連接池，則執(zhí)行完成一次操作。

• db/execute.js

/**
 * 執(zhí)行數(shù)據(jù)庫操作【適用于不需要事務(wù)的查詢以及單條的增、刪、改操作】
 * 示例：
 * let func = async function(conn, projectId, memberId) { ... };
 * await execute( func, projectId, memberId);
 * @param func 具體的數(shù)據(jù)庫操作異步方法（第一個參數(shù)必須為數(shù)據(jù)庫連接對象connection）
 * @param params func方法的參數(shù)（不包含第一個參數(shù) connection）
 * @returns {Promise.<*>} func方法執(zhí)行后的返回值
 */
async function execute (func, ...params) {
    let connection = null;
    try {
        connection = await getConnection()
        let result = await func(connection, ...params);
        return result
    } finally {
        connection && connection.release && connection.release();
    }
}

2. 執(zhí)行事務(wù)操作

對于很多業(yè)務(wù)都需要執(zhí)行事務(wù)操作，例如：銀行轉(zhuǎn)賬，A賬戶轉(zhuǎn)賬給B賬戶 100元，這個業(yè)務(wù)操作需要執(zhí)行兩步，從A賬戶減去100元，然后給B賬戶增加100元。兩個子操作必須全部執(zhí)行成功才能完成完整的業(yè)務(wù)操作，如果任意子操作執(zhí)行失敗就需要撤銷之前的操作，進(jìn)行回滾。

對于需要使用事務(wù)的操作，獲取數(shù)據(jù)庫連接connection后，首先需要調(diào)用connection.beginTransaction() 開始事務(wù)，然后使用connection進(jìn)行多步操作，完成后執(zhí)行connection.commit() 進(jìn)行提交，則執(zhí)行完成一次事務(wù)操作。如果在執(zhí)行過程中出現(xiàn)了異常，則執(zhí)行connection.rollback() 進(jìn)行回滾操作。

• db/execute.js

/**
 * 執(zhí)行數(shù)據(jù)庫事務(wù)操作【適用于增、刪、改多個操作的執(zhí)行，如果中間數(shù)據(jù)操作出現(xiàn)異常則之前的數(shù)據(jù)庫操作全部回滾】
 * 示例：
 * let func = async function(conn) { ... };
 * await executeTransaction(func);
 * @param func 具體的數(shù)據(jù)庫操作異步方法（第一個參數(shù)必須為數(shù)據(jù)庫連接對象connection）
 * @returns {Promise.<*>} func方法執(zhí)行后的返回值
 */
async function executeTransaction(func) {
    const connection = await getConnection();
    await beginTransaction(connection);
    let result = null;
    try {
        result = await func(connection);
        await commit(connection);
        return result
    } catch (err) {
        console.error('事務(wù)執(zhí)行失敗，操作回滾');
        await rollback(connection);
        throw err;
    } finally {
        connection && connection.release && connection.release();
    }
}

四、增刪改查

增刪改查是處理數(shù)據(jù)的基本原子操作，將這些操作根據(jù)操作的特點進(jìn)行簡單的封裝。

• db/curd.js

/**
 * 查詢操作
 * @param connection 連接
 * @param sql SQL語句
 * @param val SQL參數(shù)
 * @returns {Promise} resolve查詢到的數(shù)據(jù)數(shù)組
 */
function query (connection, sql, val) {
    // console.info('sql執(zhí)行query操作：\n', sql, '\n', val);
    return new Promise((resolve, reject) => {
        connection.query(sql, val, (err, rows) => {
            if (err) {
                console.error('sql執(zhí)行失?。?, sql, '\n', val);
                reject(err);
            } else {
                let results = JSON.parse(JSON.stringify(rows));
                resolve(results);
            }
        });
    });
}
/**
 * 查詢單條數(shù)據(jù)操作
 * @param connection 連接
 * @param sql SQL語句
 * @param val SQL參數(shù)
 * @returns {Promise} resolve查詢到的數(shù)據(jù)對象
 */
function queryOne (connection, sql, val) {
    return new Promise((resolve, reject) => {
        query(connection, sql, val).then(
            results => {
                let result = results.length > 0 ? results[0] : null;
                resolve(result);
            },
            err => reject(err)
        )
    });
}
/**
 * 新增數(shù)據(jù)操作
 * @param connection 連接
 * @param sql SQL語句
 * @param val SQL參數(shù)
 * @param {boolean} skipId 跳過自動添加ID, false: 自動添加id，true: 不添加id
 * @returns {Promise} resolve 自動生成的id
 */
function insert (connection, sql, val, skipId) {
    let id = val.id;
    if (!id && !skipId) {
        id = uuid();
        val = {id, ...val};
    }
    return new Promise((resolve, reject) => {
        // console.info('sql執(zhí)行insert操作：\n', sql, '\n', val);
        connection.query(sql, val, (err, results) => {
            if (err) {
                console.error('sql執(zhí)行失敗！', sql, '\n', val);
                reject(err);
            } else {
                resolve(id);
            }
        });
    });
}
/**
 * 更新操作
 * @param connection 連接
 * @param sql SQL語句
 * @param val SQL參數(shù)
 * @returns {Promise} resolve 更新數(shù)據(jù)的行數(shù)
 */
function update (connection, sql, val) {
    // console.info('sql執(zhí)行update操作：\n', sql, '\n', val);
    return new Promise((resolve, reject) => {
        connection.query(sql, val, (err, results) => {
            if (err) {
                console.error('sql執(zhí)行失敗！', sql, '\n', val);
                reject(err);
            } else {
                resolve(results.affectedRows);
            }
        });
    });
}
/**
 * 刪除操作
 * @param connection 連接
 * @param sql SQL語句
 * @param val SQL參數(shù)
 * @returns {Promise} resolve 刪除數(shù)據(jù)的行數(shù)
 */
function del (connection, sql, val) {
    // console.info('sql執(zhí)行delete操作：\n', sql, '\n', val);
    return new Promise((resolve, reject) => {
        connection.query(sql, val, (err, results) => {
            if (err) {
                console.error('sql執(zhí)行失?。?, sql, '\n', val);
                reject(err);
            } else {
                // console.log('delete result', results);
                resolve(results.affectedRows);
            }
        });
    });
}

五、代碼分層

將代碼分層可以降低代碼的耦合度，提高可復(fù)用性、可維護(hù)性，這里將代碼分成了3層： Dao層、Service層和Controller層。

• DAO層：主要負(fù)責(zé)數(shù)據(jù)持久化工作；
• Service層：主要負(fù)責(zé)業(yè)務(wù)模塊的邏輯設(shè)計，此層的業(yè)務(wù)實現(xiàn)，可以調(diào)用DAO層的接口；
• Controller層：負(fù)責(zé)具體的業(yè)務(wù)模塊流程的控制，在此層可以調(diào)用Service層的接口。

1.DAO層

• dao/userDao.js

const { query, queryOne, update, insert, del } = require('../db/curd');
class UserDao {
    static async queryUserById (connection, id) {
        const sql = `SELECT user.id, user.account, user.name, user.email, user.phone,
                          user.birthday, user.enable, user.deleteFlag, user.creator,
                          user.createTime, user.updater, user.updateTime
                   FROM sys_user user
                   WHERE user.id = ?`;
        const user = await queryOne(connection, sql, id);
        return user;
    }
    …
}
module.exports = UserDao;

2.Service層

• service/userService.js
  簡單調(diào)用一個DAO層方法：

const { execute, executeTransaction } = require('../db/execute');
const UserDao = require('../dao/userDao');
class UserService {
    static async findUserById (id) {
        return await execute(UserDao.queryUserById, id);
}
…
}
module.exports = UserService;

對于復(fù)雜些的業(yè)務(wù)邏輯可以使用匿名函數(shù)來實現(xiàn)：

static async findUserWithRoles (id) {
    return await execute (async connection => {
        const user = await UserDao.queryUserById(connection, id);
        if (user) {
            user.roles = await RoleDao.queryRolesByUserId(connection, id);
        }
        return user;
    });
}

如果要執(zhí)行事務(wù)操作，則需要使用executeTransaction 方法：

static async updateUserRoleRelations (userId, roleIds) {
    return await executeTransaction(async connection => {
        const relations = await UserDao.queryUserRoleRelations(connection, userId);
        const oldRoleIds = relations.map(item => item.roleId);
        const newRoleIds = roleIds || [];
        // 新增的角色數(shù)組
        const addList = [];
        // 移除的角色數(shù)組
        const removeList = [];
        newRoleIds.forEach(roleId => {
            if (oldRoleIds.indexOf(roleId) === -1) {
                addList.push(roleId);
            }
        });
        oldRoleIds.forEach(roleId => {
            if (newRoleIds.indexOf(roleId) === -1) {
                removeList.push(roleId);
            }
        });
        if (addList.length > 0) {
            await UserDao.insertUserRoleRelations(connection, userId, addList);
        }
        if (removeList.length > 0) {
            await UserDao.deleteUserRoleRelations(connection, userId, removeList);
        }
    });
}

3.Controller層

• controler/userController.js

const UserService = require('../service/userService');
class UserControler {
   static async getUserById (ctx) {
        // 用戶ID
        const id = ctx.params.id;
        // 是否包含用戶角色信息，如果withRoles 為 "1" 表示需要包含角色信息
        const withRoles = ctx.query.withRoles;
        let user;
        if (withRoles === '1') {
            user = await UserService.findUserWithRoles(id);
        } else {
            user = await UserService.findUserById(id);
        }
        if (user) {
            ctx.body = user;
        } else {
            ctx.body = {
                code: 1004,
                msg: '用戶不存在!'
            }
        }
    }
    …
}
module.exports = UserControler;

此示例基于Koa框架，controller 層實現(xiàn)完成后需要添加路由：

const router = new KoaRouter();
const UserController = require('./controler/userControler');
// 獲取指定ID的用戶
router.get('/users/:id', UserController.getUserById);
// 獲取所有用戶
router.get('/users', UserControler.getUsers);

對于Koa框架如何使用，這里不再介紹，路由添加完畢后，啟動服務(wù)，即可使用這些接口，如果本地服務(wù)啟動的端口為3000，接口請求地址如下：

• http://localhost:3000/users/3571a123-0454-49b4-a2bc-8b30a37f0b14
• http://localhost:3000/users/3571a123-0454-49b4-a2bc-8b30a37f0b14?withRoles=1
• http://localhost:3000/users/

六、說明

本文介紹了mysql模塊的基本使用，對其進(jìn)行了簡單封裝，并提供了使用示例。除了使用mysql模塊來操作數(shù)據(jù)庫，也可以使用mysql2模塊，mysql2的基本用法與mysql一致，另外mysql2還支持Promise，使用起來更方便。本文相關(guān)的代碼已提交到GitHub以供參考，項目地址： https://github.com/liulinsp/node-server-typeorm-demo。

作者：劉琳