用Redis做爲Mysql數據庫的緩存【轉】

用Redis做Mysql數據庫緩存，必須解決2個問題。首先，應該肯定用何種數據結構存儲來自Mysql的數據；在肯定數據結構以後，還要考慮用什麼標識做爲該數據結構的鍵。
直觀上看，Mysql中的數據都是按表存儲的；更微觀地看，這些表都是按行存儲的。每執行一次select查詢，Mysql都會返回一個結果集，這個結果集由若干行組成。因此，一個天然而然的想法就是在Redis中找到一種對應於Mysql行的數據結構。Redis中提供了五種基本數據結構，即字符串（string）、列表（list）、哈希（hash）、集合（set）和有序集合（sorted set）。通過調研，發現適合存儲行的數據結構有兩種，即string和hash。
要把Mysql的行數據存入string，首先須要對行數據進行格式化。事實上，結果集的每一行均可以看作若干由字段名和其對應值組成的鍵值對集合。這種鍵值對結構很容易讓咱們想起Json格式。所以，這裏選用Json格式做爲結果集每一行的格式化模板。根據這一想法，咱們能夠實現將結果集格式化爲若干Json對象，並將Json對象轉化爲字符串存入Redis的代碼：
view sourceprint?
01.
// 該函數把結果集中的每一行轉換爲一個Json格式的字符串並存入Redis的STRING結構中，
02.
// STRING鍵應該包含結果集標識符和STRING編號，形式如「cache.string:123456:1」
03.
string Cache2String(sql::Connection *mysql_connection,
04.
redisContext *redis_connection,
05.
sql::ResultSet *resultset,
06.
const string &resultset_id, int ttl) {
07.
if (resultset->rowsCount() == 0) {
08.
throw runtime_error("FAILURE - no rows");
09.
}
10.
// STRING鍵的前綴，包含告終果集的標識符
11.
string prefix("cache.string:" + resultset_id + ":");
12.
unsigned int num_row = 1;  // STRING編號，附加於STRING鍵的末尾，從1開始
13.
sql::ResultSetMetaData *meta = resultset->getMetaData();
14.
unsigned int num_col = meta->getColumnCount();
15.
// 將結果集中全部行對應的全部STRING鍵存入該SET，SET鍵包含告終果集的標識符
16.
string redis_row_set_key("resultset.string:" + resultset_id);
17.
redisReply *reply;
18.
string ttlstr;
19.
stringstream ttlstream;
20.
ttlstream << ttl;
21.
ttlstr = ttlstream.str();
22.
resultset->beforeFirst();
23.
// 將結果集中的每一行轉爲Json格式的字符串，將這些Json字符串存入STRING，
24.
// 每一個STRING對應結果集中的一行
25.
while (resultset->next()) {
26.
string redis_row_key;  // STRING鍵名，由前綴和STRING編號組成
27.
stringstream keystream;
28.
keystream << prefix << num_row;
29.
redis_row_key = keystream.str();
30.
Json::Value row;
31.
for (int i = 1; i <= num_col; ++i) {
32.
string col_label = meta->getColumnLabel(i);
33.
string col_value = resultset->getString(col_label);
34.
row[col_label] = col_value;
35.
}
36.
Json::FastWriter writer;
37.
string redis_row_value = writer.write(row);
38.
// 將STRING鍵及Json格式的對應值對存入Redis
39.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
40.
"SET %s %s",
41.
redis_row_key.c_str(),
42.
redis_row_value.c_str()));
43.
freeReplyObject(reply);
44.
// 將STRING鍵加入SET中
45.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
46.
"SADD %s %s",
47.
redis_row_set_key.c_str(),
48.
redis_row_key.c_str()));
49.
freeReplyObject(reply);
50.
// 設置STRING的過時時間
51.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
52.
"EXPIRE %s %s",
53.
redis_row_key.c_str(),
54.
ttlstr.c_str()));
55.
freeReplyObject(reply);
56.
++num_row;
57.
}
58.
// 設置SET的過時時間
59.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
60.
"EXPIRE %s %s",
61.
redis_row_set_key.c_str(),
62.
ttlstr.c_str()));
63.
freeReplyObject(reply);
64.
return redis_row_set_key;  // 返回SET鍵，以便於其餘函數獲取該SET中的內容
65.
}


