redis6.2.6

一、简介

redis中位图不是一个数据类型，而是基于String类型的一系列位操作。并且redis的String类型是二进制安全的，最大长度512M，所以可以建立2322^{32}232个比特位。
位图操作分为两类：

常量时间操作
比如对某位的设置、获取。
范围操作
比如对某个范围内的位进行统计计数等，这样时间复杂度就不是常量级别了。

因为最大分配512M, 所以当使用的位图很大时，第一次访问不存在的位图时，分配大的空间，容易导致主线程阻塞

二、位图操作

2. 1 位图的最大限制

可通过redis.conf配置文件进行修改

# In the Redis protocol, bulk requests, that are, elements representing single
# strings, are normally limited to 512 mb. However you can change this limit
# here, but must be 1mb or greater
#
# proto-max-bulk-len 512mb

如果不配置，则默认值如下

...
standardConfig configs[] = {...
createLongLongConfig("proto-max-bulk-len", NULL, MODIFIABLE_CONFIG, 1024*1024, LONG_MAX, server.proto_max_bulk_len, 512ll*1024*1024, MEMORY_CONFIG, NULL, NULL), /* Bulk request max size */
...
};

如果修改，也不能太小，因为这也是请求协议数据最大的限制

2.2 SETBIT key offset value （设置）

时间复杂度 O(1)
比如下面的命令，设置key为name的第10bit为1

127.0.0.1:6379> setbit name 10 1

2.2.1 解析offset

因为网络上传输的都是字符串，这里需要将数字字符串转换为整数类型，比如本例子中的“10” -> 10

void setbitCommand(client *c) {...if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset,0,0) != C_OK)return;
...
}

int getBitOffsetFromArgument(client *c, robj *o, uint64_t *offset, int hash, int bits) {long long loffset;char *err = "bit offset is not an integer or out of range";char *p = o->ptr;size_t plen = sdslen(p);...if (string2ll(p+usehash,plen-usehash,&loffset) == 0) {addReplyError(c,err);return C_ERR;}.../* Limit offset to server.proto_max_bulk_len (512MB in bytes by default) */if ((loffset < 0) || (loffset >> 3) >= server.proto_max_bulk_len){addReplyError(c,err);return C_ERR;}*offset = loffset;return C_OK;
}

2.2.2 解析value

转换value的值，并且只能是0 或者 1

void setbitCommand(client *c) {...if (getLongFromObjectOrReply(c,c->argv[3],&on,err) != C_OK)return;/* Bits can only be set or cleared... */if (on & ~1) {addReplyError(c,err);return;}
}

可以看到，最终都是调用的string2ll

getLongFromObjectOrReply|VgetLongLongFromObjectOrReply|VgetLongLongFromObject|Vstring2ll

2.2.3 查询key

从全局字典中查询key

void setbitCommand(client *c) {...if ((o = lookupStringForBitCommand(c,bitoffset)) == NULL) return;
...
}

（1）未找到

1. 创建key, 插入到全局字典中

2. 创建 offset/8+1 字节的sds，作为值，默认清空的

robj *lookupStringForBitCommand(client *c, uint64_t maxbit) {...if (o == NULL) {o= createObject(OBJ_STRING,sdsnewlen(NULL, byte+1));dbAdd(c->db,c->argv[1],o);}
...
}

（2）找到

robj *lookupStringForBitCommand(client *c, uint64_t maxbit) {...if (o == NULL) {...} else { //找到o = dbUnshareStringValue(c->db,c->argv[1],o);o->ptr = sdsgrowzero(o->ptr,byte+1);}return o;
}

1. 判断是否为共享对象或者非原始字符串对象编码,是则创建一个新对象

robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o) {serverAssert(o->type == OBJ_STRING);if (o->refcount != 1 || o->encoding != OBJ_ENCODING_RAW) {robj *decoded = getDecodedObject(o);o = createRawStringObject(decoded->ptr, sdslen(decoded->ptr));decrRefCount(decoded);dbOverwrite(db,key,o);}return o;
}

2. 判断是否需要扩容，进行自动的扩容

sds sdsgrowzero(sds s, size_t len) {size_t curlen = sdslen(s);if (len <= curlen) return s;s = sdsMakeRoomFor(s,len-curlen);if (s == NULL) return NULL;/* Make sure added region doesn't contain garbage */memset(s+curlen,0,(len-curlen+1)); /* also set trailing \0 byte */sdssetlen(s, len);return s;
}

2.2.4 获取当前值，用于返回值

 /* Get current values */byte = bitoffset >> 3;byteval = ((uint8_t*)o->ptr)[byte];bit = 7 - (bitoffset & 0x7);bitval = byteval & (1 << bit);

2.2.5 设置值

 /* Update byte with new bit value and return original value */byteval &= ~(1 << bit);byteval |= ((on & 0x1) << bit);((uint8_t*)o->ptr)[byte] = byteval;

