如何優雅地使用Redis之位圖操作
在進入今天的主題前,先簡單地解釋下Redis中的位圖到底是什么。Redis官方文檔對于位圖的介紹如下:
位圖不是一個真實的數據類型,而是定義在字符串類型上的面向位的操作的集合。由于字符串類型是二進制安全的二進制大對象,并且***長度是 512MB,適合于設置 2^32個不同的位。
位操作分為兩組:常量時間單個位的操作,像設置一個位為 1 或者 0,或者獲取該位的值。對一組位的操作,例如計算指定范圍位的置位數量。
位圖的***優勢是有時是一種非常顯著的節省空間來存儲信息的方式。例如,在一個系統中,不同用戶由遞增的用戶 ID 來表示,可以使用 512MB 的內存來表示 400 萬用戶的單個位信息(例如他們是否需要接收信件)。
簡而言之,位圖操作是用來操作比特位的,其優點是節省內存空間。為什么可以節省內存空間呢?假如我們需要存儲100萬個用戶的登錄狀態,使用位圖的話最少只需要100萬個比特位(比特位1表示登錄,比特位0表示未登錄)就可以存儲了,而如果以字符串的形式存儲,比如說以userId為key,是否登錄(字符串“1”表示登錄,字符串“0”表示未登錄)為value進行存儲的話,就需要存儲100萬個字符串了,相比之下使用位圖存儲占用的空間要小得多,這就是位圖存儲的優勢。
位圖常用操作
位圖的常用操作如下:
- setbit
設置特定key對應的比特位的值。
- getbit
獲取特定key對應的比特位的值。
- bitcount
統計給定key對應的字符串比特位為1的數量。
使用位圖存儲用戶登錄狀態
位圖的常見應用是用來存儲狀態值,比如存儲用戶的登錄狀態。
假設我們現在有一個需求,需要記錄用戶注冊以來每天的登錄狀態,那么我們就可以以用戶id為key,然后以日期或者日期的偏移量作為下標,將登錄狀態存儲到對應的比特位中,這樣就可以很方便地獲取用戶某一天的登錄狀態了。
接下來看代碼:
- public class UserLoginStatusService {
- private static final String host="111.111.111.111";
- private static final int port=6379;
- private static final Jedis jedis=new Jedis(host,port);
- //日期的初始值(也可以理解為用戶的注冊時間),
- //下文需要使用日期的偏移量作為redis位圖的offset,
- //因此需要將要保存登錄狀態的日期減去該初始日期。
- //這里使用了Java 8的新日期API
- private static final LocalDate beginDate=LocalDate.of(2018,1,1);
- static {
- jedis.connect();
- }
- public void setLoginStatus(String userId, LocalDate date,boolean isLogin){
- long offset = getDateDuration(beginDate, date);
- jedis.setbit(userId,offset,isLogin);
- }
- public boolean getLoginStatus(String userId,LocalDate date){
- long offset = getDateDuration(beginDate, date);
- return jedis.getbit(userId,offset);
- }
- private long getDateDuration(LocalDate start ,LocalDate end){
- return start.until(end, ChronoUnit.DAYS);
- }
- public static void main(String[] args) {
- UserLoginStatusService userLoginStatusService=new UserLoginStatusService();
- String userId="user_1";
- LocalDate today = LocalDate.now();
- userLoginStatusService.setLoginStatus(userId,today,true);
- boolean todayLoginStatus = userLoginStatusService.getLoginStatus(userId, today);
- System.out.println(String.format("The loginStatus of %s in %s is %s",userId,today,todayLoginStatus));
- LocalDate yesterday = LocalDate.now().minusDays(1);
- boolean yesterdayLoginStatus = userLoginStatusService.getLoginStatus(userId, yesterday);
- System.out.println(String.format("The loginStatus of %s in %s is %s",userId,yesterday,yesterdayLoginStatus));
- }
- }
代碼不復雜,我們在main方法中設置當天的登錄狀態為true,然后分別查出當天的登錄狀態和昨天的登錄狀態,由于redis位圖的比特位默認是0,所以該代碼的正確輸出應該是今天已登錄,昨天未登錄,我們運行一次看看結果。
從程序運行結果來看,Redis的位圖確實滿足了我們的需求,且兼有節省存儲空間的優點。
使用位圖統計登錄天數
接下來我們有一個新需求,就是統計某個用戶注冊后前10天的登錄天數,Redis中有個bitcount命令,可以統計某個字符串中的比特位為1的數量,其還有2個參數start和end,表示要統計的范圍,咋一看好像可以用來實現我們這個需求,但是這里有一個坑需要注意下,bitcount命令的start和end參數指的是字節的索引,而不是比特位的索引,而我們如果要使用位圖來統計某個用戶注冊后前10天的登錄天數的話,需要統計的是比特位索引從0到9的比特值為1的數量,所以直接使用bitcount命令顯然是無法滿足要求的。那么假如說我們一定要用位圖來存儲登錄狀態呢,應該咋辦呢?其實辦法還是有的。我們可以先拿到比特位索引從0到9所在的字節數組,再將該字節數組解析成二進制形式,進而統計出比特位索引從0到9比特值為1的數量。
要拿到比特位索引所在的字節在字節數組中的下標比較簡單,只要將比特位索引除以8(一個字節包含8個比特位)再向下取整就行了。接下來就是使用redis的getrange命令來截取字節數組了。
拿到了字節數組,接下來就是解析字節數組,統計其中比特值為1的數量了。我們先從最簡單的單個字節說起,假設一個字節的各個比特位的值如下:
我們設比特位索引為index,假如我們要計算比特位為7的比特值,只需要將原值直接跟1進行與運算就行了。要計算比特位為6的比特值,只需要將原值右移1位,再跟1進行與運算。以此類推,要計算第index位的比特值,只需要先右移(7-index)位,再跟1進行與運算即可。
只要能夠統計出截取出來的的字節數組中比特位的值為1的數量,接下來再減去不包含在對應比特索引中的比特值為1的數量,即可統計出給定的比特索引范圍內比特值為1的數量。
這么說有點拗口,我們以上述例子為例進行講解吧。我們要統計出用戶注冊后前10天的登錄天數,如果用位圖存儲用戶登錄狀態,位圖中的索引為注冊天數的話,那么我們需要統計比特索引從0到9的比特值為1的數量,才能計算出該用戶注冊后前10天的登錄天數。
我們先計算出比特索引從0到9包含在哪一段字節數組中,前面說了,只需要將對應的索引除以8,再向下取整就行了。從而可以得知比特位索引從0到9對應的是下標從0到1的字節數組。
接下來使用getrange命令截取該字節數組,假設其值如下:
假設比特索引0到9對應的字節數組的比特值情況如上所示,我們需要統計的是***個字節(下標為0)中的0到7位中比特值為1的數量,再加上第二個字節(下標為1)中的第0到1位中比特值為1的數量。加起來剛好10位,也就是對應用戶注冊前10天的登錄天數。當然我們也可以統計出這2個字節中的比特值為1的總數,再減去第二個字節的從2到7位(上述表格標紅的地方)中比特值為1的數量,也可統計出該用戶注冊后前10天的登錄天數。本文用的是第二種方法。
接下來上代碼:
- private static final int BIT_AMOUNT_IN_ONE_BYTE =8;
- private Jedis jedis;
- public int bitCountByBitIndex(String key, long startBitIndex, long endBitIndex) {
- int startByteIndex = getByteIndexInTheBytes(startBitIndex);
- int endByteIndex = getByteIndexInTheBytes(endBitIndex);
- byte[] bytes = jedis.getrange(key.getBytes(), startByteIndex, endByteIndex);
- int totalBitInBytes = getTotalBitInBytes(bytes);
- int startBitIndexInFirstByte = getBitIndexInTheByte(startBitIndex);
- int endBitIndexInLastByte = getBitIndexInTheByte(endBitIndex);
- byte firstByte = bytes[0];
- byte lastByte = bytes[bytes.length-1];
- for(int i=7;i>(BIT_AMOUNT_IN_ONE_BYTE-1-startBitIndexInFirstByte);i--){
- if(((firstByte>>i)&1)==1){
- totalBitInBytes--;
- }
- }
- for(int i=0;i<(BIT_AMOUNT_IN_ONE_BYTE-1-endBitIndexInLastByte);i++){
- if(((lastByte>>i)&1)==1){
- totalBitInBytes--;
- }
- }
- return totalBitInBytes;
- }
- private int getTotalBitInBytes(byte[] bytes){
- int count=0;
- for(byte b:bytes){
- for(int i = 0; i< BIT_AMOUNT_IN_ONE_BYTE; i++){
- if(((b>>i)&1)==1){
- count++;
- }
- }
- }
- return count;
- }
- private int getByteIndexInTheBytes(long offset){
- return (int) offset/ BIT_AMOUNT_IN_ONE_BYTE;
- }
- private int getBitIndexInTheByte(long offset){
- return (int)(offset-offset/ BIT_AMOUNT_IN_ONE_BYTE * BIT_AMOUNT_IN_ONE_BYTE);
- }
代碼就不注釋了,整體思路已經在上面講解了。
當然要實現本文所述的功能,也不一定非要這么做,還是有其他的方案的。比如:可以將放入位圖的offset統一乘以8(一個字節占8比特),這樣一來就可以直接用redis的bitcount來統計對應索引范圍內的比特值為1的數量了,當然這種方案的缺點也相當明顯,就是浪費內存,因為原先只需要1比特存儲的數據,現在需要8比特存儲,所以這種方案不能很好地利用位圖索引節省存儲空間的特點。