要把Mysql的行數據存入hash，過程要比把數據存入string直觀不少。這是由hash的結構性質決定的——hash自己就是一個鍵值對集合：一個「父鍵」下面包含了不少「子鍵」，每一個「子鍵」都對應一個值。根據前面的分析可知，結果集中的每一行實際上也是鍵值對集合。用Redis鍵值對集合表示Mysql鍵值對集合應該再合適不過了：對於結果集中的某一行，字段對應於hash的「子鍵」，字段對應的值就是hash「子鍵」對應的值，即結果集的一行恰好對應一個hash。這一想法的實現代碼以下：
view sourceprint?
01.
// 該函數把結果集中的每一行都存入一個HASH結構。HASH鍵應當包括結果集標識符和HASH編號，
02.
// 形如「cache.string:123456:1」
03.
string Cache2Hash(sql::Connection *mysql_connection,
04.
redisContext *redis_connection,
05.
sql::ResultSet *resultset,
06.
const string &resultset_id, int ttl) {
07.
if (resultset->rowsCount() == 0) {
08.
throw runtime_error("FAILURE - no rows");
09.
}
10.
// HASH鍵的前綴，包含告終果集的標識符
11.
string prefix("cache.hash:" + resultset_id + ":");
12.
unsigned int num_row = 1;  // HASH編號，附加於HASH鍵的末尾，從1開始
13.
sql::ResultSetMetaData *meta = resultset->getMetaData();
14.
unsigned int num_col = meta->getColumnCount();
15.
// 將結果集中全部行對應的全部HASH鍵存入該SET，SET鍵包含告終果集的標識符
16.
string redis_row_set_key("resultset.hash:" + resultset_id);
17.
redisReply *reply;
18.
string ttlstr;
19.
stringstream ttlstream;
20.
ttlstream << ttl;
21.
ttlstr = ttlstream.str();
22.
// 結果集中的每一行對應於一個HASH，將結果集的全部行都存入相應HASH中
23.
resultset->beforeFirst();
24.
while (resultset->next()) {
25.
string redis_row_key;  // HASH鍵名，由前綴和HASH編號組成
26.
stringstream keystream;
27.
keystream << prefix << num_row;
28.
redis_row_key = keystream.str();
29.
for (int i = 1; i <= num_col; ++i) {
30.
string col_label = meta->getColumnLabel(i);
31.
string col_value = resultset->getString(col_label);
32.
// 將結果集中一行的字段名和對應值存入HASH
33.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
34.
"HSET %s %s %s",
35.
redis_row_key.c_str(),
36.
col_label.c_str(),
37.
col_value.c_str()));
38.
freeReplyObject(reply);
39.
}
40.
// 將HASH鍵加入SET中
41.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
42.
"SADD %s %s",
43.
redis_row_set_key.c_str(),
44.
redis_row_key.c_str()));
45.
freeReplyObject(reply);
46.
// 設置HASH的過時時間
47.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
48.
"EXPIRE %s %s",
49.
redis_row_key.c_str(),
50.
ttlstr.c_str()));
51.
freeReplyObject(reply);
52.
++num_row;
53.
}
54.
// 設置SET的過時時間
55.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
56.
"EXPIRE %s %s",
57.
redis_row_set_key.c_str(),
58.
ttlstr.c_str()));
59.
freeReplyObject(reply);
60.
return redis_row_set_key;  // 返回SET鍵，以便於其餘函數獲取該SET中的內容
61.
}

至此，咱們已經給出了兩種存儲Mysql結果集的方案，這就是咱們在篇首提出的第一個問題，即選擇何種數據結構存儲Mysql結果集的答案。下一篇文章將研究第二個問題，即數據結構鍵的標識符選擇問題。

 

 

