redis接收客户端连接通过设置的acceptTcpHandler 进行处理
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
// MAX_ACCEPTS_PER_CALL = 1000
int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
char cip[NET_IP_STR_LEN];
UNUSED(el);
UNUSED(mask);
UNUSED(privdata);
// 接收客户端连接
while(max--) {
// accpet
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK)
serverLog(LL_WARNING,
"Accepting client connection: %s", server.neterr);
return;
}
serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
// handler accept
acceptCommonHandler(cfd,0,cip);
}
}
accpet过程
MAX_ACCEPTS_PER_CALL 标识每次最多accept 1000个客户端，如果超过1000，等到下次epoll通知在进行处理。
redis_max_accepts_per_call.png
需要注意的是，这种处理方式只能在LT模式下进行，ET下是不可以的，原因在于：
- ET模式下，epollo告诉系统fd发生READABLE事件时，必须一次读完，否则会丢事件。
- LT模式下，epollo告诉系统fd发生READABLE事件时，不需要一次读完，没有读的事件，下次epoll_wait时依然会告知。
redis对这部分的处理模式是LT的，也就是epoll默认的模式。
int anetTcpAccept(char *err, int s, char *ip, size_t ip_len, int *port) {
int fd;
struct sockaddr_storage sa;
socklen_t salen = sizeof(sa);
if ((fd = anetGenericAccept(err,s,(struct sockaddr*)&sa,&salen)) == -1)
return ANET_ERR;
if (sa.ss_family == AF_INET) {
struct sockaddr_in *s = (struct sockaddr_in *)&sa;
if (ip) inet_ntop(AF_INET,(void*)&(s->sin_addr),ip,ip_len);
if (port) *port = ntohs(s->sin_port);
} else {
struct sockaddr_in6 *s = (struct sockaddr_in6 *)&sa;
if (ip) inet_ntop(AF_INET6,(void*)&(s->sin6_addr),ip,ip_len);
if (port) *port = ntohs(s->sin6_port);
}
return fd;
}
anetTcpAccpet 会调用anetGenericAccpet接收客户端连接，然后解析IP地址和端口
static int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *len) {
int fd;
while(1) {
fd = accept(s,sa,len);
if (fd == -1) {
// EINTR 错误
if (errno == EINTR)
continue;
else {
anetSetError(err, "accept: %s", strerror(errno));
return ANET_ERR;
}
}
break;
}
return fd;
}
anetGenericAccept 调用socket的accpet 接收连接，当发生EINTR 系统中断时, 重启accept
handler过程
accept成功会通过acceptCommonHandler处理客户端的请求
static void acceptCommonHandler(int fd, int flags, char *ip) {
// 创建client
client *c;
if ((c = createClient(fd)) == NULL) {
serverLog(LL_WARNING,
"Error registering fd event for the new client: %s (fd=%d)",
strerror(errno),fd);
close(fd); /* May be already closed, just ignore errors */
return;
}
// client过多
// 这里没有直接close(fd)而是接收了fd并创建了客户端,原因在于redis希望把错误原因通过
// IO告诉client, 之后再关闭client
if (listLength(server.clients) > server.maxclients) {
char *err = "-ERR max number of clients reached\r\n";
/* That's a best effort error message, don't check write errors */
if (write(c->fd,err,strlen(err)) == -1) {
/* Nothing to do, Just to avoid the warning... */
}
server.stat_rejected_conn++;
freeClient(c);
return;
}
// 服务器处于protected_mode(保护模式)且没有设置密码, 并且绑定的端口也非特定端口,
// 同时ip是本地lo网卡, 出于安全性考虑, redis会关闭client, 并告知client错误原因
if (server.protected_mode &&
server.bindaddr_count == 0 &&
server.requirepass == NULL &&
!(flags & CLIENT_UNIX_SOCKET) &&
ip != NULL)
{
if (strcmp(ip,"127.0.0.1") && strcmp(ip,"::1")) {
char *err =
"-DENIED Redis is running in protected mode because protected "
"mode is enabled, no bind address was specified, no "
"authentication password is requested to clients. In this mode "
"connections are only accepted from the loopback interface. "
"If you want to connect from external computers to Redis you "
"may adopt one of the following solutions: "
"1) Just disable protected mode sending the command "
"'CONFIG SET protected-mode no' from the loopback interface "
"by connecting to Redis from the same host the server is "
"running, however MAKE SURE Redis is not publicly accessible "
"from internet if you do so. Use CONFIG REWRITE to make this "
"change permanent. "
"2) Alternatively you can just disable the protected mode by "
"editing the Redis configuration file, and setting the protected "
"mode option to 'no', and then restarting the server. "
"3) If you started the server manually just for testing, restart "
"it with the '--protected-mode no' option. "
"4) Setup a bind address or an authentication password. "
"NOTE: You only need to do one of the above things in order for "
"the server to start accepting connections from the outside.\r\n";
if (write(c->fd,err,strlen(err)) == -1) {
/* Nothing to do, Just to avoid the warning... */
}
server.stat_rejected_conn++;
freeClient(c);
return;
}
}
server.stat_numconnections++;
c->flags |= flags;
}
最关键的部分在createClient
client *createClient(int fd) {
client *c = zmalloc(sizeof(client));
// fd != -1时, 设置fd对应的TCP状态
if (fd != -1) {
// 设置fd非阻塞IO
anetNonBlock(NULL,fd);
// 设置TCPNoDeplay（非常重要，必须设置）
anetEnableTcpNoDelay(NULL,fd);
// 根据配置设置keeapalive (tcp层面)
if (server.tcpkeepalive)
anetKeepAlive(NULL,fd,server.tcpkeepalive);
// 设置事件处理回调
if (aeCreateFileEvent(server.el,fd,AE_READABLE,
readQueryFromClient, c) == AE_ERR)
{
close(fd);
zfree(c);
return NULL;
}
}
// 选择db
selectDb(c,0);
uint64_t client_id;
// 根据当前client_id设置下一个client_id
atomicGetIncr(server.next_client_id,client_id,1);
// 省略....设置client各种状态
return c;
}
createClient 在fd == -1 的情况下还是会创建一个client，只不过这个client是 “空的client” （没有连接），原因在于client是一个抽象概念，client并不等于实际的TCP连接，除了通过网络连接创建的client外，redis内部一些执行需要client对象。
对真正的连接，createClient会设置对应的TCP状态，其实就是常用的网络编程三板斧
- 事件驱动下必须设置的非阻塞IO
- 必须设置的Tcp No Deplay
- tcp 的 keepalive
之后会创建FileEvent，监听client的可读事件，并设置readQueryFromClient 作为其回调函数。
激发maxclients错误
为了测试该错误，首先将 redis.conf中的 maxclients设置为100，然后使用go client去连接
package main
import (
"context"
"flag"
"fmt"
"sync"
"github.com/go-redis/redis/v8"
)
var (
host = flag.String("host", "localhost:6379", "redis host")
redisMaxClients = flag.Int("maxclients", 100, "redis.conf maxclients options")
)
func main() {
flag.Parse()
wg := sync.WaitGroup{}
wg.Add(*redisMaxClients)
for i := 0; i < *redisMaxClients; i++ {
go func(client int) {
defer wg.Done()
ctx := context.Background()
rdb := redis.NewClient(&redis.Options{
Addr: *host,
Password: "", // no password set
DB: 0, // use default DB
})
_, err := rdb.Ping(ctx).Result()
if err != nil {
fmt.Printf("%d client error: %s\n", client, err.Error())
}
}(i)
}
wg.Wait()
}
执行结果
➜ go git:(dev) ✗ ./maxclient -maxclients=101
32 client error: ERR max number of clients reached
➜ go git:(dev) ✗ ./maxclient -maxclients=100
