前言

用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 <

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（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（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（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（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（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（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（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（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 GetCache（sql：：Connection *mysql_connection，

03.

redisContext *redis_connection，

04.

const string &sql， int ttl， int type） {

05.

vector 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（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（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（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（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（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 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 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（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（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中對結果集進行簡單排序操作的功能就實現了。