把Mysql結果集緩存到Redis的字符串或哈希結構中之後，咱們面臨一個新的問題，即如何爲這些字符串或哈希命名，也就是如何肯定它們的鍵。由於這些數據結構所對應的行都屬於某個結果集，假如能夠找到一種惟一標識結果集的方法，那麼只需爲這些數據結構分配一個惟一的序號，而後把結果集標識符與該序號結合起來，就能惟一標識一個數據結構了。因而，爲字符串和哈希命名的問題就轉化爲肯定結果集標識符的問題。
通過調研，發現一種較爲通用的肯定結果集標識符的方法。正如咱們所知道的，緩存在Redis中的結果集數據都是利用select等sql語句從Mysql中獲取的。一樣的查詢語句會生成一樣的結果集（這裏暫時不討論結果集中每條記錄的順序問題），這一性質恰好能夠用來肯定結果集的惟一標識符。固然，簡單地把整個sql語句做爲結果集標識符是不可取的，一個顯而易見的理由是，未經處理的sql查詢語句均包含若干空格，而Redis的鍵是不容許存在空格的。這時，咱們須要一個能夠把sql語句轉換爲惟一標識符的函數。一般，這一功能由散列函數完成，包括MD5，SHA系列等加密散列函數在內的不少算法都可達到這一目的。
肯定結果集標識符以後，從Redis讀數據或向Redis寫數據的思路就很清晰了。對於一個sql語句格式的數據請求，首先計算該語句的MD5並據此獲得結果集標識符，而後利用該標識符在Redis中查找該結果集。注意，結果集中的每一行都有一個相應的鍵，這些鍵都存儲在一個Redis集合結構中。這個集合剛好對應了所需的結果集，因此，該集合的鍵必須包含結果集標識符。若是Redis中不存在這樣一個集合，說明要找的結果集不在Redis中，因此須要執行相應的sql語句，在Mysql中查詢到相應的結果集，而後按照上面所說的辦法把結果集中的每一行以字符串或哈希的形式存入Redis。在Redis中查找相應結果集的代碼以下：
view sourceprint?
01.
// 該函數根據sql語句在Redis中查詢相應的結果集，並返回結果集中每一行所對應的數據結構的鍵
02.
vector<string> GetCache(sql::Connection *mysql_connection,
03.
redisContext *redis_connection,
04.
const string &sql, int ttl, int type) {
05.
vector<string> redis_row_key_vector;
06.
string resultset_id = md5(sql);  // 計算sql語句的md5，這是惟一標識結果集的關鍵
07.
// type==1時，該函數將查詢相應的STRING集合或將結果集寫入若干STRING
08.
string cache_type = (type == 1) ? "string" : "hash";
09.
// 根據type信息和結果集標識符合成SET鍵
10.
string redis_row_set_key = "resultset." + cache_type + ":" + resultset_id;
11.
redisReply *reply;
12.
// 嘗試從reply中獲取SET中保存的全部鍵
13.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
14.
"SMEMBERS %s",
15.
redis_row_set_key.c_str()));
16.
if (reply->type == REDIS_REPLY_ARRAY) {
17.
// 若是要找的SET不存在，說明Redis中沒有相應的結果集，須要調用Cache2String或
18.
// Cache2Hash函數把數據從Mysql拉取到Redis中
19.
if (reply->elements == 0) {
20.
freeReplyObject(reply);
21.
sql::Statement *stmt = mysql_connection->createStatement();
22.
sql::ResultSet *resultset = stmt->executeQuery(sql);
23.
if (type == 1) {
24.
redis_row_set_key = Cache2String(mysql_connection, redis_connection,
25.
resultset, resultset_id, ttl);
26.
} else {
27.
redis_row_set_key = Cache2Hash(mysql_connection, redis_connection,
28.
resultset, resultset_id, ttl);
29.
}
30.
// 再次嘗試從reply中獲取SET中保存的全部鍵
31.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
32.
"SMEMBERS %s",
33.
redis_row_set_key.c_str()));
34.
delete resultset;
35.
delete stmt;
36.
}
37.
// 把SET中的每一個STRING或HASH鍵存入redis_row_key_vector中
38.
string redis_row_key;
39.
for (int i = 0; i < reply->elements; ++i) {
40.
redis_row_key = reply->element[i]->str;
41.
redis_row_key_vector.push_back(redis_row_key);
42.
}
43.
freeReplyObject(reply);
44.
} else {
45.
freeReplyObject(reply);
46.
throw runtime_error("FAILURE - SMEMBERS error");
47.
}
48.
return redis_row_key_vector;
49.
}

如今咱們已經掌握了肯定Redis中的結果集標識符以及各數據結構的鍵的方法。下一篇文章將研究結果集在Redis中的排序和分頁問題。