2.2.6 其他操作

 signalModifiedKey(c,c->db,c->argv[1]);notifyKeyspaceEvent(NOTIFY_STRING,"setbit",c->argv[1],c->db->id);server.dirty++;

2.2.7 响应客户端

  addReply(c, bitval ? shared.cone : shared.czero);

2.3 GETBIT key offset （获取）

2.3.1 解析offset

将字符串的数字offset转换为整数

void getbitCommand(client *c) {...if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset,0,0) != C_OK)return;
...
}

2.3.2 根据key查找对象，并进行类型判断

当key不存在时，直接返回0

   if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||checkType(c,o,OBJ_STRING)) return;

robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply) {robj *o = lookupKeyRead(c->db, key);if (!o) SentReplyOnKeyMiss(c, reply);return o;
}

127.0.0.1:6379> getbit nokey 10
(integer) 0

2.3.3 计算在字节数据中的实际偏移位置

byte = bitoffset >> 3;
bit = 7 - (bitoffset & 0x7);

2.3.4 获取值

当offset大于最大的长度时，返回0

 if (sdsEncodedObject(o)) {if (byte < sdslen(o->ptr)) //只处理范围内的bitval = ((uint8_t*)o->ptr)[byte] & (1 << bit);} else {if (byte < (size_t)ll2string(llbuf,sizeof(llbuf),(long)o->ptr))bitval = llbuf[byte] & (1 << bit);}

else分支，当创建string对象时，如果值是一个数字，并且在[LONG_MIN, LONG_MAX]范围内，这不用创建value的sds对象了，直接将ptr指针空间存储这个数字，比内嵌字符串对象更节省空间

2.3.5 响应客户端

addReply(c, bitval ? shared.cone : shared.czero);

2.4 BITCOUNT key [start end]（统计）

统计某个key中1的个数，可以设置start和end，只统计[start, end]范围内的置1的bit位

2.4.1 查询key

（1）key不存在，则直接返回0

（2）key存在，则对值进行类型检查，不是string则返回错误

void bitcountCommand(client *c) {.../* Lookup, check for type, and return 0 for non existing keys. */if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||checkType(c,o,OBJ_STRING)) return;
...
}

2.4.2 获取key对应的值

void bitcountCommand(client *c) {...
p = getObjectReadOnlyString(o,&strlen,llbuf);
...
}

如果是整数编码，则将整数转换位字符串返回，否则直接返回地址

unsigned char *getObjectReadOnlyString(robj *o, long *len, char *llbuf) {serverAssert(o->type == OBJ_STRING);unsigned char *p = NULL;/* Set the 'p' pointer to the string, that can be just a stack allocated* array if our string was integer encoded. */if (o && o->encoding == OBJ_ENCODING_INT) {p = (unsigned char*) llbuf;if (len) *len = ll2string(llbuf,LONG_STR_SIZE,(long)o->ptr);} else if (o) {p = (unsigned char*) o->ptr;if (len) *len = sdslen(o->ptr);} else {if (len) *len = 0;}return p;
}

2.4.3 解析参数start，end

（1）如果没有指定start,end，则start=0,end=strlen-1

（2）如果指定参数，从参数中提取start,end,并对其进行处理

（2.1）如果start,end都为负数，并且start > end，则返回0

（2.1）如果start,end为负数，则加上strlen

（2.2）如果start,end还为负数，则都等于0

（2.3）end>=start, end = strlen-1

（3）其他，

 /* Parse start/end range if any. */if (c->argc == 4) {if (getLongFromObjectOrReply(c,c->argv[2],&start,NULL) != C_OK)return;if (getLongFromObjectOrReply(c,c->argv[3],&end,NULL) != C_OK)return;/* Convert negative indexes */if (start < 0 && end < 0 && start > end) {addReply(c,shared.czero);return;}if (start < 0) start = strlen+start;if (end < 0) end = strlen+end;if (start < 0) start = 0;if (end < 0) end = 0;if (end >= strlen) end = strlen-1;} else if (c->argc == 2) {/* The whole string. */start = 0;end = strlen-1;} else {/* Syntax error. */addReplyErrorObject(c,shared.syntaxerr);return;}

2.4.4 判断start,end

如果start > end，则直接返回0

...
if (start > end) {addReply(c,shared.czero);
}
...

2.4.5 计算范围中的置1的bit位的个数，并返回

从代码可以看出，这个偏移是字节偏移，而非bit偏移

...
long bytes = end-start+1;addReplyLongLong(c,redisPopcount(p+start,bytes));

