Redis學習三（進階功能）.

1、排序

redis 支持對 list，set 和 zset 元素的排序，排序的時間複雜度是 O(N+M*log(M))。（N 是集合大小，M 爲返回元素的數量）

sort key [BY pattern] [LIMIT offset count] [GET pattern [GET pattern ...]] [ASC|DESC] [ALPHA] [STORE destination]

• [BY pattern]：sort 命令默認使用集合元素進行排序，能夠經過 「BY pattern」 使用外部 key 的數據做爲權重排序。
• [LIMIT offset count]：排序以後返回元素的數量能夠經過 LIMIT 修飾符進行限制，修飾符接受 offset （要跳過的元素數量，即起始位置）和 count （返回的元素數量）兩個參數。
• [GET pattern [GET pattern ...]]：get 能夠根據排序的結果來取出相應的鍵值，「get #」 表示返回自身元素，「get pattern」 能夠返回外部 key 的數據 。
• [ASC|DESC] [ALPHA]：選擇按照順序、逆序或者字符串排序，set 集合（自己沒有索引值）排序操做必須指定 ALPHA。
• [STORE destination]：默認狀況下， sort 操做只是簡單地返回排序結果，並不進行任何保存操做。經過給 store 選項指定一個 key 參數，能夠將排序結果保存到給定的鍵上。

假設如今有用戶數據以下：

127.0.0.1:6379> sadd uid 1 2 3 4
127.0.0.1:6379> mset user_name_1 admin user_level_1 9999
127.0.0.1:6379> mset user_name_2 jack user_level_2 10
127.0.0.1:6379> mset user_name_3 peter user_level_3 25
127.0.0.1:6379> mset user_name_4 mary user_level_4 70

首先，直接利用集合內的元素作排序操做：

127.0.0.1:6379> sort uid
1) "1"
2) "2"
3) "3"
4) "4"

接着，咱們來試試 [BY pattern] 和 [GET pattern [GET pattern ...]] 操做：

127.0.0.1:6379> sort uid by user_name_* get # get user_name_* get user_level_* alpha
 1) "1"
 2) "admin"
 3) "9999"
 4) "2"
 5) "jack"
 6) "10"
 7) "4"
 8) "mary"
 9) "70"
10) "3"
11) "peter"
12) "25"

這個語句有點晦澀，試着這麼理解 「by user_name_* 」， user_name_* 是一個佔用符，它先取出 uid 中的值，而後再用這個值拼接成外部鍵，而真正進行排序的正是這些外部鍵值；「get # get user_name_* get user_level_* 」 的含義也能夠這麼理解，get # 表示返回本身元素，[get pattern] 表示返回外部鍵值。

2、事務

redis 的事務機制主要是由下面的幾個指令來完成：

• multi：標記一個事務塊的開始
• exec：執行全部事務塊中的命令
• discard：取消事務，放棄執行事務塊中的全部指令
• watch key [key...]：監視一個或多個 key，若是在事務執行以前這個（或這些key）被其餘命令所改動，這個改動也被稱爲 CAS 錯誤，那麼事務將被打斷
• unwatch：取消 watch 命令對全部 key 的監視

當 redis 接受到 multi 指令時，這個鏈接會進入一個事務上下文，該鏈接後續的命令並非當即執行，而是先放到一個隊列中；當從鏈接受到 exec 命令後，redis 會順序的執行隊列中的全部命令。並將全部命令的運行結果打包到一塊兒返回給 client。而後此鏈接就結束事務上下文。

redis 將是否有 watch 命令分爲普通類型事務和 CAS（Check And Set）類型事務，無 watch 命令的爲普通類型事務，有 watch 命令的爲 CAS類型事務。

事務失敗的緣由能夠分爲靜態錯誤（如不存在的命令）和運行時錯誤（如 CAS 錯誤、對 string 用 lpop 操做等）。靜態錯誤會在提交 exec 時返回錯誤信息，使事務不能執行；而除 CAS 之外的運行時錯誤不會阻止事務繼續執行。所以，Redis 的事務機制並不具備原子性。

127.0.0.1:6379> multi
OK
127.0.0.1:6379> lpush list java
QUEUED
127.0.0.1:6379> scard list
QUEUED
127.0.0.1:6379> exec
1) (integer) 1
2) (error) WRONGTYPE Operation against a key holding the wrong kind of value
127.0.0.1:6379> lrange list 0 -1
1) "java"

