Python을 사용하여 Redis 클러스터를 빠르게 구축하는 방법

　Redis 통신 프로토콜

　다음 프로그램의 이해를 돕기 위해 주요 사항을 나열해 보세요.

　 Redis는 TCP 포트 6379(기본 포트, 구성에서 수정 가능)에서 들어오는 연결을 모니터링합니다. 클라이언트와 서버 간에 전송되는 모든 Redis 명령 또는 데이터는 rn으로 끝납니다.

　Reply(서버가 클라이언트로부터 복구할 수 있는 프로토콜)

　Redis는 다양한 응답 유형으로 명령에 응답합니다. 서버에서 보낸 첫 번째 바이트부터 시작하여 응답 유형을 확인할 수 있습니다.

　 * 한 줄 응답(상태 응답)의 경우 응답의 첫 번째 바이트는 "+"입니다.

　 * 응답의 오류 메시지 첫 번째 바이트는 "-"

　 * 정수, 답장의 첫 번째 바이트는 ":" $”

　 * 여러 개의 대량 답글, 답장의 첫 번째 바이트는 "*"입니다. 대량 문자열(대량 답장) )

　 대량 응답은 서버에서 단일 바이너리 안전 문자열을 반환하는 데 사용됩니다.

　 C: GET mykey

　 S: $6rnfoobarrn

　 서버는 첫 번째 응답 줄을 보냅니다. 이 줄은 "$"로 시작하고 그 뒤에 보낼 실제 바이트 수, CRLF가 뒤따르고 실제 데이터를 보냅니다. 그 다음에는 최종 CRLF에 대한 2바이트의 추가 데이터가 있습니다. 서버에서 보낸 정확한 순서는 다음과 같습니다.

　　"$6rnfoobarrn"

　요청한 값이 존재하지 않는 경우 일괄 응답에서는 특수 값 -1을 데이터 길이로 사용합니다. 예:

　C: GET nonexistingkey

　S: $- 1

　요청한 개체가 존재하지 않는 경우 클라이언트 라이브러리 API는 빈 문자열이 아닌 빈 개체를 반환합니다. 예를 들어 Ruby 라이브러리는 "nil"을 반환하는 반면 C 라이브러리는 NULL을 반환합니다(또는 응답 객체에 지정된 플래그를 설정합니다).

　　Binary

　쉽게 말하면 바이너리는 포함한다는 의미입니다. 따라서 C 언어에서 처리할 때 strlen, strcpy 등의 str 함수는 문자열의 끝을 결정하는 데 사용되므로 사용할 수 없습니다.

　 redis 클러스터

Redis 클러스터 구축 방법은 초간단

공식 홈페이지에서도 Centos6.5를 사용하기 때문에 Redis 클러스터 구축 방법을 소개하고 있어서 더 번거로울 수 있습니다. 나아지 다.

　 Redis 클러스터의 데이터 샤딩

　 Redis 클러스터는 일관된 해싱을 사용하지 않지만 해시 슬롯 개념을 도입합니다.

　 Redis 클러스터에는 16384개의 해시 슬롯이 있으며, 각 키는 CRC16을 통과한 후 확인됩니다. 클러스터의 각 노드는 해시 슬롯의 일부를 담당합니다. 예를 들어 현재 클러스터에 3개의 노드가 있는 경우

　 * 노드 A에는 해시 슬롯 0~5500이 포함됩니다.

　 * 노드 B에는 다음이 포함됩니다. 해시 슬롯 5501~11000.

　 * 노드 C에는 해시 슬롯 11001~16384가 포함됩니다.

이 구조를 사용하면 노드를 쉽게 추가하거나 삭제할 수 있습니다. 예를 들어 새 노드 D를 추가하려면 노드에서 시작해야 합니다. A, B와 C의 일부 슬롯이 D로 이동되었습니다. 노드 A를 제거하려면 A의 슬롯을 노드 B와 C로 이동한 다음 이동한 이후로 슬롯이 없는 A 노드를 제거해야 합니다. 한 노드에서 다른 노드로의 해시 슬롯은 서비스를 중지하지 않으며, 노드의 해시 슬롯 수를 추가, 삭제 또는 변경해도 클러스터를 사용할 수 없게 되지 않습니다.

