搭建node服务（二）：操做MySQL

时间 2020-06-04

标签搭建 node 服务 mysql 栏目 MySQL 繁體版

原文原文链接

为了进行复杂信息的存储和查询，服务端系统每每须要数据库操做。数据库分为关系型数据库和非关系型数据库，关系型数据库有MySQL、Oracle、SQL Server等，非关系型数据库有Redis（经常使用来作缓存）、MongoDB等。MySQL是目前很流行的数据库，本文将要介绍如何在node服务中进行MySQL数据库操做。

1、安装依赖

npm install mysql --save

或者node

yarn add mysql

2、创建链接

要想进行数据库操做就须要和数据库创建链接，而后经过链接进行数据库的操做。MySQL的数据库链接方式有如下几种：mysql

mysql.createConnection() 每次请求创建一个链接
mysql.createPool() 建立链接池，从链接池中获取链接
mysql.createPoolCluster() 建立链接池集群，链接池集群能够提供多个主机链接

mysqljs文档中推荐使用第一种方式：每次请求创建一个链接，可是因为频繁的创建、关闭数据库链接，会极大的下降系统的性能，因此我选择了使用链接池的方式，若是对性能有更高的要求，安装了MySQL 集群，能够选择使用链接池集群。git

1. 数据库配置

将数据库相关的配置添加到公用的配置文件中，方便项目的初始化。github

config.js

module.exports = {
    …
    // mysql数据库配置
    mysql: {
        // 主机
        host: 'localhost',
        // 端口
        port: 3306,
        // 用户名
        user: 'root',
        // 密码
        password: '123456',
        // 数据库名
        database: 'server-demo',
        // 链接池容许建立的最大链接数，默认值为10
        connectionLimit: 50,
        // 容许挂起的最大链接数,默认值为0,表明挂起的链接数无限制
        queueLimit: 0
    }
};

connectionLimit 和 queueLimit 是数据链接池特有的配置项。sql

connectionLimit 是指链接池容许建立的最大链接数，默认值为10。当获取链接时，若是链接池中有空闲的链接则直接返回一个空闲链接。若是全部链接都被占用，则判断链接池中的链接数是否达到了容许的最大数，若是未达到则建立新的链接，若是已达到则获取链接的请求挂起，等待其余请求完成操做后释放的链接。
queueLimit 是指容许挂起的最大链接数，默认值为0，表明挂起的链接数无限制。当链接池中容许建立的全部链接都被占用时，获取链接的请求挂起，等待可用的链接，全部挂起的请求造成一个队列，queueLimit则是指这个队列的最大长度。须要注意的是，当queueLimit为0时并不表示不容许挂起，而是表示对挂起的数目没有限制。

2. 建立链接池

db/pool.js

/**
 * 数据库链接池
 */
const mysql = require('mysql');
const config = require('../config');

// 建立数据库链接池
const pool = mysql.createPool(config.mysql);

pool.on('acquire', function (connection) {
    console.log(`获取数据库链接 [${connection.threadId}]`);
});
pool.on('connection', function (connection) {
    console.log(`建立数据库链接 [${connection.threadId}]`);
});
pool.on('enqueue', function () {
    console.log('正在等待可用数据库链接');
});
pool.on('release', function (connection) {
    console.log(`数据库链接 [${connection.threadId}] 已释放`);
});

module.exports = pool;

建立数据库链接池pool后，就能够经过pool获取数据库链接了，另外经过监听链接池的事件能够了解链接池中链接的使用状况。
若是将connectionLimit 设为2，queueLimit 设为0，当同时有5个请求获取数据库链接时，线程池的事件日志以下：数据库

正在等待可用数据库链接
正在等待可用数据库链接
正在等待可用数据库链接
建立数据库链接 [1011]
获取数据库链接 [1011]
数据库链接 [1011] 已释放
获取数据库链接 [1011]
建立数据库链接 [1012]
获取数据库链接 [1012]
数据库链接 [1011] 已释放
获取数据库链接 [1011]
数据库链接 [1012] 已释放
获取数据库链接 [1012]
数据库链接 [1011] 已释放
数据库链接 [1012] 已释放

因为线程池容许的最大链接数是2，5个请求中会有2个请求可以获得链接，另外3个请求挂起等待可用链接。因为建立数据库链接的代价比较大，线程池在建立链接时采用懒汉式，也就是，用到时才建立。先获得链接的请求在完成操做后释放链接，放回到链接池，而后挂起的请求从线程池取出空闲的链接进行操做。npm

3、执行操做

因为mysql 模块的接口都为回调方式的，为了操做方便简单地将接口封装为Promise，相关方法封装以下：数组

const pool = require('./pool');

// 获取链接
function getConnection () {
    return new Promise((resolve, reject) => {
        pool.getConnection((err, connection) => {
            if (err) {
                console.error('获取数据库链接失败！', err)
                reject(err);
            } else {
                resolve(connection);
            }
        });
    });
}

// 开始数据库事务
function beginTransaction (connection) {
    return new Promise((resolve, reject) => {
        connection.beginTransaction(err => {
            if (err) {
                reject(err);
            } else {
                resolve();
            }
        });
    });
}

// 提交数据库操做
function commit (connection) {
    return new Promise((resolve, reject) => {
        connection.commit(err => {
            if (err) {
                reject(err);
            } else {
                resolve();
            }
        });
    })
}

// 回滚数据库操做
function rollback (connection) {
    return new Promise((resolve, reject) => {
        connection.rollback(err => {
            if (err) {
                reject(err);
            } else {
                resolve();
            }
        });
    })
}

1. 执行普通操做

