搜尋
首頁資料庫Redis怎麼用python快速搭建redis集群

  redis通訊協定

  列出主要的點,以便於對於下面程式的理解。

  Redis在TCP端口6379(預設端口,在配置可以修改)上監聽到來的連接,在客戶端與伺服器端之間傳輸的每個Redis命令或者資料都以rn結尾。

  回覆(服務端可客戶端復原的協定)

  Redis用不同的回覆類型回覆指令。它可能從伺服器發送的第一個位元組開始校驗回覆類型:

  * 用單行回覆(狀態回覆),回覆的第一個字節將是“ ”

  * 錯誤訊息,回覆的第一個位元組將是“-”

  * 整型數字,回覆的第一個字節將是“:”

* 批次回复,回复的第一個位元組將是“$”

  * 多個批次回复,回复的第一個字節將是“*”

Bulk Strings(批次回覆)

  批次回覆被伺服器用來傳回單一二進位安全字串。

  C: GET mykey

  S: $6rnfoobarrn

  伺服器發送第一行回复,該行以“$”開始後面跟隨實際要發送的字節數,隨後是CRLF,然後發送實際數據,隨後是2個位元組的額外數據用於最後的CRLF。伺服器發送的準確序列如下:

  ”$6rnfoobarrn”

  如果請求的值不存在,批量回復將使用特殊的值-1來作為資料長度,例如:

  C: GET nonexistingkey

  S: $-1

  當請求的物件不存在時,客戶端程式庫API不會傳回空字串,而會傳回空物件。例如:Ruby庫回傳‘nil’,而C庫則回傳NULL(或在回覆的物件裡設定指定的標誌)等等。

  二進位

  簡單說下二進制,就是會包含,所以C語言在處理的時候,就不能用str函數,像strlen、strcpy等,因為它們都是以來判斷字串結尾的。

  redis集群

  超簡單搭建redis集群

  官網也介紹瞭怎麼搭建redis集群,試過比較麻煩,因為用的centos6.5,如果用較新的centos,可能會好一點。

  Redis 集群的資料分片

  Redis 集群沒有使用一致性hash, 而是引入了哈希槽的概念.

  Redis 集群有16384個哈希槽,每個key透過CRC16校驗後對16384取模來決定放置哪個槽.集群的每個節點負責一部分hash槽,舉個例子,比如當前集群有3個節點,那麼:

  *節點A 包含0 到5500號哈希槽.

  * 節點B 包含5501 到11000 號哈希槽.

  * 節點C 包含11001 到16384號哈希槽.

#  這種結構很容易添加或刪除節點. 例如如果我想新新增個節點D, 我需要從節點A, B, C中得部分槽到D上. 如果我想移除節點A,需要將A中的槽移到B和C節點上,然後將沒有任何槽的A節點從集群中移除即可.   由於從一個節點將哈希槽移動到另一個節點並不會停止服務,所以無論添加刪除或改變某個節點的雜湊槽的數量都不會造成叢集不可用的狀態.

  Redis 叢集協定中的客戶端和伺服器端

  在Redis 叢集中,節點負責儲存資料、記錄叢集的狀態(包括鍵值到正確節點的對應)。叢集節點同樣能自動發現其他節點,偵測出沒正常運作的節點, 並且在需要的時候在從節點中推選出主節點。

  為了執行這些任務,所有的叢集節點都透過TCP連接(TCP bus?)和一個二進位協定(叢集連接,cluster bus)建立通訊。每一個節點都透過叢集連接(cluster bus)與叢集上的其餘每個節點連接起來。  節點們使用一個 gossip 協定來傳播叢集的訊息,這樣可以:發現新的節點、 發送ping包(用來確保所有節點都在正常工作中)、在特定情況發生時發送叢集訊息。集群連接也用於在集群中發布或訂閱訊息。

  由於叢集節點不能代理(proxy)請求,所以客戶端在接收到重定向錯誤(redirections errors) -MOVED 和 -ASK 的時候, 將指令重定向到其他節點。理論上來說,客戶端是可以自由地向叢集中的所有節點發送請求,在需要的時候把請求重定向到其他節點,所以客戶端是不需要保存叢集狀態。不過客戶端可以快取鍵值和節點之間的映射關係,這樣能明顯提高指令執行的效率。

  -MOVED

  簡單說下返回-MOVED的情況,就是客戶端連節點A請求處理key,但其實key其實在節點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就行了,其它都是一些socket的它都是一些socket的函數就行了,其它都是一些socket的函數都是一些socket封裝。

  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 = &#39;set %s_white_ip %srn&#39; % (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中文網其他相關文章!

陳述
本文轉載於:亿速云。如有侵權,請聯絡admin@php.cn刪除
REDIS:NOSQL方法的優勢REDIS:NOSQL方法的優勢Apr 27, 2025 am 12:09 AM

