阅读目录

• 一、排序
• 二、事务
• 三、流水线（pipeline）
• 四、发布订阅

回到顶部

一、排序

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]

SQL 复制 全屏

• [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] 表示返回外部键值。

回到顶部

二、事务

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 工具进行修复。

回到顶部

三、流水线（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);
        }
    }

回到顶部

四、发布订阅

继 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");
    }