long long redisPopcount(void *s, long count) {long long bits = 0;unsigned char *p = s;uint32_t *p4;static const unsigned char bitsinbyte[256] = {0,1,1,2,1,2,2,3,1,2,2,3,2,3,3,4,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,4,5,5,6,5,6,6,7,5,6,6,7,6,7,7,8};/* Count initial bytes not aligned to 32 bit. */while((unsigned long)p & 3 && count) {bits += bitsinbyte[*p++];count--;}/* Count bits 28 bytes at a time */p4 = (uint32_t*)p;while(count>=28) {uint32_t aux1, aux2, aux3, aux4, aux5, aux6, aux7;aux1 = *p4++;aux2 = *p4++;aux3 = *p4++;aux4 = *p4++;aux5 = *p4++;aux6 = *p4++;aux7 = *p4++;count -= 28;aux1 = aux1 - ((aux1 >> 1) & 0x55555555);aux1 = (aux1 & 0x33333333) + ((aux1 >> 2) & 0x33333333);aux2 = aux2 - ((aux2 >> 1) & 0x55555555);aux2 = (aux2 & 0x33333333) + ((aux2 >> 2) & 0x33333333);aux3 = aux3 - ((aux3 >> 1) & 0x55555555);aux3 = (aux3 & 0x33333333) + ((aux3 >> 2) & 0x33333333);aux4 = aux4 - ((aux4 >> 1) & 0x55555555);aux4 = (aux4 & 0x33333333) + ((aux4 >> 2) & 0x33333333);aux5 = aux5 - ((aux5 >> 1) & 0x55555555);aux5 = (aux5 & 0x33333333) + ((aux5 >> 2) & 0x33333333);aux6 = aux6 - ((aux6 >> 1) & 0x55555555);aux6 = (aux6 & 0x33333333) + ((aux6 >> 2) & 0x33333333);aux7 = aux7 - ((aux7 >> 1) & 0x55555555);aux7 = (aux7 & 0x33333333) + ((aux7 >> 2) & 0x33333333);bits += ((((aux1 + (aux1 >> 4)) & 0x0F0F0F0F) +((aux2 + (aux2 >> 4)) & 0x0F0F0F0F) +((aux3 + (aux3 >> 4)) & 0x0F0F0F0F) +((aux4 + (aux4 >> 4)) & 0x0F0F0F0F) +((aux5 + (aux5 >> 4)) & 0x0F0F0F0F) +((aux6 + (aux6 >> 4)) & 0x0F0F0F0F) +((aux7 + (aux7 >> 4)) & 0x0F0F0F0F))* 0x01010101) >> 24;}/* Count the remaining bytes. */p = (unsigned char*)p4;while(count--) bits += bitsinbyte[*p++];return bits;
}

redis之瘦小精干的位图 (一)相关推荐

Redis 精确去重计数 —— 咆哮位图
如果要统计一篇文章的阅读量,可以直接使用 Redis 的 incr 指令来完成.如果要求阅读量必须按用户去重,那就可以使用 set 来记录阅读了这篇文章的所有用户 id,获取 set 集合的长度就是去 ...
Redis（十一）：Redis特殊类型之Bitmap位图
1.位存储只有0和1两种状态! Bitmap 位图:数据结构,都是操作二进制位来进行记录登录/未登录活跃/不活跃打卡两个状态的都可以使用Bitmap! 2.常用命令 2.1.用Bitmap来 ...
数据结构--位图 BitMap
文章目录 1. 位图 2. 位图代码 3. 布隆过滤器 Bloom Filter 4. 总结 1. 位图我们有1千万个整数,整数的范围在1到1亿之间.如何快速查找某个整数是否在这1千万个整数中呢? ...
1000+Redis实例，100+集群，Redis 在海量数据和高并发下的优化实践
墨墨导读:Redis 对于从事互联网技术工程师来说并不陌生,几乎所有的大中型企业都在使用 Redis 作为缓存数据库. 但是对于绝大多数企业来说只会用到它的最基础的 KV 缓存功能,还有很多 Redi ...
【缓存中间件】redis 支持的数据类型
文章目录前言一.字符串二.哈希三. 列表四.集合五.有序集合六.位图 Redis Bitmap 七.基数统计 HyperLogLog 八.Geo 地理位置九.Streams 流应用场 ...
Redis从入门到集群分布式实践
Redis 分布式实践一. Redis 初识 1.1 Redis 导学 Redis有哪些特点? 高性能的key-value服务器多种数据结构丰富的功能高可用分布式支持适合人群: 初学者进阶 ...
钱文品 | 《Redis在海量数据和高并发下的优化实践》主题分享
原文:http://www.enmotech.com/web/detail/1/750/1.html 导读:Redis 对于从事互联网技术工程师来说并不陌生,几乎所有的大中型企业都在使用 Redis ...
Redis应用---Redis可以用来做什么？
Redis可以用来干什么? 1.记录帖子的点赞数.评论数和点击数(hash) 2.记录用户的梯子ID列表(排序),便于快速显示用户的帖子列表(zset). 3.记录帖子的标题.摘要.作者和封面信息,用 ...
高并发和海量数据下的 9 个 Redis 经典案例剖析！
文章转载于:https://mp.weixin.qq.com/s?__biz=MzI1NDQ3MjQxNA==&mid=2247493368&idx=1&sn=3d6e44cb ...

redis之瘦小精干的位图 (一)