　Redis 클러스터 프로토콜에서 클라이언트 및 서버 측

Redis 클러스터에서 노드는 데이터를 저장하고 클러스터 상태를 기록하는 역할을 담당합니다(올바른 노드에 대한 키 값 매핑 포함). 또한 클러스터 노드는 자동으로 다른 노드를 검색하고, 제대로 작동하지 않는 노드를 검색하고, 필요할 때 슬레이브 노드에서 마스터 노드를 선택할 수 있습니다.

　이러한 작업을 수행하기 위해 모든 클러스터 노드는 TCP 연결(TCP 버스?)과 바이너리 프로토콜(클러스터 연결, 클러스터 버스)을 통해 통신합니다. 각 노드는 클러스터 버스를 통해 클러스터의 다른 모든 노드에 연결됩니다. 노드는 가십 프로토콜을 사용하여 클러스터 정보를 전파합니다. 이를 통해 새 노드를 검색하고, 핑 패킷을 보내고(모든 노드가 제대로 작동하는지 확인하기 위해), 특정 상황이 발생할 때 클러스터 메시지를 보낼 수 있습니다. 클러스터 연결은 클러스터의 메시지를 게시하거나 구독하는 데에도 사용됩니다.

클러스터 노드는 요청을 프록시할 수 없으므로 클라이언트는 리디렉션 오류 -MOVED 및 -ASK를 수신하면 명령을 다른 노드로 리디렉션합니다. 이론적으로 클라이언트는 자유롭게 클러스터의 모든 노드에 요청을 보내고 필요할 때 요청을 다른 노드로 리디렉션할 수 있으므로 클라이언트는 클러스터 상태를 저장할 필요가 없습니다. 그러나 클라이언트는 키 값과 노드 간의 매핑 관계를 캐시할 수 있으므로 명령 실행 효율성을 크게 향상시킬 수 있습니다.

　 -MOVED

　 간단히 말하면 -MOVED가 반환되면 클라이언트가 노드 A에 연결하여 키 처리를 요청하지만 실제로는 키가 실제로 노드 B에 있으므로 -MOVED가 반환되는 프로토콜입니다. -MOVED 3999 127.0.0.1:6381

-ASK의 상황을 고려할 필요가 없습니다.

　　C 언어로 구현된 Redis 클라이언트

　코드는 다음과 같습니다:

#include <string.h>#include <sys/socket.h>#include <arpa/inet.h>#include <errno.h>#include <fcntl.h>#include <netdb.h>#include <sys/poll.h>#include <unistd.h>#include <sys/types.h>#include <stdlib.h>#include <stdio.h>ssize_t sock_write_loop( int fd, const void *vptr, size_t n )
{
    size_t nleft = 0;
    ssize_t nwritten = 0;const char *ptr;

    ptr = (char *) vptr;
    nleft = n;while( nleft > 0 )
    {if( (nwritten = write(fd, ptr, nleft) ) <= 0 )
        {if( errno == EINTR )
            {
                nwritten = 0;  //再次调用write            }else{return -5;
            }
        }
        nleft = nleft - nwritten;
        ptr = ptr + nwritten;
    }return(n);
}int sock_read_wait( int fd, int timeout )
{struct pollfd pfd;

    pfd.fd = fd;
    pfd.events = POLLIN;
    pfd.revents = 0;

    timeout *= 1000;for (;;) 
    {switch( poll(&pfd, 1, timeout) ) 
        {case -1:if( errno != EINTR ) 
                {return (-2);
                }continue;case 0:
                errno = ETIMEDOUT;return (-1);default:if( pfd.revents & POLLIN )return (0);elsereturn (-3);
        }
    }

}

