通过UDF使mysql主动刷新redis缓存

UDF是mysql的一个拓展接口，UDF（Userdefined function）用户自定义函数。在什么地方使用这个功能呢，试想有如下场景：

你的网站使用mysql作为最终数据落地的存储引擎，而redis作为缓存以减小查询请求穿透到mysql的数量，可以极大的降低数据库性能瓶颈带来的整个网站对外服务的卡顿、不可用等情况。这种方式的架构，当有查询请求的时候，我们可以在业务逻辑层控制，先从缓存中查询，无命中的情况下，再到数据库中查询，同时缓存到redis中；当有修改请求的时候，我们可以先修改数据库，然后删除或更新缓存。

以上方式是我们业务量不大，开发简单的，少横向扩展的情况下做的。当开发复杂度随着业务量并发增大，呈现横向扩展和垂直方向上螺旋迭代上升趋势的时候，逻辑复杂度直线上升。还采用在业务逻辑层做缓存控制将变得很复杂，运维上也容易出错。

这个时候，如果能将缓存逻辑和业务逻辑分离，缓存层对业务逻辑提供服务透明，业务逻辑不用关心缓存逻辑，缓存逻辑也不用随业务变化而改动，互相做自己的事情，这样高内聚低耦合可以极大的增加扩展性和健壮性，也是我们做架构应该努力发展的方向。

那么具体来说，我的这里要做的事情，其实就是把缓存更新的逻辑，放到mysql中去做。写一个trigger触发器监控insert/update/delete这些修改数据的操作，当有修改操作的时候，调用对应的自定义UDF函数来远程回写redis缓存，而我们在业务逻辑层则只管更新数据就行了，缓存更新的操作都放给以上的缓存层逻辑来完成。

当然，以上操作也可以反方向来，先写redis，然后由redis同步到mysql去。两种方式各有利弊，看你的具体场景如何选择了，这里不讨论。我们此处的目的是使用mysql的udf函数更新数据到redis中。

开发环境

操作系统:centos 6.4 server，内核2.6.32-358.18.1.el6.x86_64

编译器：gcc 4.4.7

数据库：mysql server 5.1.73

mysql服务器之前已经安装了，现在我们要安装mysql开发包

yum install mysql-devel -y

一、UDF函数入门

首先学习下UDF函数的使用方法。我们自定义一个函数文件，test_add.cpp如下

#include <mysql.h>

extern "C" long long testadd(UDF_INIT *initid, UDF_ARGS *args, char *is_null, char *error)
{
    int a = *((long long *)args->args[0]);
    int b = *((long long *)args->args[1]);
    return a + b;
}

