nodejs实现redis ORM

电玩女神 2022-05-31 04:24 306阅读 0赞

nodejs实现redis ORM。即操作数据库的方式操作redis。

实现思路：

需要保存一条用户数据 name='test',age=22,sex=0

1.获取自增ID，自增ID=1

2.redis key=redis\_proxy\_user\_1，生成规则为前缀+表名+自增ID，保存为redis的hash数据类型，即：

hmset redis\_proxy\_user\_1 name "test" age 22 sex 0

3.生成字段-值的映射关系，保存为redis的set数据类型，key生成规则前缀前缀+表名+字段+值，值为自增ID，即：

sadd redis\_set\_proxy\_user\_name\_test 1;

sadd redis\_set\_proxy\_user\_age\_22 1;

sadd redis\_set\_proxy\_user\_sex\_0 1;

4.判断字段-值的类型，如果为数字类型，需要添加分数映射，作为区间判断时的查询，保存为redis的sort set数据类型，key生成规则前缀+表名+字段，值为自增ID，分数为字段-值，这样就可以通过分数的区间进行查询，这里age和sex是数字类型，即：

zadd redis\_sort\_set\_user\_age 22 1;

zadd redis\_sort\_set\_user\_sex 0 1;

添加数据就完成了。

查询的思路：

需要查询name='test'的用户

1.组装redis key，查询set是否存在，key=redis\_set\_proxy\_user\_name\_test，即：

smembers redis\_set\_proxy\_user\_name\_test

找到\[1\]

2.根据上面查询结果，找到的自增ID，组装redis key，查询redis hash查询数据，key的生成规则生成规则为前缀+表名+自增ID，key=redis\_set\_proxy\_user\_name\_test，即：

hmget redis\_set\_proxy\_user\_name\_test name age sex

查询结束。

如果是需要查询age>=22的用户

1.查询sort set，判断分数在22到正无穷的所有成员，组装redis key，生成规则前缀+表名+字段，key=redis\_sort\_set\_user\_age，即：

zrangebyscore redis\_sort\_set\_user\_age 22 +inf

找到\[1\]

hmget redis\_set\_proxy\_user\_name\_test name age sex

查询结束。

代码实现了插入数据时可设置添加查询索引字段，多条件查询，分页，排序，占位符，修改，删除。

功能函数类redisDB.js：