ssize_t sock_read_tmo( int fd, void *vptr, size_t len, int timeout )
{   if( timeout > 0 && sock_read_wait(fd, timeout) < 0 )return (-1);elsereturn (read(fd, vptr, len));

}int sock_connect_nore(const char *IPaddr , int port , int timeout)
{   // char temp[4096];int sock_fd = 0, n = 0, errcode = 0;struct sockaddr_in servaddr;if( IPaddr == NULL )
    {return -1;
    }if( (sock_fd = socket(AF_INET, SOCK_STREAM, 0) ) < 0 )
    {return -1;
    }

    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port   = htons(port);//changed by navy 2003.3.3 for support domain addr//if( (servaddr.sin_addr.s_addr = inet_addr(IPaddr) ) == -1 )if( (errcode = inet_pton(AF_INET, IPaddr, &servaddr.sin_addr) ) <= 0 )
    {//added by navy 2003.3.31 for support domain addrstruct hostent* pHost = NULL, host;char sBuf[2048], sHostIp[17];int h_errnop = 0;

        memset(&host, 0, sizeof(host));
        memset(sBuf, 0, sizeof(sBuf));
        memset(sHostIp, 0 , sizeof(sHostIp));
        pHost = &host;

#ifdef _SOLARIS_PLAT//solarisif( (gethostbyname_r(IPaddr, pHost, sBuf, sizeof(sBuf), &h_errnop) == NULL) || 
#else//linuxif( (gethostbyname_r(IPaddr, pHost, sBuf, sizeof(sBuf), &pHost, &h_errnop) != 0) || 
#endif(pHost == NULL) ) 
                {
                close(sock_fd);return -1;
                }if( pHost->h_addrtype != AF_INET && pHost->h_addrtype != AF_INET6 )
                {
                close(sock_fd);return -1;
                }//目前仅取第一个IP地址if( (inet_ntop(pHost->h_addrtype, *(pHost->h_addr_list), sHostIp, sizeof(sHostIp)) ) == NULL )
                {
                    close(sock_fd);return -1;
                }                if( (errcode = inet_pton(AF_INET, sHostIp, &servaddr.sin_addr) ) <= 0 )
                {
                    close(sock_fd);                    return -1;
                }//end added by navy 2003.3.31    }if( (errcode = sock_timed_connect(sock_fd, (struct sockaddr *)&servaddr, sizeof(servaddr), timeout) ) < 0 )
    {
        close(sock_fd);        return -1;
    }return sock_fd;
}int sock_connect(const char *IPaddr , int port , int timeout)
{char temp[4096];int sock_fd = 0, n = 0, errcode = 0;struct sockaddr_in servaddr;if( IPaddr == NULL )
    {return -1;
    }if( (sock_fd = socket(AF_INET, SOCK_STREAM, 0) ) < 0 )
    {return -1;
    }

    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port   = htons(port);//changed by navy 2003.3.3 for support domain addr//if( (servaddr.sin_addr.s_addr = inet_addr(IPaddr) ) == -1 )if( (errcode = inet_pton(AF_INET, IPaddr, &servaddr.sin_addr) ) <= 0 )
    {//added by navy 2003.3.31 for support domain addrstruct hostent* pHost = NULL, host;char sBuf[2048], sHostIp[17];int h_errnop = 0;

        memset(&host, 0, sizeof(host));
        memset(sBuf, 0, sizeof(sBuf));
        memset(sHostIp, 0 , sizeof(sHostIp));
        pHost = &host;

#ifdef _SOLARIS_PLAT//solarisif( (gethostbyname_r(IPaddr, pHost, sBuf, sizeof(sBuf), &h_errnop) == NULL) || 
#else//linuxif( (gethostbyname_r(IPaddr, pHost, sBuf, sizeof(sBuf), &pHost, &h_errnop) != 0) || 
#endif(pHost == NULL) ) 
                {
                close(sock_fd);return -1;
                }if( pHost->h_addrtype != AF_INET && pHost->h_addrtype != AF_INET6 )
                {
                close(sock_fd);return -1;
                }//目前仅取第一个IP地址if( (inet_ntop(pHost->h_addrtype, *(pHost->h_addr_list), sHostIp, sizeof(sHostIp)) ) == NULL )
                {
                    close(sock_fd);return -1;
                }                if( (errcode = inet_pton(AF_INET, sHostIp, &servaddr.sin_addr) ) <= 0 )
                {
                    close(sock_fd);                    return -1;
                }//end added by navy 2003.3.31    }if( (errcode = sock_timed_connect(sock_fd, (struct sockaddr *)&servaddr, sizeof(servaddr), timeout) ) < 0 )
    {
        close(sock_fd);        return -1;
    }

    n = sock_read_tmo(sock_fd, temp, 4096, timeout);//一般错误if( n <= 0 ) 
    {
        close(sock_fd);
        
        sock_fd = -1;
    }return sock_fd;
}int sock_non_blocking(int fd, int on)
{int     flags;if ((flags = fcntl(fd, F_GETFL, 0)) < 0){return -10;
    }if (fcntl(fd, F_SETFL, on ? flags | O_NONBLOCK : flags & ~O_NONBLOCK) < 0){return -10;
    }return 0;
}int sock_write_wait(int fd, int timeout)
{struct pollfd pfd;

    pfd.fd = fd;
    pfd.events = POLLOUT;
    pfd.revents = 0;

    timeout *= 1000;for (;;) 
    {switch( poll(&pfd, 1, timeout) ) 
        {case -1:if( errno != EINTR ) 
            {return (-2);
            }continue;case 0:
            errno = ETIMEDOUT;return (-1);default:if( pfd.revents & POLLOUT )return (0);elsereturn (-3);
        }
    }

}int sock_timed_connect(int sock, struct sockaddr * sa, int len, int timeout)
{int error = 0;
    socklen_t error_len;

    sock_non_blocking(sock, 1);if( connect(sock, sa, len) == 0 )
    {
        sock_non_blocking(sock, 0);return (0);
    }if( errno != EINPROGRESS )
    {
        sock_non_blocking(sock, 0);return (-1);
    }/* * A connection is in progress. Wait for a limited amount of time for
     * something to happen. If nothing happens, report an error.     */if( sock_write_wait(sock, timeout) != 0)
    {
        sock_non_blocking(sock, 0);return (-2);
    }/* * Something happened. Some Solaris 2 versions have getsockopt() itself
     * return the error, instead of returning it via the parameter list.     */error = 0;
    error_len = sizeof(error);if( getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *) &error, &error_len) != 0 )
    {
        sock_non_blocking(sock, 0);return (-3);
    }if( error ) 
    {
        errno = error;
        sock_non_blocking(sock, 0);return (-4);
    }

    sock_non_blocking(sock, 0);/* * No problems.     */return (0);

}static int check_ip_in_list(const char *ip, char *iplist)
{        char *token = NULL;char *saveptr = NULL;
    token = strtok_r(iplist, ",", &saveptr);while(token != NULL)
    {        char *ptmp = NULL;                        char *ip_mask = strtok_r(token, "/", &ptmp);if(!ip_mask)                    return -1;                     char *ip_bit = strtok_r(NULL, "/", &ptmp);        if(ip_bit)
        {int mask_bit = atoi(ip_bit);if(mask_bit < 0 || mask_bit >32)continue;

            unsigned long addr[4] = { 0 };
            sscanf( ip_mask, "%lu.%lu.%lu.%lu", addr, addr + 1, addr + 2, addr + 3 );
            unsigned long vl1 = addr[0] << 24 | addr[1] << 16 | addr[2] << 8 | addr[3];

            sscanf( ip, "%lu.%lu.%lu.%lu", addr, addr + 1, addr + 2, addr + 3 );
            unsigned long vl2 = addr[0] << 24 | addr[1] << 16 | addr[2] << 8 | addr[3];

            vl1 = ( vl1 >> ( 32 - mask_bit ) );
            vl2 = ( vl2 >> ( 32 - mask_bit ) );if( vl1 == vl2 )                        return 1;                          
        }else{if(strcmp(ip,ip_mask) == 0)            return 1;                            
        }                    

        token = strtok_r(NULL, ",", &saveptr);                
    }        return 0;
}static int check_ip_in_redis(const char *redis_host, const char *ip,const char *rq_pro)
{char buf[128];int loops = 0;    

    strcpy(buf, redis_host);    do{
        loops ++;char *ptmp = NULL;char *host = strtok_r(buf, ":", &ptmp);if(!host) return -1;char *s_port = strtok_r(NULL, ":", &ptmp);if(!s_port) return -1;int port = atoi(s_port);char respone[40] = {0};int sock_fd = -1;if((sock_fd = sock_connect_nore(host, port, 5))<0)return -1;if(sock_write_loop(sock_fd, rq_pro, strlen(rq_pro)) != strlen(rq_pro))
        {
            close(sock_fd);return -1;
        }if(sock_read_tmo(sock_fd, respone, sizeof(respone)-1, 5)<=0)
        {
            close(sock_fd);return -1;
        }        if(strncmp(":0", respone, 2) == 0)
        {
            close(sock_fd);return 0;
        }            else if(strncmp(":1", respone, 2) == 0)
        {
            close(sock_fd);return 1;
        }            else if(strncmp("$", respone, 1) == 0)
        {                                    int data_size = 0;   int ret = 0;char *data_line = strstr(respone,"rn");if(!data_line)
            {
                close(sock_fd);return -1;
            }
            data_line = data_line+2;

            data_size = atoi(respone+1);if(data_size == -1)
            {
                close(sock_fd);return 0;
            }if(strlen(data_line) == data_size+2)
            {
                printf("line = %d, data_line = %sn",__LINE__,data_line);
                ret=check_ip_in_list(ip, data_line);
                close(sock_fd);return ret;
            }char *data = calloc(data_size+3,1);if(!data)
            {
                close(sock_fd);return -1;
            }
            strcpy(data,data_line);int read_size = strlen(data);int left_size = data_size + 2 - read_size;while(left_size > 0)
            {int nread = sock_read_tmo(sock_fd, data+read_size, left_size, 5);if(nread<=0)
                {free(data);
                    close(sock_fd);            return -1;
                }
                read_size += nread;
                left_size -= nread;
            }
            close(sock_fd);
            printf("line = %d, data = %sn",__LINE__,data);
            ret=check_ip_in_list(ip, data);free(data);return ret;
        }            else if(strncmp("-MOVED", respone, 6) == 0)
        {
            close(sock_fd);char *p = strchr(respone, &#39; &#39;);if(p == NULL)return -1;

            p = strchr(p+1, &#39; &#39;);if(p == NULL)return -1;

            strcpy(buf, p+1);
        }else{
            close(sock_fd);return -1;
        }            
        
    }while(loops < 2);return -1;
}int main(int argc,char *argv[])
{if(argc != 2)
    {
        printf("please input ipn");return -1;
    }     const char *redis_ip = "127.0.0.1:7002";const char *domain = "test.com";char exist_pro[128] = {0};char get_pro[128] = {0};    
    snprintf(exist_pro,sizeof(exist_pro),"EXISTS test|%s|%srn",domain,"127.0.0.1");        
    snprintf(get_pro,sizeof(get_pro),"GET test_%srn",domain);int loops = 0;int ret = 0;do{
        loops ++;
        ret = check_ip_in_redis(redis_ip, argv[1],exist_pro);if(ret == 0)
            ret = check_ip_in_redis(redis_ip, argv[1],get_pro);
    }while(loops < 3 && ret < 0);

    printf("line = %d, ret = %dn",__LINE__,ret);return ret;
}

c_redis_cli.c

　가장 중요한 것은 check_ip_in_redis 함수이고, 나머지는 소켓 캡슐화입니다.

　Python은 redis 클라이언트를 구현합니다

#!/usr/bin/pythonimport sys  
import socketdef main(argv):if(len(argv) != 3):print "please input domain ip!"returnhost = "192.168.188.47"   
    port = 7002while 1:
        s = socket.socket()                
        s.connect((host, port))
        
        cmd = 'set %s_white_ip %srn' % (argv[1],argv[2])        
        s.send(cmd)
        res = s.recv(32)
        s.close()        
    if res[0] == "+":print "set domain white  ip suc!"return    elif res[0:6] == "-MOVED":
            list = res.split(" ")
            ip_list = list[2].split(":")            
            host = ip_list[0]    
            port = int(ip_list[1])                            else:print "set domain white  ip error!"return                               if __name__ == "__main__":
    main(sys.argv)

위 내용은 Python을 사용하여 Redis 클러스터를 빠르게 구축하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!