Redis系列(三)--消息隊列、排行榜、慢查詢、pipeline等實現

Redis命令執行生命周期：

　　發送命令--->排隊（單線程）--->執行命令--->返回結果

慢查詢：

　　只是針對命令執行階段

　　慢查詢日志通過一個固定長度的FIFO queue，這個queue保存在內存中，通過設置命令執行時間慢查詢范圍，超過這個范圍進入慢查詢范圍，就

會保存到queue中

慢查詢有兩個相關參數：

　　slowlog-log-slower-than 1000

　　slowlog-max-len 1000

可以通過修改redis.conf或者命令config set slowlog-log-slower-than 1000設置，通過config get獲取參數

慢查詢命令：

slowlog get：

127.0.0.1:6379> slowlog get
1) 1) (integer) 2
   2) (integer) 1558081229
   3) (integer) 293
   4) 1) "COMMAND"
   5) "127.0.0.1:58194"
   6) ""
2) 1) (integer) 1
   2) (integer) 1552552609
   3) (integer) 15589
   4) 1) "save"
   5) "127.0.0.1:54516"
   6) ""

參數說明：

1、慢查詢記錄id

2、發起命令的時間戳

3、 命令耗時，單位為微秒

4、 該條記錄的命令及參數

5、客戶端網絡套接字（ip: port）

slowlog len：慢查詢隊列長度

127.0.0.1:6379> slowlog len
(integer) 2

slowlog reset：清空慢查詢隊列

127.0.0.1:6379> slowlog reset
OK

慢查詢日志優化：

　　1、slowlog-log-slower-than默認10000微秒，就是10ms，通常設置1ms

　　2、slowlog-max-lan默認128，通常設置1000，當超過最大queue長度，最先進入的記錄被剔除，最新的一條記錄加入slow log

　　3、參數可以動態設置，前面說了

　　4、可以定期將慢查詢日志進行持久化，因為它保存在內存中

pipeline：

　　1次網絡+n次命令時間

　　pipeline也就是流水線，將多個命令進行打包，在Redis server端計算出來，然后依次將結果返回

簡單應用：

@Test
public void countTimes() {
    SimpleDateFormat format = new SimpleDateFormat("hh:mm:ss");
    String startTime = format.format(new Date());
    log.info("開始時間：{}", startTime);
    String realKey = "test";
    Pipeline pipeline = redisService.pipeline();
    for (int i=0; i<10000; i++) {
        pipeline.set(realKey + i, "a" );
    }
    for (int i=0; i<10000; i++) {
        pipeline.del(realKey +i);
    }
    pipeline.sync();
    log.info("結束時間：{}", format.format(new Date()));
}

2019-05-17 16:57:14.357  INFO 11132 --- [           main] com.it.RedisServiceTest                  : 開始時間：04:57:14
2019-05-17 16:57:16.184  INFO 11132 --- [           main] com.it.RedisServiceTest                  : 結束時間：04:57:16

　　如果使用set和del各自操作10000次，由於本人在上海，買的阿里雲歸屬地是北京，加上配置太渣，網絡問題等，1分鍾過后才插入5000條。是

在等不下去了可以看出pipeline的速度是有多快

注意點：

　　m相關命令是原子操作，而pipeline不是，會拆分為很多子命令

計數器：

　　通過incr、incrby實現

應用場景：

　　用戶登錄次數記錄

　　社交點贊等

消息隊列：

一般可以用來單對單消息隊列，這不是Redis本身的功能，而是通過list實現，不保證可靠性投遞。如果真的需要消息隊列，還是通過MQ實現

實現：

127.0.0.1:6379> lpush list 1
(integer) 1
127.0.0.1:6379> blpop list 10                    //blpop，從左邊彈出一個元素，在timeout時間內如果沒有元素就阻塞
1) "list"
2) "1"
(4.98s)
127.0.0.1:6379> brpop list 10                    //brpop，從右邊談，和blpop相同
1) "list"
2) "1"
(4.98s)

java代碼實現：

Redis基本方法：lpush、rpush、lpop、rpop、brpop、blpop

public void lpush(byte[] key, byte[] value) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            jedis.lpush(key, value);
        } finally {
            returnToPool(jedis);
        }
    }

    public void rpush(byte[] key, byte[] value) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            jedis.rpush(key, value);
        } finally {
            returnToPool(jedis);
        }
    }

    public Object lpop(String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            String message = jedis.lpop(key);
            return message;
        } finally {
            returnToPool(jedis);
        }
    }

    public byte[] rpop(byte[] key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.rpop(key);
        } finally {
            returnToPool(jedis);
        }
    }

    public List<byte[]> brpop(int timeout, String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return jedis.brpop(timeout, key.getBytes());
        } finally {
            returnToPool(jedis);
        }
    }

@Data
@AllArgsConstructor
public class RedisMessage implements Serializable {

    private int id;
    private String message;

}

@Slf4j
@Service
public class RedisQueue {
    @Autowired
    private RedisService redisService;

    public void sendRedisMessage(int id, String message) {
        RedisMessage redisMessage = new RedisMessage(id, message);
        String key = RedisConstant.LIST_KEY + id;
        try {
            redisService.lpush(key.getBytes(), ObjectUtils.object2Bytes(redisMessage));
        } catch (IOException e) {
            log.error("Redis消息發送失敗：{}",e);
        }
    }

    public RedisMessage receiveMessage(int id){
        String key = RedisConstant.LIST_KEY + id;
        List<byte[]> list = redisService.brpop(0, key);
        RedisMessage message = null;
        try {
            message = (RedisMessage)ObjectUtils.bytes2Object(list.get(1));
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        return message;

    }
}

@Slf4j
public class RedisServiceTest extends ApplicationTests{

    @Autowired
    private RedisQueue redisQueue;

