Redis利用管道技术+事务实现数据高效批量处理

小咪咪 2022-12-07 15:23 311阅读 0赞

**目录**

Redis 管道技术

Redis管道实例

管道技术的优势

Redis事务

事务操作实例

Redis 事务命令

Java实现管道+事务提交批量处理

RedisCallback

管道调用API

--------------------

Redis是一种基于客户端-服务端模型以及请求/响应协议的TCP服务。这意味着通常情况下一个请求会遵循以下步骤：

*  客户端向服务端发送一个查询请求，并监听Socket返回，通常是以阻塞模式，等待服务端响应。
 *  服务端处理命令，并将结果返回给客户端。

# Redis 管道技术 #

Redis 管道技术可以在服务端未响应时，客户端可以继续向服务端发送请求，并最终一次性读取所有服务端的响应。

## Redis管道实例 ##

查看 redis 管道，只需要启动 redis 实例并输入以下命令：

$(echo -en "PING\r\n SET w3ckey redis\r\nGET w3ckey\r\nINCR visitor\r\nINCR visitor\r\nINCR visitor\r\n"; sleep 10) | nc localhost 6379
     
    +PONG
    +OK
    redis
    :1
    :2
    :3

以上实例中我们通过使用 **PING** 命令查看redis服务是否可用， 之后我们们设置了 w3ckey 的值为 redis，然后我们获取 w3ckey 的值并使得 visitor 自增 3 次。

在返回的结果中我们可以看到这些命令一次性向 redis 服务提交，并最终一次性读取所有服务端的响应

--------------------

## 管道技术的优势 ##

管道技术最显著的优势是提高了 redis 服务的性能。

在下面的测试中，我们将使用Redis的Ruby客户端，支持管道技术特性，测试管道技术对速度的提升效果。示例：

require 'rubygems' 
    require 'redis'
    def bench(descr) 
    start = Time.now 
    yield 
    puts "#{descr} #{Time.now-start} seconds" 
    end
    def without_pipelining 
    r = Redis.new 
    10000.times { 
            r.ping 
    } 
    end
    def with_pipelining 
    r = Redis.new 
    r.pipelined { 
            10000.times { 
                    r.ping 
            } 
    } 
    end
    bench("without pipelining") { 
            without_pipelining 
    } 
    bench("with pipelining") { 
            with_pipelining 
    }

从处于局域网中的Mac OS X系统上执行上面这个简单脚本的数据表明，开启了管道操作后，往返时延已经被改善得相当低了。

without pipelining 1.185238 seconds 
    with pipelining 0.250783 seconds

如你所见，开启管道后，我们的速度效率提升了5倍。

# Redis事务  #

Redis 事务可以一次执行多个命令， 并且带有以下两个重要的保证：

*  事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。
 *  事务是一个原子操作：事务中的命令要么全部被执行，要么全部都不执行。

一个事务从开始到执行会经历以下三个阶段：

*  开始事务。
 *  命令入队。
 *  执行事务。

## 事务操作实例 ##

以下是一个事务的例子， 它先以 **MULTI** 开始一个事务， 然后将多个命令入队到事务中， 最后由 **EXEC** 命令触发事务， 一并执行事务中的所有命令：

redis 127.0.0.1:6379> MULTI
    OK
     
    redis 127.0.0.1:6379> SET book-name "Mastering C++ in 21 days"
    QUEUED
     
    redis 127.0.0.1:6379> GET book-name
    QUEUED
     
    redis 127.0.0.1:6379> SADD tag "C++" "Programming" "Mastering Series"
    QUEUED
     
    redis 127.0.0.1:6379> SMEMBERS tag
    QUEUED
     
    redis 127.0.0.1:6379> EXEC
    1) OK
    2) "Mastering C++ in 21 days"
    3) (integer) 3
    4) 1) "Mastering Series"
       2) "C++"
       3) "Programming"

--------------------

## Redis 事务命令 ##

下表列出了 redis 事务的相关命令：

