12. SDR 管道API-pipleline

喜欢ヅ旅行 2022-02-03 09:37 245阅读 0赞

## 1. Redis 的pipeline ##

### 1.1 什么是pipeline ###

Redis 中的pipeline 管道命令类似于批处理执行命令。 普通执行一次Redis 命令会进行一次TCP连接的创建与销毁, 执行多条Redis 命令就会进行多次TCP连接的创建与销毁， 而TCP 连接的创建与销毁时比较耗时。 采用pipeline技术, 是先将多条命令添加到队列中，然后创建一次TCP连接，将每次命令的执行结果(返回值为非void的命令)依次添加到list中，然后一次性返回，最后关闭TCP连接。

### 1.2 pipeline 特点 ###

*  一次pipeline 操作只创建和销毁一次TCP连接
 *  pipeline 中的多条命令存储在一个队列中，所以命令执行是有顺序的
 *  pipeline 中每条返回值不是void的命令都会返回一个结果，然后存储在List中. 纵使返回值为null
 *  pipeline 中返回值为void的命令, 不会返回结果, 也就是说list中不会有此对象的返回值.
 *  pipeline 中允许执行不同的命令

### 1.3 SDR API ###

SDR 提供了两中API来执行pipeline命令，区别在于对命令返回的结果进行反向序列化时，使用redisTemplate自身的valueSerializer，还是使用自定义的valueSerializer. 通常使用第一种方式即可.

<table> 
 <thead> 
  <tr> 
   <th align="left">API 签名</th> 
   <th align="left">方法描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">public List executePipelined(RedisCallback&lt;?&gt; action)</td> 
   <td align="left">执行pipeline命令，使用redisTemplate的valueSerializer对结果进行解析</td> 
  </tr> 
  <tr> 
   <td align="left">public List executePipelined(RedisCallback&lt;?&gt; action, @Nullable RedisSerializer&lt;?&gt; resultSerializer)</td> 
   <td align="left">执行pipleLine命令, 使用自定义的serializer 对结果集进行解析</td> 
  </tr> 
 </tbody> 
</table>

## 2. API 测试 ##

笔者构造四条命令来测试pipeline 的执行机制:

*  命令一返回为A， 所以list中添加第一个元素
 *  命令二返回为void，所以list中不添加
 *  命令三返回为null， 所以list中添加第二个元素，只不过值为null
 *  命令四返回为B，所以list中添加第三个元素

### 2.1 默认序列化方式执行pipeline ###

*  笔者使用stringRedisTemplate 开启pipeline , 所以pipeline 中value序列化为stringRedisTemplate的序列化方式, 即StringRedisSerializer.
 *  每个命令返回值都会使用StringRedisSerializer进行反向序列化。

@Test
    public void test() { 
    
        // 初始化数据10条数据
        ListOperations<String, String> ops = stringRedisTemplate.opsForList();
        ops.leftPush("queue", "{\"@class\":\"org.zongf.learn.sdr.l01.UserPO\",\"username\":\"zhangsan\",\"password\":\"123456\",\"age\":20}");
        ops.leftPush("queue", "{\"@class\":\"org.zongf.learn.sdr.l01.UserPO\",\"username\":\"zhangsan\",\"password\":\"123456\",\"age\":20}\n");
    
        // 管道符获取命令
        List<Object> list = stringRedisTemplate.executePipelined(new RedisCallback<Object>() { 
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException { 
                // 使用StringRedisTemplate, 可以将RedisConnection 转换为StringRedisConnection
                StringRedisConnection stringRedisConnection = (StringRedisConnection) connection;
    
                // 返回值为string
                stringRedisConnection.rPop("queue");
    
                // 返回值为void
                stringRedisConnection.slaveOfNoOne();
    
                // 返回值为null
                stringRedisConnection.rPop("no-exsists-queue");
    
                // 返回值为string
                stringRedisConnection.rPop("queue");
    
                return null;
            }
        });
    
        for (Object object : list) { 
            String clz = object != null ? object.getClass().toString() : null;
            System.out.println(clz + "--> " + object);
    
        }
    }

**输出结果:**

class java.lang.String--> {"@class":"org.zongf.learn.sdr.l01.UserPO","username":"zhangsan","password":"123456","age":20}
    null--> null
    class java.lang.String--> {"@class":"org.zongf.learn.sdr.l01.UserPO","username":"zhangsan","password":"123456","age":20}

### 2.2 自定义序列化方式 ###

*  笔者使用json 序列化工具反向解析每个命令的执行结果, 所以返回结果可以直接是List
 *  使用指定Serializer 方式API时, RedisConnection 不能再转换为StringRedisConnection

@Test
    public void test2() { 
    
        // 初始化数据10条数据
        ListOperations<String, String> ops = stringRedisTemplate.opsForList();
        ops.leftPush("queue", "{\"@class\":\"org.zongf.learn.sdr.l01.UserPO\",\"username\":\"zhangsan\",\"password\":\"123456\",\"age\":20}");
        ops.leftPush("queue", "{\"@class\":\"org.zongf.learn.sdr.l01.UserPO\",\"username\":\"zhangsan\",\"password\":\"123456\",\"age\":20}\n");
    
        // 管道符获取命令
        List<UserPO> list = redisTemplate.executePipelined(new RedisCallback<Object>() { 
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException { 
    
                // 返回值为string
                connection.rPop("queue".getBytes());
    
                // 返回值为void
                connection.slaveOfNoOne();
    
                // 返回值为null
                connection.rPop("no-exsists-queue".getBytes());
    
                // 返回值为string
                connection.rPop("queue".getBytes());
    
                return null;
            }
        }, new GenericJackson2JsonRedisSerializer());
    
        System.out.println("result:" + list);
    }

**输出结果:**

class org.zongf.learn.sdr.l01.UserPO--> UserPO{username='zhangsan', password='123456', age=20}
    null--> null
    class org.zongf.learn.sdr.l01.UserPO--> UserPO{username='zhangsan', password='123456', age=20}