[大数据]Redis事物机制实现缓存+DB一致性保障

一、前言

1.1、Redis事务基本概念

事务支持一次执行多个命令，一个事务中所有命令都会被序列化。在事务执行过程，会按照顺序串行化执行队列中的命令，其他客户端提交的命令请求不会插入到事务执行命令序列中

（redis事务就是一次性、顺序性、排他性的执行一个队列中的一组命令）。

?Redis的事务总是具有ACID中的原子性、一致性和隔离性。

1.2、Redis事务基础命令：

1. watch：?watch某个key,当该key被其它客户端改变时,则会中断当前的操作? ? ? ? ? ? ? ? ? ? ? ? ? （?带有WATCH命令的事务会将客户端和被监视的键在数据库的watched_keys字典关联，当键被修改时，程序会将所有监视被修改键的客户端的REDIS_DIRTY_CAS标识打开，服务只有在REDIS_DIRTY_CAS标识没有打开时，才会执行客户端提交的事务，否则服务器拒绝执行事务。）
2. multi：开启事物
3. exec：提交一组事物
4. discard：取消事物（Redis事务不支持回滚机制，只支持取消了）

Redis事务允许在一次单独的步骤中执行一组命令（Redis要么执行其中的所有命令，要么什么都不执行——原子性），Redis会将一个事务中的所有命令序列化，然后按顺序执行

（1）当某个客户端正在执行一次事务时，如果它在调用MULTI命令之前就从Redis服务端断开连接，那么就不会执行事务中的任何操作；

（2）如果它在调用EXEC命令之后才从Redis服务端断开连接，那么就会执行事务中的所有操作

二、demo1

 stringRedisTemplate.execute(new SessionCallback<Object>() {
        @SneakyThrows
        @Override
        @SuppressWarnings({ "unchecked"})
        public Object execute(@NotNull RedisOperations operations) throws DataAccessException {
          operations.watch(redisUsernameMap.keySet());
          operations.multi();
          try {
            redisUsernameMap.forEach((k, v) -> {
              // hash key操作
              operations.opsForHash().putAll(k, v);
            });
            // db操作
            xxMapper.batchBandUpdate(db);
            operations.exec();
          } catch (Exception e) {
            log.error(e.getMessage(), e);
            // 放弃执行
            operations.discard();
            throw new IoTApiException(
                AccessErrorCode.ACCESS_INFO_REDIS_DB_EXECUTE_ERROR.getErrorCode(),
                "redis db transaction consistency failed.");
          }
          return null;
        }
      });

三、demo2

      stringRedisTemplate.execute((RedisConnection connection) -> {
        byte[][] rawKeys = mqttAuthUtil.rawArrays(cacheKeysDeleting);
        connection.watch(rawKeys);
  
        connection.multi();
        try {
          connection.del(rawKeys);
          byte[] channel = mqttAuthUtil.rawStr(Constants.CACHE_CHANGE_CHANNEL);
          byte[] msg = xxx;
          // 发布/订阅操作
          connection.publish(channel, msg);
          // 提交一组事务
          connection.exec();
        } catch (Exception e) {
          log.error(e.getMessage(), e);
          connection.discard();
          throw new IotApiRuntimeException("redis db transaction consistency failed.");
        }
        return null;
      });