Bagaimana dengan cepat membina kluster redis menggunakan python

　Redis Communication Protocol

　Senaraikan perkara utama untuk memudahkan pemahaman program berikut.

Redis memantau sambungan masuk pada port TCP 6379 (port lalai, yang boleh diubah suai dalam konfigurasi Setiap arahan atau data Redis yang dihantar antara klien dan pelayan berakhir dengan rn).

Balas (protokol yang pelayan boleh pulih daripada klien)

Redis membalas arahan dengan jenis balasan yang berbeza. Ia mungkin menyemak jenis balasan bermula dari bait pertama yang dihantar oleh pelayan:

* Dengan balasan satu baris (balas status), bait pertama balasan akan menjadi “+”

* Mesej ralat, bait pertama balasan ialah "-"

* Nombor integer, bait pertama balasan ialah ":" > * Balasan kelompok, bait pertama balasan akan menjadi "$”

* Balasan berbilang kelompok, bait pertama balasan ialah "*"

Rentetan Pukal

Balasan pukal digunakan oleh pelayan untuk mengembalikan rentetan selamat binari tunggal.

C: GET mykey

S: $6rnfoobarrn

Pelayan menghantar baris pertama balasan, yang bermula dengan "$" diikuti dengan bait sebenar yang akan dihantar. nombor, diikuti dengan CRLF, kemudian data sebenar dihantar, diikuti dengan 2 bait data tambahan untuk CRLF akhir. Urutan tepat yang dihantar oleh pelayan adalah seperti berikut:

"$6rnfoobarrn"

Jika nilai yang diminta tidak wujud, balasan kelompok akan menggunakan nilai khas -1 sebagai panjang data, contohnya:

C: GET nonexistingkey

S: $-1

Apabila objek yang diminta tidak wujud, API pustaka klien tidak akan mengembalikan rentetan kosong, tetapi objek kosong. Sebagai contoh: perpustakaan Ruby mengembalikan "nihil", manakala perpustakaan C mengembalikan NULL (atau menetapkan bendera yang ditentukan dalam objek balasan), dsb.

Binari

Ringkasnya, binari bermakna ia akan dimasukkan Oleh itu, apabila memproses dalam bahasa C, anda tidak boleh menggunakan fungsi str, seperti strlen, strcpy, dll., kerana ia. adalah berdasarkan rentetan penghakiman.

kluster redis

Sangat mudah untuk membina kluster redis

Laman web rasmi juga memperkenalkan cara membina kluster redis Ia lebih menyusahkan untuk mencuba kerana centos6.5 digunakan . Jika anda menggunakan centos yang lebih baru mungkin lebih baik.

　Pecahan data gugusan Redis

　Kluster Redis tidak menggunakan pencincangan yang konsisten, tetapi memperkenalkan konsep slot cincang.

　Kluster Redis mempunyai 16384 slot cincang , setiap kunci ditandakan modulo 16384 selepas lulus semakan CRC16 untuk menentukan slot yang hendak diletakkan Setiap nod dalam gugusan bertanggungjawab untuk sebahagian daripada slot cincang Contohnya, jika gugusan semasa mempunyai 3 nod, maka:

*. Nod A mengandungi slot cincang 0 hingga 5500.

* Nod B mengandungi slot cincang 5501 hingga 11000.

* Nod C mengandungi slot cincang 11001 hingga 16384.

Struktur ini ia mudah untuk menambah atau memadam nod Sebagai contoh, jika saya ingin menambah nod baru D, saya perlu mendapatkan beberapa slot dari nod A, B dan C ke D. Jika saya ingin mengalih keluar nod A, saya perlu The. slot dalam A dialihkan ke nod B dan C, dan kemudian nod A tanpa sebarang slot dialih keluar daripada gugusan Memandangkan mengalihkan slot cincang dari satu nod ke nod lain tidak menghentikan perkhidmatan, tidak kira apa yang ditambahkan Memadam atau menukar. bilangan slot cincang nod tidak akan menyebabkan kluster menjadi tidak tersedia.

　Pihak klien dan pelayan dalam protokol kluster Redis

Dalam kluster Redis, nod Bertanggungjawab untuk menyimpan data dan rakaman status kluster (termasuk pemetaan nilai utama ke nod yang betul). Nod kluster juga boleh menemui nod lain secara automatik, mengesan nod yang tidak berfungsi dengan betul dan memilih nod induk daripada nod hamba apabila diperlukan.

Untuk melaksanakan tugasan ini, semua nod kluster berkomunikasi melalui sambungan TCP (bas TCP?) dan protokol binari (sambungan kluster, bas kluster). Setiap nod disambungkan kepada setiap nod lain dalam kluster melalui bas kluster. Nod menggunakan protokol gosip untuk menyebarkan maklumat kluster, yang boleh: menemui nod baharu, menghantar paket ping (untuk memastikan semua nod berfungsi dengan betul) dan menghantar mesej kluster apabila situasi tertentu berlaku. Sambungan kluster juga digunakan untuk menerbitkan atau melanggan mesej dalam kluster.

Memandangkan nod kluster tidak boleh meminta proksi, klien akan mengubah hala arahan ke nod lain apabila menerima ralat ubah hala -MOVED dan -ASK. Secara teorinya, klien bebas menghantar permintaan kepada semua nod dalam kluster dan mengalihkan permintaan ke nod lain apabila diperlukan, jadi klien tidak perlu menyimpan keadaan kluster. Walau bagaimanapun, pelanggan boleh menyimpan perhubungan pemetaan antara nilai utama dan nod, yang boleh meningkatkan kecekapan pelaksanaan arahan dengan ketara.

　 -MOVED

Ringkasnya, apabila -MOVED dikembalikan, ia adalah bahawa klien menyambung ke nod A dan meminta untuk memproses kunci, tetapi sebenarnya kunci itu sebenarnya pada nod B, jadi ia kembali -MOVED Protokolnya adalah seperti berikut: - MOVED 3999 127.0.0.1:6381

Jangan pertimbangkan situasi -TANYA.

Pelaksanaan bahasa C klien redis

Kodnya adalah seperti berikut:

#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

Terutamanya lihat fungsi check_ip_in_redis ini, yang lain adalah beberapa soket pembungkusan.

Python melaksanakan klien 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)

Atas ialah kandungan terperinci Bagaimana dengan cepat membina kluster redis menggunakan python. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!