'use strict';
    
    const F = require('../common/function');
    const _ = require('underscore');
    _.str = require('underscore.string');
    
    module.exports = function (app, commonManager) {
    
        let managerMap = commonManager.mgr_map;
        let that = this;
    
        // redis key 前缀 - hash
        this.key_prefix_hash = 'redis_db_key_hash_%s_%s'; // table increment
    
        // redis key 前缀 - set
        this.key_prefix_set = 'redis_db_key_set_%s_%s_%s'; // table field value
    
        // redis hash自增key
        this.key_prefix_has_inc = 'redis_db_key_hash_inc';
    
        // redis 临时区间结果集自增key
        this.key_prefix_tmp_set_inc = 'redis_db_key_tmp_set_inc';
    
        // redis sort set key 前缀
        this.key_prefix_sort_set = 'redis_db_key_sort_set_%s_%s'; // table field
    
        this.getKeyHash = function (table, increment) {
            return _.str.vsprintf(this.key_prefix_hash, [table, increment]);
        }
    
        this.getKeySet = function (table, field, value) {
            return _.str.vsprintf(this.key_prefix_set, [table, field, value]);
        }
    
        this.getKeySortSet = function (table, field) {
            return _.str.vsprintf(this.key_prefix_sort_set, [table, field]);
        }
    
        /**
         * 获取最优过期时间
         * @param key
         * @param expire
         * @returns {int}
         */
        this.getBestExpire = function* (key, expire) {
            let key_expire = yield managerMap.redis.ttl(key);
            if(key_expire < expire) {
                key_expire = expire;
            }else if(expire == -1) {
                key_expire = expire;
            }
            return key_expire;
        }
    
        /**
         * 获取数组交集
         * @param arrays
         * @returns {array}
         */
        this.intersection = function (arrays) {
            if(arrays.length == 0) return new Array();
            if(arrays.length == 1) return arrays[0];
    
            // 获取长度最小数组的索引
            let min_index = -1;
            for (var i = 0; i < arrays.length; i++) {
                if(min_index == -1 || arrays[min_index].length > arrays[i].length) {
                    min_index = i;
                }
            }
            let intersection_array = new Array();
            for (var i = 0; i < arrays[min_index].length; i++) {
                let is_in = true; // 是否交集 默认true
                for (var j = 0; j < arrays.length; j++) {
                    if(j == min_index) continue;
                    if(arrays[j].indexOf(arrays[min_index][i]) == -1) {
                        is_in = false;
                        break;
                    }
                }
                if(is_in == true && intersection_array.indexOf(arrays[min_index][i]) == -1) {
                    intersection_array.push(arrays[min_index][i]);
                }
            }
            return intersection_array;
        }
    
        /**
         * 判断是否为数字类型
         * @param str
         * @returns {bool}
         */
        this.checkIsNumber = function (str) {
            if(F.isNull(str)) return false;
            let number_map = '1234567890';
            let is_number = true;
            for (var i = 0; i < str.length; i++) {
                if(number_map.indexOf(str.charAt(i)) == -1) {
                    is_number = false;
                    break;
                }
            }
            return is_number;
        }
    
        /**
         * 查询结果集
         * @param table
         * @param option
         * option包括：
         * fields 字符串 例如："name,age" 不能空
         * where 字符串 例如"name = ? and age > ?" 不能空 只支持 = > < >= <=
         * values 数组 例如['fdl', 2]
         * @returns {}
         */
        this.queryResultFromRedisDB = function* (table, option) {
            let key_array = -1; // 等值操作符key集合
            let res_array = -1; // 等值查询结果集
            let res_between_array = -1; // 区间操作符结果集
            // 解析where
            let where_array = option.where.split('and');
            for (var i = 0; i < where_array.length; i++) {
                let split_str;
                if(where_array[i].indexOf('>=') != -1) {
                    split_str = '>=';
                }else if(where_array[i].indexOf('<=') != -1) {
                    split_str = '<=';
                }else if(where_array[i].indexOf('>') != -1){
                    split_str = '>';
                }else if(where_array[i].indexOf('<') != -1) {
                    split_str = '<';
                }else if(where_array[i].indexOf('=') != -1) {
                    split_str = '=';
                }else {
                    F.throwErr('where sql err.');
                }
                let param_array = where_array[i].split(split_str);
                let field = _.str.trim(param_array[0], ' ');
                let value = _.str.trim(param_array[1], [' ', '"', '\'']);
                if(value == '?') {
                    value = option.values.shift();
                }
                let option_min;
                let option_max;
                switch(split_str) {
                    case '=':
                        if(key_array == -1) {
                            key_array = new Array();
                        }
                        key_array.push(this.getKeySet(table, field, value));
                        break;
                    case '>':
                        option_min = '(' + value;
                        option_max = '+inf';
                        break;
                    case '<':
                        option_min = '-inf';
                        option_max = '(' + value;
                        break;
                    case '>=':
                        option_min = value;
                        option_max = '+inf';
                        break;
                    case '<=':
                        option_min = '-inf';
                        option_max = value;
                        break;
                }
                if(split_str != '=') {
                    if(res_between_array == -1) {
                        res_between_array = new Array();
                    }
                    res_between_array.push(yield managerMap.redis.zrangebyscore(this.getKeySortSet(table, field), option_min, option_max));
                }
            }
            let query_res = new Array();
            if(key_array != -1) {
                res_array = yield managerMap.redis.sinter(key_array);
                query_res.push(res_array);
            }
            if(res_between_array != -1) {
                if(query_res.length == 0) {
                    query_res = res_between_array;
                }else {
                    query_res.push.apply(query_res, res_between_array);
                }
            }
            // 获取等值查询与区间查询结果集的交集
            return this.intersection(query_res);
        }
    
        /**
         * 插入
         * @param table
         * @param option
         * option包括：
         * data 数据对象
         * expire 过期时间 -1为永不过期
         * index 字符串 创建索引的字段 例如：'name,age'
         * @returns {}
         */
        this.insertRedisDB = function* (table, option) {
            if(F.isNull(table) || F.isNull(option) || F.isNull(option.data) || F.isNull(option.expire)) F.throwErr('params err.');
    
            // 添加 hash映射 set映射
            let key_hash_inc = yield managerMap.redis.hincrby(this.key_prefix_has_inc, table);
            let key_hash = this.getKeyHash(table, key_hash_inc);
            let fields_data = new Array();
            let index_fields = -1;
            if(!F.isNull(option.index)) {
                index_fields = option.index.split(',');
                for (var i = 0; i < index_fields.length; i++) {
                    index_fields[i] = _.str.trim(index_fields[i], ' ');
                }
            }
            for(let field in option.data) {
                fields_data.push(field);
                fields_data.push(option.data[field]);
    
                // 判断是否添加索引
                if(index_fields != -1 && index_fields.indexOf(field) == -1) continue;
    
                // 添加 set映射
                let key_set = this.getKeySet(table, field, option.data[field]);
                let expire_set = yield this.getBestExpire(key_set, option.expire);
                yield managerMap.redis.sadd(key_set, key_hash_inc, expire_set);
    
                // 判断是否number类型 添加number类型数据 sort set映射
                if(typeof(option.data[field]) == 'number') {
                    let ket_sort_set = this.getKeySortSet(table, field);
                    let expire_sort_set = yield this.getBestExpire(ket_sort_set, option.expire);
                    yield managerMap.redis.zadd(ket_sort_set, key_hash_inc, option.data[field], expire_sort_set);
                }
            }
            let expire_hash = yield this.getBestExpire(key_hash, option.expire);
            yield managerMap.redis.hmset(key_hash, fields_data, expire_hash);
            return {key_hash:key_hash};
        };
    
        /**
         * 查询
         * @param table
         * @param option
         * option包括：
         * fields 字符串 例如："name,age" 不能空
         * where 字符串 例如"name = ? and age > ?" 不能空 只支持 = > < >= <=
         * values 数组 例如['fdl', 2]
         * order 字符串 为空默认desc 排序字段必须在fields出现
         * limit 当获取一条数据时赋值1，为1时返回对象
         * @returns {*}
         */
        this.queryRedisDB = function* (table, option) {
            if(F.isNull(table) || F.isNull(option) || F.isNull(option.fields) || F.isNull(option.where)) F.throwErr('params err.');
    
            let query_res = yield this.queryResultFromRedisDB(table, option);
            if(F.isNull(query_res)) {
                if(F.isNull(option.limit) || option.limit != 1) {
                    return new Array();
                }else {
                    return new Object();
                }
            }
            let fields = option.fields.split(',');
            for (var i = 0; i < fields.length; i++) {
                fields[i] = _.str.trim(fields[i], ' ');
            }
            let list = new Array();
            for (var i = 0; i < query_res.length; i++) {
                let key_hash = this.getKeyHash(table, query_res[i]);
                let has_set = yield managerMap.redis.exists(key_hash);
                if(has_set == false) continue;
                let data = yield managerMap.redis.hmget(key_hash, fields);
                let item = {};
                for (var j = 0; j < fields.length; j++) {
                    item[fields[j]] = data[j];
                }
                list.push(item);
            }
            if(!F.isNull(option.order)) {
                let order_array = option.order.split(' ');
                let field = _.str.trim(order_array[0], ' ');
                let order = _.str.trim(order_array[1], ' ');
                list.sort(function(objectA, objectB) {
                    if(order == 'asc') {
                        if(that.checkIsNumber(objectA[field]) == true) {
                            return parseInt(objectA[field]) > parseInt(objectB[field]);
                        }else {
                            return objectA[field] > objectB[field];
                        }
                    }else {
                        if(that.checkIsNumber(objectA[field]) == true) {
                            return parseInt(objectB[field]) > parseInt(objectA[field]);
                        }else {
                            return objectB[field] > objectA[field];
                        }
                    }
                });
            }
            if(F.isNull(option.limit)) {
                return list;
            }else if(option.limit == 1){
                return list[0];
            }else {
                let limit_array = option.limit.split(',');
                let offset = parseInt(_.str.trim(limit_array[0], ' '));
                let limit = offset + parseInt(_.str.trim(limit_array[1], ' '));
                return list.slice(offset, limit);
            }
        };
    
        /**
         * 更新
         * @param table
         * @param option
         * option包括：
         * where 字符串 例如"name = ? and age > ?" 不能空 只支持 = > < >= <=
         * values 数组 例如['fdl', 2]
         * data 数据对象
         * expire 过期时间 -1为永不过期
         * index 字符串 创建索引的字段 例如：'name,age'
         * @returns {}
         */
        this.updateRedisDB = function* (table, option) {
            if(F.isNull(table) || F.isNull(option) || F.isNull(option.where) || F.isNull(option.data) || F.isNull(option.expire)) {
                F.throwErr('params err.');
            }
    
            let query_res = yield this.queryResultFromRedisDB(table, option);
            if(F.isNull(query_res)) return true;
            let del_key_array = new Array();
            let del_key_record = new Object();
            let update_data = new Array();
            for (var i = 0; i < query_res.length; i++) {
                let key_hash = this.getKeyHash(table, query_res[i]);
                let fields_keys = yield managerMap.redis.hkeys(key_hash);
                if(F.isNull(fields_keys)) continue;
                let fields_data = yield managerMap.redis.hmget(key_hash, fields_keys);
                if(F.isNull(fields_data)) continue;
                // 查询旧数据字段键值映射 添加到删除数据
                let insert_item = new Object();
                del_key_array.push(key_hash);
                for (var j = 0; j < fields_keys.length; j++) {
                    if(option.data.hasOwnProperty(fields_keys[j])) { // 只删除更新字段的映射key
                        // 查询删除key
                        let key_set = this.getKeySet(table, fields_keys[j], fields_data[j]);
                        if(F.isNull(del_key_record[key_set])) {
                            del_key_array.push(key_set);
                            del_key_record[key_set] = 1;
                        }
                        if(typeof(option.data[fields_keys[j]]) == 'number') {
                            // 删除区间查询映射key
                            let key_sort_set = this.getKeySortSet(table, fields_keys[j]);
                            if(F.isNull(del_key_record[key_sort_set])) {
                                yield managerMap.redis.zrem(this.getKeySortSet(table, fields_keys[j]), query_res[i]);
                                del_key_record[key_sort_set] = 1;
                            }
                        }
                        insert_item[fields_keys[j]] = option.data[fields_keys[j]];
                    }else {
                        insert_item[fields_keys[j]] = fields_data[j];
                    }
                }
                // 检测更新字段中是否存在新增字段
                for (var field in option.data) {
                    if(insert_item.hasOwnProperty(field) == false) {
                        insert_item[field] = option.data[field];
                    }
                }
                update_data.push(insert_item);
            }
            yield managerMap.redis.del(del_key_array);
            
            // 添加更新的数据
            for (var i = 0; i < update_data.length; i++) {
                yield this.insertRedisDB(table, {
                    'data': update_data[i],
                    'expire': option.expire
                });
            }
    
            return {del_key_array:del_key_array};
        }
    
        /**
         * 删除
         * @param table
         * @param option
         * option包括：
         * where 字符串 例如"name = ? and age > ?" 不能空 只支持 = > < >= <=
         * @returns {}
         */
        this.deleteRedisDB = function* (table, option) {
            if(F.isNull(table) || F.isNull(option) || F.isNull(option.where)) F.throwErr('params err.');
    
            let query_res = yield this.queryResultFromRedisDB(table, option);
            if(F.isNull(query_res)) return true;
            let del_key_array = new Array();
            let del_key_record = new Object();
            for (var i = 0; i < query_res.length; i++) {
                let key_hash = this.getKeyHash(table, query_res[i]);
                let fields_keys = yield managerMap.redis.hkeys(key_hash);
                if(F.isNull(fields_keys)) continue;
                let fields_data = yield managerMap.redis.hmget(key_hash, fields_keys);
                if(F.isNull(fields_data)) continue;
                // 查询数据字段键值映射 添加到删除数据
                del_key_array.push(key_hash);
                for (var j = 0; j < fields_keys.length; j++) {
                    // 查询删除key
                    let key_set = this.getKeySet(table, fields_keys[j], fields_data[j]);
                    if(F.isNull(del_key_record[key_set])) {
                        del_key_array.push(key_set);
                        del_key_record[key_set] = 1;
                    }
                    if(this.checkIsNumber(fields_data[j]) == true) {
                        // 删除区间查询映射key
                        let key_sort_set = this.getKeySortSet(table, fields_keys[j]);
                        if(F.isNull(del_key_record[key_sort_set])) {
                            yield managerMap.redis.zrem(this.getKeySortSet(table, fields_keys[j]), query_res[i]);
                            del_key_record[key_sort_set] = 1;
                        }
                    }
                }
            }
            yield managerMap.redis.del(del_key_array);
    
            return {del_key_array:del_key_array};
        }
    
    };