在實現緩存排序功能以前，必須先明白這一功能的合理性。不妨思考一下，既然能夠在數據庫中排序，爲何還要把排序功能放在緩存中實現呢？這裏簡單總結了兩個緣由：首先，排序會增長數據庫的負載，難以支撐高併發的應用；其次，在緩存中排序不會遇到表鎖定的問題。Redis剛好提供了排序功能，使咱們能夠方便地實現緩存排序。
Redis中用於實現排序功能的是SORT命令。該命令提供了多種參數，能夠對列表，集合和有序集合進行排序。SORT命令格式以下：
view sourceprint?
1.
SORT key [BY pattern] [LIMIT offset count] [GET pattern [GET pattern ...]] [ASC | DESC] [ALPHA] [STORE destination]

BY參數用於指定排序字段，功能相似於SQL中的order by。對於列表和集合而言，僅按照它們的值進行排序每每沒有實際意義。以函數Cache2Hash返回的集合爲例（實際上返回的是集合鍵），該集合中存儲的是一系列完整的哈希鍵，只按照這些鍵進行排序，結果無非是按照數字或字典順序排列，其用處顯然不大。這是由於真正存儲行數據的是哈希結構自己，而非哈希鍵。假設集合鍵爲"resultset.hash:123456"，集合中每一個哈希鍵對應的哈希結構中都有一個名爲「timestamp」的字段，如今要把集合中的全部哈希鍵按照timestamp字段進行排序，這時，只需執行如下命令：
view sourceprint?
1.
SORT resultset.hash:123456 BY *->timestamp

從上例能夠看出，BY的真正威力在於它可讓SORT命令按照一個指定的外部鍵的外部字段進行排序。SORT用集合resultset.hash:123456中的每一個值（即每一個哈希鍵）替換BY參數後的第一個「*」，並依據「->」後面給出的字段獲取其值，最後根據這些字段值對哈希鍵進行排序。
LIMIT參數用於限制排序之後返回元素的數量，功能相似於SQL中的limit。該參數接受另外兩個參數，即offset和count，LIMIT offset count表示跳過前offset個元素，返回以後的連續count個元素。可見，LIMIT參數能夠用於實現分頁功能。
GET參數用於返回指定的字段值。以集合resultset.hash:123456爲例，使用BY參數對集合中的全部哈希鍵按照哈希結構中的timestamp字段排序後，SORT命令返回全部排序以後的哈希鍵。若是某個請求須要不是鍵而是某些字段值，這時就要使用GET參數，使SORT命令返回指定字段值。假設除timestamp字段之外，集合中每一個哈希鍵對應的哈希結構中還有一個名爲「id」的字段，經過如下命令可使SORT返回按照timestamp排序之後的每一個哈希鍵對應的哈希結構中的timestamp和id值：
view sourceprint?
1.
SORT resultset.hash:123456 BY *->timestamp GET *->timestamp GET *->id

SORT用集合resultset.hash:123456中的每一個值（即每一個哈希鍵）替換GET參數以後的第一個「*」，並將其做爲返回值。值得注意的是，利用GET #可以獲得集合中的哈希鍵自己。
ASC和DESC參數用於指定排序順序（默認爲ASC，即從低到高），ALPHA參數用於按照字典順序排列非數字元素。
STORE參數用於將SORT命令的返回值，即排序結果存入一個指定的列表。加上STORE參數後，SORT命令的返回值就變爲排序結果的個數。
下面的代碼實現了按照哈希的某個字段對集合中的哈希鍵排序，並將結果存入列表的過程：
view sourceprint?
01.
// 該函數對集合中的全部HASH鍵進行排序，排序依據是HASH鍵所對應的HASH中的某個字段，
02.
// 排序結果被存入一個LIST結構，LIST鍵應當包含結果集標識符和排序字段標識符，
03.
// 形如「sorted:123456:1234」
04.
string SortHash(sql::Connection *mysql_connection,
05.
redisContext *redis_connection,
06.
const string &resultset_id,
07.
const string &sort_field,
08.
int offset, int count, int order, int ttl) {
09.
// 只考慮存儲HASH鍵的SET
10.
string redis_row_set_key = "resultset.hash:" + resultset_id;
11.
redisReply *reply;
12.
// 檢測SET是否存在
13.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
14.
"EXISTS %s",
15.
redis_row_set_key.c_str()));
16.
if (reply->integer == 0) {
17.
freeReplyObject(reply);
18.
throw runtime_error("FAILURE - no resultsets");
19.
} else {
20.
freeReplyObject(reply);
21.
}
22.
string field_md5 = md5(sort_field);  // 利用MD5排除排序字段中空格形成的影響
23.
// 將排序結果存入該LIST
24.
string redis_sorted_list_key = "sorted:" + resultset_id + ":" + field_md5;
25.
string by("*->" + sort_field);  //肯定排序字段
26.
string ord = (order == 1) ? "ASC" : "DESC";  //order==1時按照升序排列；不然爲降序
27.
stringstream ofsstream, cntstream;
28.
ofsstream << offset;
29.
cntstream << count;
30.
// 執行排序命令，並把排序結果存入LIST
31.
reply = static_cast<redisReply*>(redisCommand(
32.
redis_connection,
33.
"SORT %s BY %s LIMIT %s %s GET %s ALPHA STORE %s",
34.
redis_row_set_key.c_str(),
35.
by.c_str(),
36.
ofsstream.str().c_str(),
37.
cntstream.str().c_str(),
38.
"#",
39.
redis_sorted_list_key.c_str()));
40.
freeReplyObject(reply);
41.
stringstream ttlstream;
42.
ttlstream << ttl;
43.
// 設置LIST的過時時間
44.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
45.
"EXPIRE %s %s",
46.
redis_sorted_list_key.c_str(),
47.
ttlstream.str().c_str()));
48.
freeReplyObject(reply);
49.
return redis_sorted_list_key;  // 返回LIST鍵，以便於其餘函數獲取該LIST中的內容

