同步方式
不過大多數情況下，我們采用的都是同步的調用方式。
|
1
2
3
4
5
6
7
8
9
10
11
|
/* 和redis服務器建立TCP連接 */
redisContext *redisConnectWithTimeout(const char *ip, int port, const struct timeval tv);
/* 發送指令到redis服務器，並取得結果 */
void *redisCommand(redisContext *c, const char *format, ...);
/* 處理完結果后，釋放內存 */
void freeReplyObject(void *reply);
/* 釋放和redis服務器的連接 */
void redisFree(redisContext *c);
|
沒錯，同步的方式就是這么簡單易用。
異步方式
不過說到了同步，那必然得說異步。同步簡單，異步性能高。為了性能，或者說為了裝B，必須得用一下異步調用方式才能顯示得出高大上。
當然，本文不是描述基於libevent，或者基於redis本身得ae網絡庫的異步方式。而是基於公司（或者自己）已有的異步服務器框架上，我們需要：
- 將發送redis的請求按照redis的既有協議打包到一個buffer。
- 將這個buffer丟給異步服務器框架，由框架負責和redis服務器建立（維護）連接，發送請求buffer，並取得redis返回的數據buffer。框架保證這個過程是異步高性能的。
- 然后框架會給出redis的回包buffer。我們需要按照redis的既有協議解開這個包，並得到結果。
redis的既有協議，這里先不展開描述。回到上面的需求，我們怎么才能打包redis的請求，解包redis的返回數據呢？用習慣了同步的方式，還曾經抱怨hiredis為什么不提供一個打包的函數和一個解包的函數呢。原來是自己眼拙，沒看出hiredis庫的打包和解包的使用方式。
打包函數系列：
|
1
2
3
4
|
/* Functions to format a command according to the protocol. */
int redisvFormatCommand(char **target, const char *format, va_list ap);
int redisFormatCommand(char **target, const char *format, ...);
int redisFormatCommandArgv(char **target, int argc, const char **argv, const size_t *argvlen);
|
解包是通過一個redisReader來完成的。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
/* State for the protocol parser */
typedef struct redisReader {
int err; /* Error flags, 0 when there is no error */
char errstr[128]; /* String representation of error when applicable */
char *buf; /* Read buffer */
size_t pos; /* Buffer cursor */
size_t len; /* Buffer length */
size_t maxbuf; /* Max length of unused buffer */
redisReadTask rstack[9];
int ridx; /* Index of current read task */
void *reply; /* Temporary reply pointer */
redisReplyObjectFunctions *fn;
void *privdata;
} redisReader;
/* Public API for the protocol parser. */
redisReader *redisReaderCreate(void);
void redisReaderFree(redisReader *r);
int redisReaderFeed(redisReader *r, const char *buf, size_t len);
int redisReaderGetReply(redisReader *r, void **reply);
|
呵呵，打包和解包看來不是一個簡單的函數搞定的，就原諒自己的眼拙了。哈哈。
實戰演習
來一個簡單的例子（demo.c），假設實例里面的函數
int asyncFrameWork(const uint8_t* sendBuf, int sendBufLen, uint8_t* recvBuf, int* recvBufLen);
就是公司高大上的異步服務器框架了啊。（汗顏一下。。。）
或者把這個asyncFrameWork替換為目前很前沿的微線程（協程）技術，那性能也是相當高的。
|
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
|
#include "hiredis.h"
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <errno.h>
#include <unistd.h>
#include <stdint.h>
int asyncFrameWork(const uint8_t* sendBuf, int sendBufLen, uint8_t* recvBuf, int* recvBufLen)
{
if (sendBuf == NULL || recvBuf == NULL || sendBufLen < 1 || *recvBufLen < 1) {
printf("asyncFrameWork param error.\n");
return -1;
}
int fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd < 0) {
printf("socket error : %d\n", errno);
return -1;
}
struct sockaddr_in addr;
memset(&addr, 0x00, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_port = htons(6379);
addr.sin_addr.s_addr = inet_addr("127.0.0.1");
int ret = connect(fd, (struct sockaddr*)&addr, sizeof(addr));
if (ret) {
printf("connect error : %d\n", errno);
close(fd);
return -2;
}
ret = send(fd, sendBuf, sendBufLen, 0);
if (ret < 0) {
printf("send error : %d\n", errno);
close(fd);
return -3;
}
ret = recv(fd, recvBuf, *recvBufLen, 0);
if (ret <= 0) {
printf("recv error : %d\n", errno);
close(fd);
return -4;
}
*recvBufLen = ret;
printf("asyncFrameWork recv len = %d\n", ret);
close(fd);
return 0;
}
int setCmd(const char* key, const char* value)
{
char* cmdBuf = NULL;
int len = 0;
len = redisFormatCommand(&cmdBuf, "SET %s %s", key, value);
if (len < 0) {
printf("redisFormatCommand error! len = %d\n", len);
return -1;
}
printf("redisFormatCommand result: len = %d, cmdBuf = %s\n", len, cmdBuf);
int ret = 0;
char recvBuf[1024] = {0};
int recvBufLen = sizeof(recvBuf);
ret = asyncFrameWork((const uint8_t*)cmdBuf, len, (uint8_t*)recvBuf, &recvBufLen);
if (ret) {
printf("asyncFrameWork error:%d!\n", ret);
free(cmdBuf);
cmdBuf = NULL;
return -2;
}
printf("asyncFrameWork return : recvBuf = %s, recvBufLen = %d\n", recvBuf, recvBufLen);
free(cmdBuf);
cmdBuf = NULL;
redisReader *reader = NULL;
void *reply = NULL;
reader = redisReaderCreate();
if (reader == NULL) {
printf("redisReaderCreate error!\n");
return -3;
}
redisReaderFeed(reader, recvBuf, recvBufLen);
ret = redisReaderGetReply(reader, &reply);
if (ret != REDIS_OK) {
printf("redisReaderGetReply error:%d!\n", ret);
redisReaderFree(reader);
reader = NULL;
return -4;
}
if (reply == NULL) {
printf("redisReply is NULL!\n");
redisReaderFree(reader);
reader = NULL;
return -5;
}
printf("redisReader get reply : %d\n", *((int*)reply));
struct redisReply *r = (struct redisReply*)reply;
printf("r->type = %d\n", r->type);
if (REDIS_REPLY_ERROR == r->type) {
printf("redisReply return REDIS_REPLY_ERROR!\n");
freeReplyObject(reply);
redisReaderFree(reader);
reader = NULL;
return -6;
}
if (REDIS_REPLY_STATUS != r->type) {
printf("redisReply return is't REDIS_REPLY_STATUS!\n");
freeReplyObject(reply);
redisReaderFree(reader);
reader = NULL;
return -7;
}
printf("-------------------------------------\n");
if (r->integer == REDIS_OK)
printf("set key(%s) OK\n", key);
else
printf("set key(%s) FAIL\n", key);
printf("-------------------------------------\n");
freeReplyObject(reply);
redisReaderFree(reader);
return 0;
}
int getCmd(const char* key)
{
char* cmdBuf = NULL;
int len = 0;
len = redisFormatCommand(&cmdBuf, "GET %s", key);
if (len < 0) {
printf("redisFormatCommand error! len = %d\n", len);
return -1;
}
printf("redisFormatCommand result: len = %d, cmdBuf = %s\n", len, cmdBuf);
int ret = 0;
char recvBuf[1024] = {0};
int recvBufLen = sizeof(recvBuf);
ret = asyncFrameWork((const uint8_t*)cmdBuf, len, (uint8_t*)recvBuf, &recvBufLen);
if (ret) {
printf("asyncFrameWork error:%d!\n", ret);
free(cmdBuf);
cmdBuf = NULL;
return -2;
}
printf("asyncFrameWork return : recvBuf = %s, recvBufLen = %d\n", recvBuf, recvBufLen);
free(cmdBuf);
cmdBuf = NULL;
redisReader *reader = NULL;
void *reply = NULL;
reader = redisReaderCreate();
if (reader == NULL) {
printf("redisReaderCreate error!\n");
return -3;
}
redisReaderFeed(reader, recvBuf, recvBufLen);
ret = redisReaderGetReply(reader, &reply);
if (ret != REDIS_OK) {
printf("redisReaderGetReply error:%d!\n", ret);
redisReaderFree(reader);
reader = NULL;
return -4;
}
if (reply == NULL) {
printf("redisReply is NULL!\n");
redisReaderFree(reader);
reader = NULL;
return -5;
}
printf("redisReader get reply : %d\n", *((int*)reply));
struct redisReply *r = (struct redisReply*)reply;
printf("r->type = %d\n", r->type);
if (REDIS_REPLY_ERROR == r->type) {
printf("redisReply return REDIS_REPLY_ERROR!\n");
freeReplyObject(reply);
redisReaderFree(reader);
reader = NULL;
return -6;
}
if (REDIS_REPLY_STRING != r->type) {
if (REDIS_REPLY_NIL == r->type) {
printf("-------------------------------------\n");
printf("key(%s) not exists!\n", key);
printf("-------------------------------------\n");
}
else
printf("redisReply return is't REDIS_REPLY_STRING!\n");
freeReplyObject(reply);
redisReaderFree(reader);
reader = NULL;
return -7;
}
printf("r->len = %d\n", r->len);
printf("r->str = %s\n", r->str);
printf("-------------------------------------\n");
printf("key(%s), value=(%s)\n", key, r->str);
printf("-------------------------------------\n");
freeReplyObject(reply);
redisReaderFree(reader);
return 0;
}
int main(int argc, char** argv)
{
setCmd("company", "leoox");
getCmd("company..");
getCmd("company");
return 0;
}
|
編譯
|
1
|
gcc -o demo -I/home/leoox/soft/hiredis-master demo.c /home/leoox/soft/hiredis-master/libhiredis.a
|
執行
用valgrind執行一下，看結果是否正確的同時，也順便看看是否有內存泄露。
[amcool@leoox redisDemo]$ valgrind –leak-check=full –tool=memcheck ./demo
==3868== Memcheck, a memory error detector.
==3868== Copyright (C) 2002-2006, and GNU GPL’d, by Julian Seward et al.
==3868== Using LibVEX rev 1658, a library for dynamic binary translation.
==3868== Copyright (C) 2004-2006, and GNU GPL’d, by OpenWorks LLP.
==3868== Using valgrind-3.2.1, a dynamic binary instrumentation framework.
==3868== Copyright (C) 2000-2006, and GNU GPL’d, by Julian Seward et al.
==3868== For more details, rerun with: -v
==3868==
redisFormatCommand result: len = 37, cmdBuf = *3
$3
SET
$7
company
$5
leoox
asyncFrameWork recv len = 5
asyncFrameWork return : recvBuf = +OK
, recvBufLen = 5
redisReader get reply : 5
r->type = 5
————————————-
set key(company) OK
————————————-
redisFormatCommand result: len = 28, cmdBuf = *2
$3
GET
$9
company..
asyncFrameWork recv len = 5
asyncFrameWork return : recvBuf = $-1
, recvBufLen = 5
redisReader get reply : 4
r->type = 4
————————————-
key(company..) not exists!
————————————-
redisFormatCommand result: len = 26, cmdBuf = *2
$3
GET
$7
company
asyncFrameWork recv len = 11
asyncFrameWork return : recvBuf = $5
leoox
, recvBufLen = 11
redisReader get reply : 1
r->type = 1
r->len = 5
r->str = leoox
————————————-
key(company), value=(leoox)
————————————-
==3868==
==3868== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 5 from 1)
==3868== malloc/free: in use at exit: 0 bytes in 0 blocks.
==3868== malloc/free: 41 allocs, 41 frees, 2,352 bytes allocated.
==3868== For counts of detected errors, rerun with: -v
==3868== All heap blocks were freed — no leaks are possible.
結果正確，並且沒有內存泄露。perfect！哦，聖誕過完了，那提前說元旦快樂，新的一年要任性！