今天碰到个很棘手的问题，因为这边我们服务器架构 ，网路是多线程 但是逻辑是单线程的，就出现了个问题，假如我此时一个操作去访问 redis 或者mysql 数据量过大，线程阻塞了，那么逻辑线程 其他消息将会延迟，如何优化？
问题如下:假如1000个玩家 都要去查询 1000名排行榜 必须在mysql 查找。
我的回答如下:
1.当时我就说设计不会出现这种情况，排序的名次，肯定是在内存实时动态更新的。(那贱逼，说我就要去mysql查,不考虑其他内存情况)
2. 做连接池，或者mysql开辟多线程可以解决一定性能问题（但是问题始终无法解决出现线程阻塞问题，无论架构设计，只要去mysql 查询数据)
3 模拟hiredis 使用异步回调。(那贱逼说我们不管redis提供的，如果写如何回调)
我回到家里，看了下hiredis 果然思路和我闲的一模一样，就是说最终我们要做成 a->b 假如b此时阻塞，但是a还是继续跑，b 阻塞结束，在通信a a在接受消息 －> client
hiredis 同步接口问题
int redisGetReply(redisContext *c, void **reply) {
int wdone = 0;
void *aux = NULL;
/* Try to read pending replies */
if (redisGetReplyFromReader(c,&aux) == REDIS_ERR)
return REDIS_ERR;
/* For the blocking context, flush output buffer and read reply */
if (aux == NULL && c->flags & REDIS_BLOCK) {
/* Write until done */
do {
if (redisBufferWrite(c,&wdone) == REDIS_ERR)// write 假如此时是select * from player
return REDIS_ERR;
} while (!wdone);
/* Read until there is a reply */
do {
if (redisBufferRead(c) == REDIS_ERR)//然后read mysql 获取的player 信息 问题在这里，这里是阻塞的，玩意数据过大，就呵呵了。
return REDIS_ERR;
if (redisGetReplyFromReader(c,&aux) == REDIS_ERR)
return REDIS_ERR;
} while (aux == NULL);
}
/* Set reply object */
if (reply != NULL) *reply = aux;
return REDIS_OK;}hiredis 异步接口问题
/* Helper function for the redisAsyncCommand* family of functions. Writes a
* formatted command to the output buffer and registers the provided callback
* function with the context. */
static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *cmd, size_t len) {
redisContext *c = &(ac->c);
redisCallback cb;//就是说我先把我需要的回调函数注册到一个list当中。然后在下一次read 出来在回调这个cb 这样就是异步回调模式了。这样变解决了，单逻辑线程出现的阻塞问题。
int pvariant, hasnext;
const char *cstr, *astr;
size_t clen, alen;
const char *p;
sds sname;
int ret;
/* Don't accept new commands when the connection is about to be closed. */
if (c->flags & (REDIS_DISCONNECTING | REDIS_FREEING)) return REDIS_ERR;
/* Setup callback */
cb.fn = fn;
cb.privdata = privdata;
/* Find out which command will be appended. */
p = nextArgument(cmd,&cstr,&clen);
assert(p != NULL);
hasnext = (p[0] == '$');
pvariant = (tolower(cstr[0]) == 'p') ? 1 : 0;
cstr += pvariant;
clen -= pvariant;
if (hasnext && strncasecmp(cstr,"subscribe\r\n",11) == 0) {
c->flags |= REDIS_SUBSCRIBED;
/* Add every channel/pattern to the list of subscription callbacks. */
while ((p = nextArgument(p,&astr,&alen)) != NULL) {
sname = sdsnewlen(astr,alen);
if (pvariant)
ret = dictReplace(ac->sub.patterns,sname,&cb);
else
ret = dictReplace(ac->sub.channels,sname,&cb);
if (ret == 0) sdsfree(sname);
}
} else if (strncasecmp(cstr,"unsubscribe\r\n",13) == 0) {
/* It is only useful to call (P)UNSUBSCRIBE when the context is
* subscribed to one or more channels or patterns. */
if (!(c->flags & REDIS_SUBSCRIBED)) return REDIS_ERR;
/* (P)UNSUBSCRIBE does not have its own response: every channel or
* pattern that is unsubscribed will receive a message. This means we
* should not append a callback function for this command. */
} else if(strncasecmp(cstr,"monitor\r\n",9) == 0) {
/* Set monitor flag and push callback */
c->flags |= REDIS_MONITORING;
__redisPushCallback(&ac->replies,&cb);
} else {
if (c->flags & REDIS_SUBSCRIBED)
/* This will likely result in an error reply, but it needs to be
* received and passed to the callback. */
__redisPushCallback(&ac->sub.invalid,&cb);
else
__redisPushCallback(&ac->replies,&cb);
}
__redisAppendCommand(c,cmd,len);
/* Always schedule a write when the write buffer is non-empty */
_EL_ADD_WRITE(ac);
return REDIS_OK;
}
int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, va_list ap) {
char *cmd;
int len;
int status;
len = redisvFormatCommand(&cmd,format,ap);
/* We don't want to pass -1 or -2 to future functions as a length. */
if (len < 0)
return REDIS_ERR;
status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
free(cmd);
return status;
}
网友评论