顯然，對結果集中的哈希鍵進行排序要比對字符串鍵排序更加直觀和方便。藉助於排序函數，能夠方便地實如今Redis中查詢排序後的結果集，代碼以下：
view sourceprint?
01.
// 該函數根據sql語句和排序參數，在Redis中查詢相應的結果集並進行排序，最後返回
02.
// 排序以後的HASH鍵
03.
vector<string> GetSortedCache(sql::Connection *mysql_connection,
04.
redisContext *redis_connection,
05.
const string &sql, const string &sort_field,
06.
int offset, int count, int order, int ttl) {
07.
vector<string> redis_row_key_vector;
08.
redisReply *reply;
09.
string resultset_id = md5(sql);  // 結果集標識符
10.
string field_md5 = md5(sort_field);  // 排序字段標識符
11.
// 嘗試獲取LIST中的全部HASH鍵
12.
string redis_sorted_list_key = "sorted:" + resultset_id + ":" + field_md5;
13.
// 嘗試獲取LIST中的全部HASH鍵
14.
reply = static_cast<redisReply*>(redisCommand(redis_connection,
15.
"LRANGE %s %s %s",
16.
redis_sorted_list_key.c_str(),
17.
"0",
18.
"-1"));
19.
if (reply->type == REDIS_REPLY_ARRAY) {
20.
// 若是LIST不存在，調用Cache2Hash函數從Mysql中拉取數據到Redis，而後調用SortHash函數
21.
// 對結果集進行排序並將排序後的HASH鍵存入LIST
22.
if (reply->elements == 0) {
23.
freeReplyObject(reply);
24.
sql::Statement *stmt = mysql_connection->createStatement();
25.
sql::ResultSet *resultset = stmt->executeQuery(sql);
26.
Cache2Hash(mysql_connection, redis_connection, resultset,
27.
resultset_id, ttl);
28.
redis_sorted_list_key = SortHash(mysql_connection, redis_connection,
29.
resultset_id, sort_field, offset,
30.
count, order, ttl);
31.
// 再次嘗試獲取LIST中的全部HASH鍵
32.
reply = static_cast<redisReply*>(redisCommand(
33.
redis_connection,
34.
"LRANGE %s %s %s",
35.
redis_sorted_list_key.c_str(),
36.
"0",
37.
"-1"));
38.
delete resultset;
39.
delete stmt;
40.
}
41.
// 將LIST中的全部HASH鍵存入redis_row_key_vector中
42.
string redis_row_key;
43.
for (int i = 0; i < reply->elements; ++i) {
44.
redis_row_key = reply->element[i]->str;
45.
redis_row_key_vector.push_back(redis_row_key);
46.
}
47.
freeReplyObject(reply);
48.
} else {
49.
freeReplyObject(reply);
50.
throw runtime_error("FAILURE - LRANGE error");
51.
}
52.
return redis_row_key_vector;
53.
}

這樣，在Redis中對結果集進行簡單排序操做的功能就實現了。