extern "C" my_bool testadd_init(UDF_INIT *initid, UDF_ARGS *args, char *message)
{
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13

这是c++代码，编译

[root@centos6 ~]# g++ -shared -fPIC -I /usr/include/mysql -o test_add.so test_add.cpp
1

拷贝到mysql插件目录下，要以root身份

[root@centos6 ~]# cp test_add.so /usr/lib64/mysql/plugin/
1

如果你不知道插件的路径，执行

mysql> show variables like '%plugin%';
+---------------+-------------------------+
| Variable_name | Value                   |
+---------------+-------------------------+
| plugin_dir    | /usr/lib64/mysql/plugin |
+---------------+-------------------------+
1 row in set (0.00 sec)
1
2
3
4
5
6
7

登录mysql，创建函数关联

mysql> create function testadd returns integer soname 'test_add.so';
Query OK, 0 rows affected (0.00 sec)
1
2

至此，UDF就搞定了，接下来测试

mysql> select testadd(1,2);
+--------------+
| testadd(1,2) |
+--------------+
|            3 |
+--------------+
1 row in set (0.00 sec)
1
2
3
4
5
6
7

可以看到testadd函数生效了，输出结果为 1 + 2 = 3.

如果要删除UDF函数

mysql> drop function testadd;
Query OK, 0 rows affected (0.01 sec)
1
2

然后删除插件目录下的.so文件

[root@centos6 ~]# rm -f /usr/lib64/mysql/plugin/test_add.so
1

二、结合redis做缓存更新

1.在DUF中访问redis

更新redis的原理其实和上面示例一样的，只是要在UDF中调用redis的api函数。在此之前，请先下载api源码
git clone https://github.com/mrpi/redis-cplusplus-client
这是redis官网给出的c++访问redis的客户端api代码，依赖于boost，先安装

[root@centos6 ~]# yum install boost boost-devel
1

然后

[root@centos6 ~]# cd redis-cplusplus-client
1

当要调用该库的时候，把如下几个文件拷贝过去一起编译就可以了
redisclient.h、anet.h、fmacros.h、anet.c。接下来看我的源码test.cpp

#include <stdio.h>
#include <mysql.h>
#include "redisclient.h"
using namespace boost;
using namespace std;

static redis::client *_client = NULL;

// 初始化连接
void check_connection()
{
    if(NULL == _client){
        const char* c_host = getenv("REDIS_HOST"); // 获取操作系统变量
        string host = "localhost";
        if(c_host)
            host = c_host;
        _client = new redis::client(host);
    }
}

// 调用redis的hset命令
extern "C" char *redis_hset(UDF_INIT *initid, UDF_ARGS *args, char *result, unsigned long *length, char *is_null, char *error)
{
    try{
        check_connection();
        if(!(args->args && args->args[0] && args->args[1] && args->args[2])){
            *is_null = 1;
            *length = 8;
            snprintf(result, 8, "is null");
            return result;
        }
        if(_client->hset(args->args[0], args->args[1], args->args[2])){
            *length = 2;
            snprintf(result, 2, "0");
            return result;
        } else {
            *error = 1;
            *length = 5;
            snprintf(result, 6, "error");
            return result;
        }
    } catch (const redis::redis_error & e){
        *length = ((std::string)e).length() + 1;
        snprintf(result, *length, "%s", (char*)e.what());
        return result;
    }
}

/*资源分配*/
extern "C" my_bool redis_hset_init(UDF_INIT *initid, UDF_ARGS *args, char *message)
{
    if (3 != args->arg_count  || args->arg_type[0] != STRING_RESULT  || args->arg_type[1] != STRING_RESULT  || args->arg_type[2] != STRING_RESULT){ // hset(key, field, value) 需要三个参数
        strncpy(message, "please input 3 args and must be string, such as: hset('key', 'feild', 'value');", MYSQL_ERRMSG_SIZE);
        return -1;
    }
    args->arg_type[0] = STRING_RESULT;
    args->arg_type[1] = STRING_RESULT;
    args->arg_type[2] = STRING_RESULT;

    initid->ptr       = NULL;
    return 0;
}

/*
// 测试
int main(){
    char is_null;
    char message[128] = {0};
    char result[128] = {0};
    unsigned long length = 0;
    UDF_ARGS args;
    UDF_INIT initid;
    args.arg_count = 3;
    args.args = new char*[3];
    args.args[0] = new char[16];
    args.args[1] = new char[16];
    args.args[2] = new char[16];
    args.arg_type = new Item_result[3];
    strcpy(args.args[0], "mykey");
    strcpy(args.args[1], "myfeild");
    strcpy(args.args[2], "myvalue");
    redis_hset_init(&initid, &args, message);
    redis_hset(&initid, &args, result, &length, &is_null, message);
    printf("%s\n", result);
    if(args.arg_type)
        delete args.arg_type;
    if(args.args)
        delete args.args;
    return 0;
}
*/
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

请提前启动redis服务器，我是在本机启动的，所以地址就是localhost，端口不写默认就是6379。编译和拷贝

[root@centos6 ~]# g++ -shared -fPIC -I /usr/include/mysql -lboost_serialization -o myredis.so anet.c test.cpp
[root@centos6 ~]# rm -f /usr/lib64/mysql/plugin/myredis.so && cp myredis.so /usr/lib64/mysql/plugin/ && chmod 777 /usr/lib64/mysql/plugin/myredis.so
1
2

登录mysql客户端，执行

mysql> DROP FUNCTION IF EXISTS `redis_hset`; create function redis_hset returns string soname 'myredis.so';
Query OK, 0 rows affected (0.01 sec)

Query OK, 0 rows affected (0.00 sec)

mysql> select * from mysql.func;
+------------+-----+------------+----------+
| name       | ret | dl         | type     |
+------------+-----+------------+----------+
| redis_hset |   0 | myredis.so | function |
+------------+-----+------------+----------+
1
2
3
4
5
6
7
8
9
10
11

安装完成，现在测试。在mysql中执行

mysql> select redis_hset('Jack', 'id', '101');
+---------------------------------+
| redis_hset('User', 'id', '101') |
+---------------------------------+
| 0                               |
+---------------------------------+
1
2
3
4
5
6

返回字符”0”，说明调用成功。到redis上看看结果

127.0.0.1:6379> hgetall User
1) "id"
2) "101"
1
2
3

数据正确，mysql的DUF和redis通信成功，并且正确的修改数据。

2.用触发器实现动态更新redis缓存

这一步的思路，就是在mysql中创建一个触发器，监听表的insert/update/delete等操作，有数据更新的时候调用上一步的UDF函数刷新信息到redis缓存中去。

先准备数据库表和触发器

CREATE TABLE `tb_user` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `username` varchar(255) CHARACTER SET utf8mb4 DEFAULT NULL,
  PRIMARY KEY (`id`)
);
1
2
3
4
5

触发器

delimiter $
create trigger tg_user
after insert on tb_user
for each row 
begin
    set @id = (select redis_hset(CONCAT('user_', new.id), 'id', CAST(new.id AS CHAR)));
    set @username = (select redis_hset(CONCAT('user_', new.id), 'username', new.username));
end $
1
2
3
4
5
6
7
8

这里触发器监控的是insert，当插入新数据的时候，会把用户id和用户名刷新到redis缓存中去。也可以针对update和delete做触发，篇幅有限，就不列出来了。

测试

mysql> insert into test.tb_user(`username`) values('Jack');
Query OK, 1 row affected (0.01 sec)

mysql> insert into test.tb_user(`username`) values('Lucy');
Query OK, 1 row affected (0.01 sec)
1
2
3
4
5

插入了两条数据，查看redis的反应

127.0.0.1:6379> keys user_*
1) "user_1"
2) "user_2"

127.0.0.1:6379> hgetall user_1
1) "id"
2) "1"
3) "username"
4) "Jack"

127.0.0.1:6379> hgetall user_2
1) "id"
2) "2"
3) "username"
4) "Lucy"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

可以看到user_1和user_2两条key都插入了，每条key的内容也和我们在mysql中插入的一致。至此，我们的目的，“数据库更新，自动刷新到缓存”就实现了，妥妥的！

总结

以上的思路我已经讲的很清楚了，我只写了一个简单的示例，使用的是redis的hset命令。这里我们找一个现成的同步mysql到redis的工具，很全面。从 http://pan.baidu.com/s/1qW9DHYc 下载mysql_udf_redis.tar.bz2，基本满足常用的redis操作命令。