Redis是NoSQL數據庫,提供高性能和靈活性。 1)通過鍵值對存儲數據,適合處理大規模數據和高並發。 2)內存存儲和單線程模型確保快速讀寫和原子性。 3)使用RDB和AOF機制進行數據持久化,支持高可用性和橫向擴展。

REDIS:了解其架構和目的REDIS:了解其架構和目的Apr 26, 2025 am 12:11 AM

Redis是一种内存数据结构存储系统,主要用作数据库、缓存和消息代理。它的核心特点包括单线程模型、I/O多路复用、持久化机制、复制与集群功能。Redis在实际应用中常用于缓存、会话存储和消息队列,通过选择合适的数据结构、使用管道和事务、以及进行监控和调优,可以显著提升其性能。

REDIS與SQL數據庫:關鍵差異REDIS與SQL數據庫:關鍵差異Apr 25, 2025 am 12:02 AM

Redis和SQL數據庫的主要區別在於:Redis是內存數據庫,適用於高性能和靈活性需求;SQL數據庫是關係型數據庫,適用於復雜查詢和數據一致性需求。具體來說,1)Redis提供高速數據訪問和緩存服務,支持多種數據類型,適用於緩存和實時數據處理;2)SQL數據庫通過表格結構管理數據,支持複雜查詢和事務處理,適用於電商和金融系統等需要數據一致性的場景。

REDIS:它如何充當數據存儲和服務REDIS:它如何充當數據存儲和服務Apr 24, 2025 am 12:08 AM

REDISACTSASBOTHADATASTOREANDASERVICE.1)ASADATASTORE,ITUSESIN-MEMORYSTOOGATOFORFOFFASTESITION,支持VariousDatharptructuresLikeKey-valuepairsandsortedsetsetsetsetsetsetsets.2)asaservice,ItprovidespunctionslikeItionitionslikepunikeLikePublikePublikePlikePlikePlikeAndluikeAndluAascriptingiationsmpleplepleclexplectiations

REDIS與其他數據庫:比較分析REDIS與其他數據庫:比較分析Apr 23, 2025 am 12:16 AM

Redis與其他數據庫相比,具有以下獨特優勢:1)速度極快,讀寫操作通常在微秒級別;2)支持豐富的數據結構和操作;3)靈活的使用場景,如緩存、計數器和發布訂閱。選擇Redis還是其他數據庫需根據具體需求和場景,Redis在高性能、低延遲應用中表現出色。

REDIS的角色:探索數據存儲和管理功能REDIS的角色:探索數據存儲和管理功能Apr 22, 2025 am 12:10 AM

Redis在數據存儲和管理中扮演著關鍵角色,通過其多種數據結構和持久化機製成為現代應用的核心。 1)Redis支持字符串、列表、集合、有序集合和哈希表等數據結構,適用於緩存和復雜業務邏輯。 2)通過RDB和AOF兩種持久化方式,Redis確保數據的可靠存儲和快速恢復。

REDIS:了解NOSQL概念REDIS:了解NOSQL概念Apr 21, 2025 am 12:04 AM

Redis是一種NoSQL數據庫,適用於大規模數據的高效存儲和訪問。 1.Redis是開源的內存數據結構存儲系統,支持多種數據結構。 2.它提供極快的讀寫速度,適合緩存、會話管理等。 3.Redis支持持久化,通過RDB和AOF方式確保數據安全。 4.使用示例包括基本的鍵值對操作和高級的集合去重功能。 5.常見錯誤包括連接問題、數據類型不匹配和內存溢出,需注意調試。 6.性能優化建議包括選擇合適的數據結構和設置內存淘汰策略。

REDIS:現實世界的用例和示例REDIS:現實世界的用例和示例Apr 20, 2025 am 12:06 AM

Redis在現實世界中的應用包括:1.作為緩存系統加速數據庫查詢,2.存儲Web應用的會話數據,3.實現實時排行榜,4.作為消息隊列簡化消息傳遞。 Redis的多功能性和高性能使其在這些場景中大放異彩。

See all articles

熱AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智慧驅動的應用程序,用於創建逼真的裸體照片

AI Clothes Remover

AI Clothes Remover

用於從照片中去除衣服的線上人工智慧工具。

Undress AI Tool

Undress AI Tool

免費脫衣圖片

Clothoff.io

Clothoff.io

AI脫衣器

Video Face Swap

Video Face Swap

使用我們完全免費的人工智慧換臉工具,輕鬆在任何影片中換臉!

熱工具

SublimeText3 Linux新版

SublimeText3 Linux新版

SublimeText3 Linux最新版

SAP NetWeaver Server Adapter for Eclipse

SAP NetWeaver Server Adapter for Eclipse

將Eclipse與SAP NetWeaver應用伺服器整合。

VSCode Windows 64位元 下載

VSCode Windows 64位元 下載

微軟推出的免費、功能強大的一款IDE編輯器

ZendStudio 13.5.1 Mac

ZendStudio 13.5.1 Mac

強大的PHP整合開發環境

SublimeText3漢化版

SublimeText3漢化版

中文版,非常好用