腾讯云 Redis 集群版配置管理揭秘 ( 上 ）

 iRet = pthread_create(&pid, NULL, LoadConfigThread, NULL);
 if (iRet)
 {
 LOG_WATER_MUTEX(&g_stServerLog,_LOG_ERROR_, "Create LoadConfigThread error!\n");
 return -1;
 }

//下面这个结构体，我们看到astServerConfigs数组可以保存两个元素，一个保存在用的配置，另一个保存待更新的临时配置
typedef struct 
{
 int dwNowConfIndex; //指向最新更新的配置共享内存的指针 
 ServerConfigs astServerConfigs[2]; 
} ServerConfigsShm;

LOG_WATER_MUTEX(&g_stServerLog,_LOG_NOTICE_, "LoadConfigThread LoadConfigFromDB succ,ddwSeq:%ld", ddwSeq);

g_stConfig.pstServerConfigsShm->dwNowConfIndex++;
g_stConfig.pstServerConfigsShm->dwNowConfIndex %= 2;
memcpy( (void*)g_stConfig.pstVServerClientUpdateInfoList->astOneVServerClientUpdateInfo,
stTmpVServerClientUpdateInfoList.astOneVServerClientUpdateInfo,
stTmpVServerClientUpdateInfoList.dwServerNum*sizeof(OneVServerClientUpdateInfo) );
g_stConfig.pstVServerClientUpdateInfoList->dwServerNum = stTmpVServerClientUpdateInfoList.dwServerNum;

memcpy((char*)g_stConfig.pstServerCommInfo->strMD5Hash,strMD5Hash,16);

typedef struct
{
 VServerConfigList stVServerConfigList;
 IPVServer stIPVServer;
 ServerNameServer stServerNameServer;
 MicsInfoList stMicsInfoList;
 InterfaceIpList stInterfaceIpList;
 AppMicsInfoList stAppMicsInfoList;
} ServerConfigs;

// 管理所有v_server的数据结构
typedef struct 
{
 uint32_t dwVServerNum; //有多少个虚拟机
 OneVServerConfig astOneVServerConfig[MAX_VIRTUAL_COUNT_PER_COPY*MAX_COPY_NUM]; //保存所有虚拟机数据结构的数组
} VServerConfigList;
#define MAX_VIRTUAL_COUNT_PER_COPY 20000 // 单拷贝最大虚拟机数量
#define MAX_COPY_NUM 12 // 最大拷贝数量

// 一个虚拟机的数据结构
typedef struct 
{
 uint32_t dwServerId; // server id, 主键
 uint8_t sServerName[32]; // Server 名字，用于做hash用。多份拷贝的机器采用一一对应方式，一一对应的机器名字相同。这样前端寻址时候hash值才能一样。
 uint32_t dwIPAddr; // 存储机Ip地址，存储的时候用主机序来存储
 uint16_t wVNumber; // 该虚拟机在物理机器中的编号
 uint16_t wRole; // 角色 ,同gro_api.h中的ROLE_TYPE 
 uint16_t wVirtualNodeNum; // 该机器的虚拟节点个数
 uint16_t wIdcId; // 所归属的idc id
 uint16_t wProcNum; // 每个虚拟机器的进程数目 
 uint16_t wWorkFlag; // 工作状态配置，人工设置是否处于运行状态,和gro_api.h中STATE_TYPE定义相同 
 uint16_t wIfScale; // 扩容标记 0--非扩容，1--扩容中 
 uint16_t wRwsFlag; // 读写同步配置，人工设置是否处于读写或者同步
 uint16_t wMaskFlg;
 uint16_t wMaster; 
 uint16_t wWorkingState; // 机器上报运行/非运行状态 
 uint16_t wCopyId; // 拷贝id
#if defined(MEDIA_IS_MEM)
 uint32_t dwMaxAllowReads;
 uint32_t dwMaxAllowWrites;
#elif defined(MEDIA_IS_SSD)
 uint8_t sDevName[32]; // ssd盘设备名，如"dev/sdj"，互为备份的VServer该字段可以不同
#else
#error "MEDIA_IS_MEM or MEDIA_IS_SSD must be defined!"
#endif
 uint8_t sReserved[64]; 
}OneVServerConfig;

// v_server客户端的更新信息
typedef struct 
{
 uint32_t dwServerNum;
 OneVServerClientUpdateInfo astOneVServerClientUpdateInfo[MAX_VIRTUAL_COUNT_PER_COPY*MAX_COPY_NUM]; 
} VServerClientUpdateInfoList;

//一条v_server客户端的更新信息
typedef struct 
{
 uint32_t dwServerId; // 虚拟机编号
 time_t dwLastUpateTime; //最后上报时间，用于判断存活心跳
 uint32_t dwLastUpdatePriority; //最后上报时客户端配置的优先级
 uint8_t sMD5Hash[16]; //client最近上报的md5值 
 time_t dwLastPushConfigTime; //配置最后推送时间
 uint8_t sReserved[32]; // 保留4字节
} OneVServerClientUpdateInfo;

//管理所有 interface 机器的配置更新信息
typedef struct 
{
 uint32_t dwServerNum;
 OneInterfaceServerClientUpdateInfo astOneInterfaceServerClientUpdateInfo[MAX_INTERFACE_MACHINE_COUNT]; 
} InterfaceServerClientUpdateInfoList;

// interface 机器信息
typedef struct 
{
 uint32_t dwIP; // ip
 time_t dwLastUpateTime; //最后上报时间
 uint32_t dwLastUpdatePriority; // 最后上报时客户端配置的优先级
 uint8_t sMD5Hash[16]; // client最近上报的md5值 
 time_t dwLastPushConfigTime; //server 给client 下发配置的最新时间
 uint8_t sReserved[32];
} OneInterfaceServerClientUpdateInfo;

typedef struct 
{
 uint16_t wIPNum; 
 OneIPVServer astOneIPVServer[MAX_HARD_COUNT_PER_COPY*MAX_COPY_NUM];
}IPVServer;
#define MAX_HARD_COUNT_PER_COPY 20000 // 单拷贝最大物理机器数量
#define MAX_COPY_NUM 12 // 最大拷贝数量

// 物理机器ip 和虚拟机id 的对应关系
// 便于根据ip寻找虚拟机器
typedef struct 
{
 uint32_t dwIP; 
 uint16_t wVServerNum;
 uint32_t dwServerIdList[MAX_VIRTUAL_NUM_PER_MACHINE];
}OneIPVServer;
#define MAX_VIRTUAL_NUM_PER_MACHINE 20 // 每台物理机器最多有多少台虚拟机器

tInterval = TimeDiffSeconds(tCur, pstOneVServerClientUpdateInfo->dwLastUpateTime);

// 如果间隔时间超过伐值
 if (tInterval > g_stConfig.iReportStatusInterval*g_stConfig.iReportStatusTryTimes) //这两个配置参考上述初始化时从配置文件读入的项
 { 
 // 当前是WORKING的
 if (WORKING== pstOneVServerConfig->wWorkingState)
 {
 aiChangeServer[iChangeSvrNum] = pstOneVServerConfig->dwServerId;
 iChangeSvrNum++;
 tm* pTM = NULL;
 char szTime1[256] = {0};
 char szTime2[256] = {0};
 pTM = localtime((time_t*)&pstOneVServerClientUpdateInfo->dwLastUpateTime);
 sprintf(szTime1,"%04d-%02d-%02d %02d:%02d:%02d",pTM->tm_year+1900,pTM->tm_mon+1,pTM->tm_mday,pTM->tm_hour,pTM->tm_min,pTM->tm_sec);
 pTM = localtime((time_t*)&tCur);
 sprintf(szTime2,"%04d-%02d-%02d %02d:%02d:%02d",pTM->tm_year+1900,pTM->tm_mon+1,pTM->tm_mday,pTM->tm_hour,pTM->tm_min,pTM->tm_sec);
 uint32_t dwNIPAddr = htonl(pstOneVServerConfig->dwIPAddr);
 LOG_WATER_MUTEX(&g_stServerLog,_LOG_NOTICE_, "xxxx001 update state WORKING-->OFF_WORKING,vid:%u tInterval:%u dwLastUpateTime:%u %d %d time2:%s time1:%s ip:%s ",pstOneVServerConfig->dwServerId,tInterval,pstOneVServerClientUpdateInfo->dwLastUpateTime,g_stConfig.iReportStatusInterval,g_stConfig.iReportStatusTryTimes,szTime2,szTime1,inet_ntoa_MT(*(struct in_addr*)&dwNIPAddr).c_str()); 
 }
 }
 else //如果最近短时间内就有过心跳上报
 { // 如果是OFF_WORKING的话，我们就要考虑把他置为WORKING了。
 if (OFF_WORKING== pstOneVServerConfig->wWorkingState)
 {
 aiChangeServer[iChangeSvrNum] = pstOneVServerConfig->dwServerId;
 iChangeSvrNum++;
 tm* pTM = NULL;
 char szTime1[256] = {0};
 char szTime2[256] = {0};
 pTM = localtime((time_t*)&pstOneVServerClientUpdateInfo->dwLastUpateTime);
 sprintf(szTime1,"%04d-%02d-%02d %02d:%02d:%02d",pTM->tm_year+1900,pTM->tm_mon+1,pTM->tm_mday,pTM->tm_hour,pTM->tm_min,pTM->tm_sec);
 pTM = localtime((time_t*)&tCur);
 sprintf(szTime2,"%04d-%02d-%02d %02d:%02d:%02d",pTM->tm_year+1900,pTM->tm_mon+1,pTM->tm_mday,pTM->tm_hour,pTM->tm_min,pTM->tm_sec);
 uint32_t dwNIPAddr = htonl(pstOneVServerConfig->dwIPAddr);
 LOG_WATER_MUTEX(&g_stServerLog,_LOG_NOTICE_, "xxxx002 update state OFF_WORKING-->WORKING,vid:%u tInterval:%u dwLastUpateTime:%u %d %d time2:%s time1:%s ip:%s ",pstOneVServerConfig->dwServerId,tInterval,pstOneVServerClientUpdateInfo->dwLastUpateTime,g_stConfig.iReportStatusInterval,g_stConfig.iReportStatusTryTimes,szTime2,szTime1,inet_ntoa_MT(*(struct in_addr*)&dwNIPAddr).c_str());
 }
 }

// 更新状态
 for (i = 0; i < iChangeSvrNum; i++)
 {
 iTmpSrvIndex = aiChangeServer[i];
 pstOneVServerConfig = (OneVServerConfig* )FindVServerConfigById(iTmpSrvIndex,pstVServerConfigList);
 if (NULL == pstOneVServerConfig)
 {
 LOG_WATER_MUTEX(&g_stServerLog,_LOG_NOTICE_, "FindVServerConfigById err, iTmpSrvIndex:%d ",iTmpSrvIndex);
 Attr_API(183911,1); //根据serverid获取server信息失败
 continue;
 }
 pstOneVServerClientUpdateInfo = (OneVServerClientUpdateInfo* )FindOneVServerClientUpdateInfoById(iTmpSrvIndex, pstVServerClientUpdateInfoList);
 if (NULL == pstOneVServerClientUpdateInfo)
 {
 LOG_WATER_MUTEX(&g_stServerLog,_LOG_NOTICE_, "FindOneVServerClientUpdateInfoById err, iTmpSrvIndex:%d ",iTmpSrvIndex);
 Attr_API(183911,1); //根据serverid获取server信息失败
 continue;
 }
 // 如果当前是WORKDING,那么就改为OFF_WORKING
 if (WORKING== pstOneVServerConfig->wWorkingState)
 {
 LOG_WATER_MUTEX(&g_stServerLog,_LOG_NOTICE_, "update state WORKING-->OFF_WORKING,vid:%u ",iTmpSrvIndex);
 pstOneVServerConfig->wWorkingState = OFF_WORKING;
 IsUpdateStatus = 1;
 }
 // 如果当前是OFF_WORKDING，就改为WORKING
 else
 {
 LOG_WATER_MUTEX(&g_stServerLog,_LOG_NOTICE_, "update state OFF_WORKING-->WORKING,vid:%u ",iTmpSrvIndex);
 pstOneVServerConfig->wWorkingState = WORKING;
 IsUpdateStatus = 1;
 }
 }

PutFlagToMsgQueue(START,(char*)strMD5Hash); 
typedef struct 
{
 uint8_t cRole=宏START值是0
 uint32_t dwIP=0
 uint8_t strMD5Hash[16]=服务器最新配置16位md5
} PushConfigInfo;

typedef enum 
{
 START = 0,
 CFG_CLIENT_FOR_CACHE = 1, 
 CFG_CLIENT_FOR_IF = 2,
 CFG_CLIENT_FOR_CACHE_AND_IF = 3,
 END = 4,
} ROLE_TYPE;

PutInterfaceClientToMsgQueue(strMD5Hash,tCur,&dwNumber2);

typedef struct 
{
 uint8_t cRole=宏CFG_CLIENT_FOR_IF值是2
 uint32_t dwIP=interface的IP
 uint8_t strMD5Hash[16]=服务器最新配置md5
} PushConfigInfo;

PutCacheServerToMsgQueue(strMD5Hash,tCur,&dwNumber1);

typedef struct 
{
 uint8_t cRole=宏CFG_CLIENT_FOR_CACHE值是1
 uint32_t dwIP=cache的IP
 uint8_t strMD5Hash[16]=服务器最新配置md5
} PushConfigInfo;

 // 如果有需要推送的机器,把END标记也压到消息队列中
 if (dwNumber > 0)
 {
 PutFlagToMsgQueue(END,(char*)strMD5Hash);
 }
 // 如果没有需要推送的机器,也把END标记压到消息队列中,然后把为0的md5值也压入.
 else
 {
 uint8_t strTmpMD5Hash[16];
 memset((char* )strTmpMD5Hash,0,16);
 PutFlagToMsgQueue(END,(char*)strTmpMD5Hash);
 }

typedef struct 
{
 uint8_t cRole=宏END值是4
 uint32_t dwIP=0
 uint8_t strMD5Hash[16]=服务器最新配置md5 或者 '0000000000000000'
} PushConfigInfo;

for (i=0;i<g_stConfig.iPushConfigThreadNum;i++)
{
 iRet = pthread_create(&pid, NULL, PushConfigThread, NULL);
 if (iRet)
 {
 LOG_WATER_MUTEX(&g_stServerLog,_LOG_ERROR_, "Create PushConfigThread error!\n");
 return -4;
 }
}