其中F.isNull函数是判断字符串，数据库，对象是否为空，managerMap.redis调用的方法是redis原生方法再封装一层，managerMap.redis.js部分函数代码：

this.hincrby = function* (key, field, increment = 1) {
        return yield redisLib.redisCo.HINCRBY(key, field, increment);
    }
    
    this.ttl = function* (key) {
        return yield redisLib.redisCo.ttl(key);
    }
    
    this.hmset = function* (key, data, expire = null) {
        let insert_data = [key];
        insert_data.push.apply(insert_data, data);
        let res = yield redisLib.redisCo.hmset(insert_data);
        if(!F.isNull(expire)) {
            res = yield redisLib.redisCo.expire(key, expire);
        }
        return res;
    };
    
    this.hmget = function* (key, fields) {
        let fields_array = [key];
        fields_array.push.apply(fields_array, fields);
        return yield redisLib.redisCo.hmget(fields_array);
    };
    
    this.hkeys = function* (key) {
        return yield redisLib.redisCo.hkeys(key);
    };
    
    this.hgetall = function* (key) {
        return yield redisLib.redisCo.hgetall(key);
    }
    
    this.sadd = function* (key, value, expire = null) {
        let fields_array = [key];
        if(Array.isArray(value)) {
            fields_array.push.apply(fields_array, value);
        }else {
            fields_array.push(value);
        }
        let res = yield redisLib.redisCo.sadd(fields_array);
        if(!F.isNull(expire)) {
            res = yield redisLib.redisCo.expire(key, expire);
        }
        return res;
    }
    
    this.sinter = function* (keys) {
        return yield redisLib.redisCo.sinter(keys);
    }
    
    this.exists = function* (key) {
        return yield redisLib.redisCo.exists(key);
    }
    
    this.zadd = function* (key, value, score, expire = null) {
        let res = yield redisLib.redisCo.ZADD([key, score, value]);
        if(!F.isNull(expire)) {
            res = yield redisLib.redisCo.expire(key, expire);
        }
        return res;
    }
    
    this.zrangebyscore = function* (key, min, max) {
        return yield redisLib.redisCo.zrangebyscore([key, min, max]);
    }
    
    this.zrem = function* (key, value) {
        let fields_array = [key];
        if(Array.isArray(value)) {
            fields_array.push.apply(fields_array, value);
        }else {
            fields_array.push(value);
        }
        return yield redisLib.redisCo.zrem(fields_array);
    }