    @Test
    public void sendMessage() {
        for (int i=0; i<5; i++)
        redisQueue.sendRedisMessage(1, "send redis message!!!");
    }

    @Test
    public void receiveMessage() {
        RedisMessage message = redisQueue.receiveMessage(1);
        log.info("接收redis message：{}",message.getMessage());
    }
}

發送消息：

127.0.0.1:6379> lrange redisQueue1 0 -1
1) "\xac\xed\x00\x05sr\x00\x19com.it.redis.RedisMessage$\xd4D'>[g\xf8\x02\x00\x02I\x00\x02idL\x00\amessaget\x00\x12Ljava/lang/String;xp\x00\x00\x00\x01t\x00\x15send redis message!!!"
2) "\xac\xed\x00\x05sr\x00\x19com.it.redis.RedisMessage$\xd4D'>[g\xf8\x02\x00\x02I\x00\x02idL\x00\amessaget\x00\x12Ljava/lang/String;xp\x00\x00\x00\x01t\x00\x15send redis message!!!"
3) "\xac\xed\x00\x05sr\x00\x19com.it.redis.RedisMessage$\xd4D'>[g\xf8\x02\x00\x02I\x00\x02idL\x00\amessaget\x00\x12Ljava/lang/String;xp\x00\x00\x00\x01t\x00\x15send redis message!!!"
4) "\xac\xed\x00\x05sr\x00\x19com.it.redis.RedisMessage$\xd4D'>[g\xf8\x02\x00\x02I\x00\x02idL\x00\amessaget\x00\x12Ljava/lang/String;xp\x00\x00\x00\x01t\x00\x15send redis message!!!"
5) "\xac\xed\x00\x05sr\x00\x19com.it.redis.RedisMessage$\xd4D'>[g\xf8\x02\x00\x02I\x00\x02idL\x00\amessaget\x00\x12Ljava/lang/String;xp\x00\x00\x00\x01t\x00\x15send redis message!!!"

接收消息：

2019-05-20 10:35:04.576  INFO 3780 --- [           main] com.it.RedisServiceTest                  : 接收redis message：send redis message!!!

解釋：

RedisMessage：實體，實現Serializable，作為收發消息載體

RedisQueue：消息隊列，包含收發消息方法

RedisServiceTest：測試類

brpop(timeout, key)，timeout取0，表示如果無法取到消息，就會一直阻塞

發布訂閱：

　　一個新的訂閱者訂閱一個頻道是無法收到以前的消息的，沒有消息堆積的能力

角色：

　　發布者publisher、訂閱者subscriber、頻道channel

命令：

127.0.0.1:6379> subscribe myChannel 
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "myChannel"
3) (integer) 1
1) "message"
2) "myChannel"
3) "aaa"
127.0.0.1:6379> psubscribe channel*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "channel*"
3) (integer) 1
1) "pmessage"
2) "channel*"
3) "channel2"
4) "bbb"
1) "pmessage"
2) "channel*"
3) "channel1"
4) "aaa"
127.0.0.1:6379> unsubscribe myChannel
1) "unsubscribe"
2) "myChannel"
3) (integer) 0

127.0.0.1:6379> publish myChannel aaa
(integer) 1

psubscribe pattern：安裝某種方式進行訂閱，可以使用通配符

抽獎功能：set實現

127.0.0.1:6379> sadd choujiang zhangsan lisi wanger                //添加抽獎名單到set
(integer) 3
127.0.0.1:6379> smembers choujiang                                //獲取抽獎名單
1) "lisi"
2) "zhangsan"
3) "wanger"
127.0.0.1:6379> srandmember choujiang 2                            //從名單中隨機抽取2名，並且不刪除已中獎名單
1) "zhangsan"
2) "wanger"
127.0.0.1:6379> spop choujiang 2                                //從名單中隨機抽取2名，並且刪除已中獎名單
1) "wanger"
2) "lisi"
127.0.0.1:6379> smembers choujiang
1) "zhangsan"

實現點贊、簽到具體用戶列表：set實現

127.0.0.1:6379> sadd article:1001 zhangsan
(integer) 1
127.0.0.1:6379> sadd article:1001 lisi                        //lisi給1001文章點贊
(integer) 1
127.0.0.1:6379> srem article:1001 lisi                        //lisi給1001文章取消點贊
(integer) 1
127.0.0.1:6379> sismember article:1001 zhangsan                //檢查lisi是否給1001文章點過贊，個人覺得sadd也是一樣的，如果返回0，證明set已經包含了
(integer) 1
127.0.0.1:6379> sadd article:1001 lisi1
(integer) 1
127.0.0.1:6379> sadd article:1001 lisi2
(integer) 1
127.0.0.1:6379> smembers article:1001                        //獲取點贊列表
1) "zhangsan"
2) "lisi2"
3) "lisi1"
127.0.0.1:6379> scard article:1001                            //點贊人數
(integer) 3

共同關注列表：set實現

通過sinter實現

127.0.0.1:6379> sadd zhangsanlist jesen kobe
(integer) 2
127.0.0.1:6379> sadd lisilist jesen gakki hebe
(integer) 3
127.0.0.1:6379> sinter zhangsanlist lisilist
1) "jesen"

排行榜：zset實現

127.0.0.1:6379> zadd NouthAmercianMovieRanking 5702 Speed_preparation 2841 The_Avengers 2482 Big_detective_Pikachu
(integer) 3
127.0.0.1:6379> zrevrange NouthAmercianMovieRanking 0 2 withscores
1) "Speed_preparation"
2) "5702"
3) "The_Avengers"
4) "2841"
5) "Big_detective_Pikachu"
6) "2482"

部分內容參考：https://mp.weixin.qq.com/s/FyYhLS3X7LDe0PLxooz_cQ