对于不须要使用事务的普通操做，获取数据库链接connection后，使用connection进行数据库操做，完成后释放链接到链接池，则执行完成一次操做。缓存

db/execute.js

/**
 * 执行数据库操做【适用于不须要事务的查询以及单条的增、删、改操做】
 * 示例：
 * let func = async function(conn, projectId, memberId) { ... };
 * await execute( func, projectId, memberId);
 * @param func 具体的数据库操做异步方法（第一个参数必须为数据库链接对象connection）
 * @param params func方法的参数（不包含第一个参数 connection）
 * @returns {Promise.<*>} func方法执行后的返回值
 */
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. 执行事务操做

对于不少业务都须要执行事务操做，例如：银行转帐，A帐户转帐给B帐户 100元，这个业务操做须要执行两步，从A帐户减去100元，而后给B帐户增长100元。两个子操做必须所有执行成功才能完成完整的业务操做，若是任意子操做执行失败就须要撤销以前的操做，进行回滚。框架

对于须要使用事务的操做，获取数据库链接connection后，首先须要调用connection.beginTransaction() 开始事务，而后使用connection进行多步操做，完成后执行connection.commit() 进行提交，则执行完成一次事务操做。若是在执行过程当中出现了异常，则执行connection.rollback() 进行回滚操做。

db/execute.js

/**
 * 执行数据库事务操做【适用于增、删、改多个操做的执行，若是中间数据操做出现异常则以前的数据库操做所有回滚】
 * 示例：
 * let func = async function(conn) { ... };
 * await executeTransaction(func);
 * @param func 具体的数据库操做异步方法（第一个参数必须为数据库链接对象connection）
 * @returns {Promise.<*>} func方法执行后的返回值
 */
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('事务执行失败，操做回滚');
        await rollback(connection);
        throw err;
    } finally {
        connection && connection.release && connection.release();
    }
}

4、增删改查

增删改查是处理数据的基本原子操做，将这些操做根据操做的特色进行简单的封装。

db/curd.js

/**
 * 查询操做
 * @param connection 链接
 * @param sql SQL语句
 * @param val SQL参数
 * @returns {Promise} resolve查询到的数据数组
 */
function query (connection, sql, val) {
    // console.info('sql执行query操做：\n', sql, '\n', val);
    return new Promise((resolve, reject) => {
        connection.query(sql, val, (err, rows) => {
            if (err) {
                console.error('sql执行失败！', sql, '\n', val);
                reject(err);
            } else {
                let results = JSON.parse(JSON.stringify(rows));
                resolve(results);
            }
        });
    });
}

/**
 * 查询单条数据操做
 * @param connection 链接
 * @param sql SQL语句
 * @param val SQL参数
 * @returns {Promise} resolve查询到的数据对象
 */
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)
        )
    });
}

/**
 * 新增数据操做
 * @param connection 链接
 * @param sql SQL语句
 * @param val SQL参数
 * @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执行insert操做：\n', sql, '\n', val);
        connection.query(sql, val, (err, results) => {
            if (err) {
                console.error('sql执行失败！', sql, '\n', val);
                reject(err);
            } else {
                resolve(id);
            }
        });
    });
}

/**
 * 更新操做
 * @param connection 链接
 * @param sql SQL语句
 * @param val SQL参数
 * @returns {Promise} resolve 更新数据的行数
 */
function update (connection, sql, val) {
    // console.info('sql执行update操做：\n', sql, '\n', val);
    return new Promise((resolve, reject) => {
        connection.query(sql, val, (err, results) => {
            if (err) {
                console.error('sql执行失败！', sql, '\n', val);
                reject(err);
            } else {
                resolve(results.affectedRows);
            }
        });
    });
}

/**
 * 删除操做
 * @param connection 链接
 * @param sql SQL语句
 * @param val SQL参数
 * @returns {Promise} resolve 删除数据的行数
 */
function del (connection, sql, val) {
    // console.info('sql执行delete操做：\n', sql, '\n', val);
    return new Promise((resolve, reject) => {
        connection.query(sql, val, (err, results) => {
            if (err) {
                console.error('sql执行失败！', sql, '\n', val);
                reject(err);
            } else {
                // console.log('delete result', results);
                resolve(results.affectedRows);
            }
        });
    });
}

5、代码分层

将代码分层能够下降代码的耦合度，提升可复用性、可维护性，这里将代码分红了3层：Dao层、Service层和Controller层。

DAO层：主要负责数据持久化工做；
Service层：主要负责业务模块的逻辑设计，此层的业务实现，能够调用DAO层的接口；
Controller层：负责具体的业务模块流程的控制，在此层能够调用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

简单调用一个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;

对于复杂些的业务逻辑可使用匿名函数来实现：

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;
    });
}

若是要执行事务操做，则须要使用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 || [];
        // 新增的角色数组
        const addList = [];
        // 移除的角色数组
        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 层实现完成后须要添加路由：

const router = new KoaRouter();
const UserController = require('./controler/userControler');

// 获取指定ID的用户
router.get('/users/:id', UserController.getUserById);

// 获取全部用户
router.get('/users', UserControler.getUsers);

对于Koa框架如何使用，这里再也不介绍，路由添加完毕后，启动服务，便可使用这些接口，若是本地服务启动的端口为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/

6、说明

本文介绍了mysql模块的基本使用，对其进行了简单封装，并提供了使用示例。除了使用mysql模块来操做数据库，也可使用mysql2模块，mysql2的基本用法与mysql一致，另外mysql2还支持Promise，使用起来更方便。本文相关的代码已提交到GitHub以供参考，项目地址：https://github.com/liulinsp/node-server-typeorm-demo。

做者：刘琳