能夠看到，即便命令錯誤，事務依然沒有被回滾。所以，redis 的事務機制過於原始，不建議使用。

須要注意的是，若是咱們使用 AOF 的方式持久化，可能存在事務被部分寫入的狀況（事務執行過程當中 redis 掛掉等）從而致使 redis 啓動失敗退出，可使用 redis-check-aof 工具進行修復。

3、流水線（pipeline）

在事務中 redis 提供了隊列，能夠批量執行任務，這樣性能就比較高，但使用 multi…exec 事務命令是有系統開銷的，由於它會檢測對應的鎖和序列化命令。有時咱們但願在沒有任何附加條件的狀況下使用隊列批量執行一系列命令，這時可使用 redis的流水線（pipeline）技術。

看看如何在 spring-data-redis 中使用 pipeline 功能，須要注意的是 RedisTemplate 的序列化須要使用 StringRedisSerializer，不能使用 JdkSerializationRedisSerializer，不然會致使序列化失敗。

@Test
    public void pipeline() {
        // 1.executePipelined 重寫 入參 RedisCallback 的doInRedis方法
        List<Object> resultList = redisTemplate.executePipelined(new RedisCallback<String>() {
            @Override
            public String doInRedis(RedisConnection redisConnection) throws DataAccessException {
                // 2.redisConnection 給本次管道內添加 要一次性執行的多條命令
                // 2.1 一個set操做
                redisConnection.set("hello".getBytes(), "world".getBytes());
                // 2.2一個批量mset操做
                Map<byte[], byte[]> tuple = new HashMap();
                tuple.put("m_hello_1".getBytes(), "m_world_1".getBytes());
                tuple.put("m_hello_2".getBytes(), "m_world_2".getBytes());
                tuple.put("m_hello_3".getBytes(), "m_world_3".getBytes());
                redisConnection.mSet(tuple);
                // 2.3一個get操做
                redisConnection.get("m_hello_1".getBytes());
                // 3 這裏必定要返回null，最終pipeline的執行結果，纔會返回給最外層
                return null;
            }
        });
        // 4. 最後對redis pipeline管道操做返回結果進行判斷和業務補償
        for (Object str : resultList) {
            System.out.println(str);
        }
    }

4、發佈訂閱

繼 RxJava、Reactor、CompletableFuture 以及 Spring 的事件驅動模型後，咱們又要接觸一種觀察者模式啦！redis 做爲一個pub/sub server，在訂閱者和發佈者之間起到了消息路由的功能。訂閱者能夠經過 subscribe 和 psubscribe 命令向 redis server 訂閱本身感興趣的channel ；發佈者經過 publish 命令向 redis server 的 channel 發送消息，訂閱該 channel 的所有 client 都會收到此消息。一個 client 能夠訂閱多個 channel，也能夠向多個 channel 發送消息。

訂閱 channel：

127.0.0.1:6379> subscribe channel
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "channel"
3) (integer) 1
1) "message"
2) "channel"
3) "Hello,World"

發佈 channel 消息：

127.0.0.1:6379> publish channel Hello,World
(integer) 1

接下來咱們來看看在 spring-data-redis 中如何實現發佈訂閱功能。首先咱們須要一個消息監聽器，只要讓它實現 MessageListener 接口便可：

public class ChannelListener implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        System.out.println("channel is：" + new String(message.getChannel()));
        System.out.println("channel content：" + new String(message.getBody()));
    }
}

那麼，下面怎麼作呢？固然是把消息監聽器和 channel 綁定在一塊兒，讓消息監聽器知道處理哪一個 channel 的消息：

/**
     * redis 消息監聽器容器， 綁定消息監聽器和 channel
     *
     * @param connectionFactory
     * @return
     */
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //訂閱了一個叫 channel 的通道
        container.addMessageListener(channelAdapter(), new PatternTopic("channel"));
        //這個 container 能夠添加多個 messageListener
        return container;
    }

    /**
     * 消息監聽器適配器
     */
    @Bean
    public MessageListenerAdapter channelAdapter() {
        return new MessageListenerAdapter(new ChannelListener());
    }

接下來，讓咱們試着往這個 channel 發佈一個消息吧！

@Test
    public void publish() {
        redisTemplate.convertAndSend("channel", "Hello,World");
    }