使用例子：

// 插入数据
    yield mgr_map.redisDB.insertRedisDB('user', {data:{name:'fdl',age:22,sex:0,score:21}, expire:5080, index:'name, age'});
    yield mgr_map.redisDB.insertRedisDB('user', {data:{name:'fdsl',age:22,sex:1,score:64}, expire:5080, index:'name, age'});
    yield mgr_map.redisDB.insertRedisDB('user', {data:{name:'fd2l',age:28,sex:0,score:58}, expire:5080, index:'name, age'});
    yield mgr_map.redisDB.insertRedisDB('user', {data:{name:'fdl',age:23,sex:0,score:22}, expire:5080, index:'name, age'});
    yield mgr_map.redisDB.insertRedisDB('user', {data:{name:'fd8l',age:12,sex:1,score:24}, expire:5080, index:'name, age'});
    
    // 查询
    res = yield mgr_map.redisDB.queryRedisDB('user', {
        where:"name=? and age > 10 and score <= ?",
        values:['fdl', 22],
        fields:'name , age,score',
        limit:'0,100',
        order:'age asc'
    });
    
    // 更新
    yield mgr_map.redisDB.updateRedisDB('user', {
        where:"name=?",
        values:['fd2l'],
        data:{
            'name':'测试',
            'age':100,
            'contry':'china'
        },
        expire:5080,
        index:'name, age'
    });
    
    // 删除
    res = yield mgr_map.redisDB.deleteRedisDB('user', {
        where:"name=? and age > 10",
        values:['fdl']
    });