漫话Redis源码之三十一

2024-02-06 09:48
文章标签 源码 redis 三十一 漫话

本文主要是介绍漫话Redis源码之三十一,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

这里没什么特别的,考大家一个问题,为什么在mstime中需要加long long转化?是为了溢出,这里做得非常专业,我曾经踩过这个坑:

typedef struct _client {redisContext *context;sds obuf;char **randptr;         /* Pointers to :rand: strings inside the command buf */size_t randlen;         /* Number of pointers in client->randptr */size_t randfree;        /* Number of unused pointers in client->randptr */char **stagptr;         /* Pointers to slot hashtags (cluster mode only) */size_t staglen;         /* Number of pointers in client->stagptr */size_t stagfree;        /* Number of unused pointers in client->stagptr */size_t written;         /* Bytes of 'obuf' already written */long long start;        /* Start time of a request */long long latency;      /* Request latency */int pending;            /* Number of pending requests (replies to consume) */int prefix_pending;     /* If non-zero, number of pending prefix commands. Commandssuch as auth and select are prefixed to the pipeline ofbenchmark commands and discarded after the first send. */int prefixlen;          /* Size in bytes of the pending prefix commands */int thread_id;struct clusterNode *cluster_node;int slots_last_update;
} *client;/* Threads. */typedef struct benchmarkThread {int index;pthread_t thread;aeEventLoop *el;
} benchmarkThread;/* Cluster. */
typedef struct clusterNode {char *ip;int port;sds name;int flags;sds replicate;  /* Master ID if node is a slave */int *slots;int slots_count;int current_slot_index;int *updated_slots;         /* Used by updateClusterSlotsConfiguration */int updated_slots_count;    /* Used by updateClusterSlotsConfiguration */int replicas_count;sds *migrating; /* An array of sds where even strings are slots and odd* strings are the destination node IDs. */sds *importing; /* An array of sds where even strings are slots and odd* strings are the source node IDs. */int migrating_count; /* Length of the migrating array (migrating slots*2) */int importing_count; /* Length of the importing array (importing slots*2) */struct redisConfig *redis_config;
} clusterNode;typedef struct redisConfig {sds save;sds appendonly;
} redisConfig;/* Prototypes */
char *redisGitSHA1(void);
char *redisGitDirty(void);
static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask);
static void createMissingClients(client c);
static benchmarkThread *createBenchmarkThread(int index);
static void freeBenchmarkThread(benchmarkThread *thread);
static void freeBenchmarkThreads();
static void *execBenchmarkThread(void *ptr);
static clusterNode *createClusterNode(char *ip, int port);
static redisConfig *getRedisConfig(const char *ip, int port,const char *hostsocket);
static redisContext *getRedisContext(const char *ip, int port,const char *hostsocket);
static void freeRedisConfig(redisConfig *cfg);
static int fetchClusterSlotsConfiguration(client c);
static void updateClusterSlotsConfiguration();
int showThroughput(struct aeEventLoop *eventLoop, long long id,void *clientData);static sds benchmarkVersion(void) {sds version;version = sdscatprintf(sdsempty(), "%s", REDIS_VERSION);/* Add git commit and working tree status when available */if (strtoll(redisGitSHA1(),NULL,16)) {version = sdscatprintf(version, " (git:%s", redisGitSHA1());if (strtoll(redisGitDirty(),NULL,10))version = sdscatprintf(version, "-dirty");version = sdscat(version, ")");}return version;
}/* Dict callbacks */
static uint64_t dictSdsHash(const void *key);
static int dictSdsKeyCompare(void *privdata, const void *key1,const void *key2);/* Implementation */
static long long ustime(void) {struct timeval tv;long long ust;gettimeofday(&tv, NULL);ust = ((long)tv.tv_sec)*1000000;ust += tv.tv_usec;return ust;
}static long long mstime(void) {struct timeval tv;long long mst;gettimeofday(&tv, NULL);mst = ((long long)tv.tv_sec)*1000;mst += tv.tv_usec/1000;return mst;
}static uint64_t dictSdsHash(const void *key) {return dictGenHashFunction((unsigned char*)key, sdslen((char*)key));
}static int dictSdsKeyCompare(void *privdata, const void *key1,const void *key2)
{int l1,l2;DICT_NOTUSED(privdata);l1 = sdslen((sds)key1);l2 = sdslen((sds)key2);if (l1 != l2) return 0;return memcmp(key1, key2, l1) == 0;
}/* _serverAssert is needed by dict */
void _serverAssert(const char *estr, const char *file, int line) {fprintf(stderr, "=== ASSERTION FAILED ===");fprintf(stderr, "==> %s:%d '%s' is not true",file,line,estr);*((char*)-1) = 'x';
}static redisContext *getRedisContext(const char *ip, int port,const char *hostsocket)
{redisContext *ctx = NULL;redisReply *reply =  NULL;if (hostsocket == NULL)ctx = redisConnect(ip, port);elsectx = redisConnectUnix(hostsocket);if (ctx == NULL || ctx->err) {fprintf(stderr,"Could not connect to Redis at ");char *err = (ctx != NULL ? ctx->errstr : "");if (hostsocket == NULL)fprintf(stderr,"%s:%d: %s\n",ip,port,err);elsefprintf(stderr,"%s: %s\n",hostsocket,err);goto cleanup;}if (config.tls==1) {const char *err = NULL;if (cliSecureConnection(ctx, config.sslconfig, &err) == REDIS_ERR && err) {fprintf(stderr, "Could not negotiate a TLS connection: %s\n", err);goto cleanup;}}if (config.auth == NULL)return ctx;if (config.user == NULL)reply = redisCommand(ctx,"AUTH %s", config.auth);elsereply = redisCommand(ctx,"AUTH %s %s", config.user, config.auth);if (reply != NULL) {if (reply->type == REDIS_REPLY_ERROR) {if (hostsocket == NULL)fprintf(stderr, "Node %s:%d replied with error:\n%s\n", ip, port, reply->str);elsefprintf(stderr, "Node %s replied with error:\n%s\n", hostsocket, reply->str);freeReplyObject(reply);redisFree(ctx);exit(1);}freeReplyObject(reply);return ctx;}fprintf(stderr, "ERROR: failed to fetch reply from ");if (hostsocket == NULL)fprintf(stderr, "%s:%d\n", ip, port);elsefprintf(stderr, "%s\n", hostsocket);
cleanup:freeReplyObject(reply);redisFree(ctx);return NULL;
}static redisConfig *getRedisConfig(const char *ip, int port,const char *hostsocket)
{redisConfig *cfg = zcalloc(sizeof(*cfg));if (!cfg) return NULL;redisContext *c = NULL;redisReply *reply = NULL, *sub_reply = NULL;c = getRedisContext(ip, port, hostsocket);if (c == NULL) {freeRedisConfig(cfg);return NULL;}redisAppendCommand(c, "CONFIG GET %s", "save");redisAppendCommand(c, "CONFIG GET %s", "appendonly");int i = 0;void *r = NULL;for (; i < 2; i++) {int res = redisGetReply(c, &r);if (reply) freeReplyObject(reply);reply = res == REDIS_OK ? ((redisReply *) r) : NULL;if (res != REDIS_OK || !r) goto fail;if (reply->type == REDIS_REPLY_ERROR) {fprintf(stderr, "ERROR: %s\n", reply->str);goto fail;}if (reply->type != REDIS_REPLY_ARRAY || reply->elements < 2) goto fail;sub_reply = reply->element[1];char *value = sub_reply->str;if (!value) value = "";switch (i) {case 0: cfg->save = sdsnew(value); break;case 1: cfg->appendonly = sdsnew(value); break;}}freeReplyObject(reply);redisFree(c);return cfg;
fail:fprintf(stderr, "ERROR: failed to fetch CONFIG from ");if (hostsocket == NULL) fprintf(stderr, "%s:%d\n", ip, port);else fprintf(stderr, "%s\n", hostsocket);int abort_test = 0;if (reply && reply->type == REDIS_REPLY_ERROR &&(!strncmp(reply->str,"NOAUTH",5) ||!strncmp(reply->str,"WRONGPASS",9) ||!strncmp(reply->str,"NOPERM",5)))abort_test = 1;freeReplyObject(reply);redisFree(c);freeRedisConfig(cfg);if (abort_test) exit(1);return NULL;
}
static void freeRedisConfig(redisConfig *cfg) {if (cfg->save) sdsfree(cfg->save);if (cfg->appendonly) sdsfree(cfg->appendonly);zfree(cfg);
}static void freeClient(client c) {aeEventLoop *el = CLIENT_GET_EVENTLOOP(c);listNode *ln;aeDeleteFileEvent(el,c->context->fd,AE_WRITABLE);aeDeleteFileEvent(el,c->context->fd,AE_READABLE);if (c->thread_id >= 0) {int requests_finished = 0;atomicGet(config.requests_finished, requests_finished);if (requests_finished >= config.requests) {aeStop(el);}}redisFree(c->context);sdsfree(c->obuf);zfree(c->randptr);zfree(c->stagptr);zfree(c);if (config.num_threads) pthread_mutex_lock(&(config.liveclients_mutex));config.liveclients--;ln = listSearchKey(config.clients,c);assert(ln != NULL);listDelNode(config.clients,ln);if (config.num_threads) pthread_mutex_unlock(&(config.liveclients_mutex));
}static void freeAllClients(void) {listNode *ln = config.clients->head, *next;while(ln) {next = ln->next;freeClient(ln->value);ln = next;}
}