<table> 
 <tbody> 
  <tr> 
   <th>序号</th> 
   <th>命令及描述</th> 
  </tr> 
  <tr> 
   <td>1</td> 
   <td><a href="https://www.redis.net.cn/order/3638.html" rel="nofollow">DISCARD</a>&nbsp;取消事务，放弃执行事务块内的所有命令。</td> 
  </tr> 
  <tr> 
   <td>2</td> 
   <td><a href="https://www.redis.net.cn/order/3639.html" rel="nofollow">EXEC</a>&nbsp;执行所有事务块内的命令。</td> 
  </tr> 
  <tr> 
   <td>3</td> 
   <td><a href="https://www.redis.net.cn/order/3640.html" rel="nofollow">MULTI</a>&nbsp;标记一个事务块的开始。</td> 
  </tr> 
  <tr> 
   <td>4</td> 
   <td><a href="https://www.redis.net.cn/order/3641.html" rel="nofollow">UNWATCH</a>&nbsp;取消 WATCH 命令对所有 key 的监视。</td> 
  </tr> 
  <tr> 
   <td>5</td> 
   <td><a href="https://www.redis.net.cn/order/3642.html" rel="nofollow">WATCH key [key ...]</a>&nbsp;监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。</td> 
  </tr> 
 </tbody> 
</table>

#  Java实现管道+事务提交批量处理 #

## RedisCallback ##

redis管道执行的回调方法，有点像Java 多线程Callback接口，不过回调并不需要我们自己处理：

/**
     * @Copyright: 2019-2021
     * @FileName: MessageStatusRedisCallback .java
     * @Author: PJL
     * @Date: 2020/6/5 14:25
     * @Description: 通知公告状态redis指令回调类
     */
    public class MessageStatusRedisCallback implements RedisCallback<Object> {
    
        Logger logger = LoggerUtils.getLogger(LoggerUtils.PatrolLoggerType.REDIS);
    
        List<MessageStatus> messageStatusList;
    
        /**
         * 构造方法
         *
         * @param messageStatusList
         */
        public MessageStatusRedisCallback(List<MessageStatus> messageStatusList) {
            this.messageStatusList = messageStatusList;
        }
    
        @Override
        public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
            // 验证指令执行方式
            if (ObjectUtils.isNotEmpty(messageStatusList)) {
                long s = System.currentTimeMillis();
                String json = null;
                try {
                    // 开启REDIS事务
                    redisConnection.multi();
                    for (MessageStatus messageStatus : messageStatusList) {
                        json = JSONObject.toJSONString(messageStatus);
                        /* redisTemplate.opsForHash().put(Constants.MOBILE_HASH_MESSAGE_KEY, userId, Boolean.toString(hasMessage));*/
                        redisConnection.hSet(Constants.MOBILE_HASH_MESSAGE_KEY.getBytes(), messageStatus.getUserId().getBytes(), Boolean.toString(messageStatus.isHasMessage()).getBytes());
                    }
                    // 提交事务批量执行
                    redisConnection.exec();
                } catch (Exception e) {
                    // 回滚事务不执行
                    redisConnection.discard();
                    logger.error("【回滚事务不执行】出问题的用户通知公告状态JSON:{}", json);
                }
                long e = System.currentTimeMillis();
                logger.info("pipeline 批量执行用户通知公告 status命令完成....耗时:{} {}", (e - s), "ms");
            }
            return null;
        }
    }

## 管道调用API ##

使用RedisTemplate提供的方法进行调用：

/**
     * @Copyright: 2019-2021
     * @FileName: MessageStatusRedisPipelineService.java
     * @Author: PJL
     * @Date: 2020/6/5 14:19
     * @Description: 通知公告redis pipeline执行命令服务
     */
    @Service
    public class MessageStatusRedisPipelineService {
    
        @Qualifier("redisTemplateByLettuce")
        @Autowired
        RedisTemplate redisTemplate;
    
        /**
         * 设置用户通知公告状态记录缓存到REDIS
         *
         * @param messageStatusList
         */
        public void saveMessageStatusToRedis(List<MessageStatus> messageStatusList) {
            redisTemplate.executePipelined(new MessageStatusRedisCallback(messageStatusList));
        }
    }

至此，我们就实现了一个机遇Pipeline管道技术和事务的批量操作。

参考阅读：

*  [redis事务][redis]
 *  [redis管道技术][redis 1]

[redis]: https://www.redis.net.cn/tutorial/3515.html
[redis 1]: https://www.redis.net.cn/tutorial/3523.html