这篇关于漫话Redis源码之三十一的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/683908

相关文章

Knife4j+Axios+Redis前后端分离架构下的 API 管理与会话方案(最新推荐)

《Knife4j+Axios+Redis前后端分离架构下的API管理与会话方案(最新推荐)》本文主要介绍了Swagger与Knife4j的配置要点、前后端对接方法以及分布式Session实现原理,... 目录一、Swagger 与 Knife4j 的深度理解及配置要点Knife4j 配置关键要点1.Spri

Redis出现中文乱码的问题及解决

《Redis出现中文乱码的问题及解决》:本文主要介绍Redis出现中文乱码的问题及解决,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1. 问题的产生2China编程. 问题的解决redihttp://www.chinasem.cns数据进制问题的解决中文乱码问题解决总结

Redis的持久化之RDB和AOF机制详解

《Redis的持久化之RDB和AOF机制详解》:本文主要介绍Redis的持久化之RDB和AOF机制,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录概述RDB(Redis Database)核心原理触发方式手动触发自动触发AOF(Append-Only File)核

Redis分片集群、数据读写规则问题小结

《Redis分片集群、数据读写规则问题小结》本文介绍了Redis分片集群的原理,通过数据分片和哈希槽机制解决单机内存限制与写瓶颈问题,实现分布式存储和高并发处理,但存在通信开销大、维护复杂及对事务支持... 目录一、分片集群解android决的问题二、分片集群图解 分片集群特征如何解决的上述问题?(与哨兵模

SpringBoot连接Redis集群教程

《SpringBoot连接Redis集群教程》:本文主要介绍SpringBoot连接Redis集群教程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1. 依赖2. 修改配置文件3. 创建RedisClusterConfig4. 测试总结1. 依赖 <de

SpringBoot+Redis防止接口重复提交问题

《SpringBoot+Redis防止接口重复提交问题》:本文主要介绍SpringBoot+Redis防止接口重复提交问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不... 目录前言实现思路代码示例测试总结前言在项目的使用使用过程中,经常会出现某些操作在短时间内频繁提交。例

Redis 配置文件使用建议redis.conf 从入门到实战

《Redis配置文件使用建议redis.conf从入门到实战》Redis配置方式包括配置文件、命令行参数、运行时CONFIG命令,支持动态修改参数及持久化,常用项涉及端口、绑定、内存策略等,版本8... 目录一、Redis.conf 是什么?二、命令行方式传参(适用于测试)三、运行时动态修改配置(不重启服务

浅析如何保证MySQL与Redis数据一致性

《浅析如何保证MySQL与Redis数据一致性》在互联网应用中,MySQL作为持久化存储引擎,Redis作为高性能缓存层,两者的组合能有效提升系统性能,下面我们来看看如何保证两者的数据一致性吧... 目录一、数据不一致性的根源1.1 典型不一致场景1.2 关键矛盾点二、一致性保障策略2.1 基础策略:更新数

Redis Cluster模式配置

《RedisCluster模式配置》:本文主要介绍RedisCluster模式配置,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录分片 一、分片的本质与核心价值二、分片实现方案对比 ‌三、分片算法详解1. ‌范围分片(顺序分片)‌2. ‌哈希分片3. ‌虚

Springboot整合Redis主从实践

《Springboot整合Redis主从实践》:本文主要介绍Springboot整合Redis主从的实例,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录前言原配置现配置测试LettuceConnectionFactory.setShareNativeConnect