Java 类redis.clients.jedis.Transaction 实例源码

@Test
public void transactionResponseWithError() {
  Transaction t = jedis.multi();
  t.set("foo", "bar");
  Response<Set<String>> error = t.smembers("foo");
  Response<String> r = t.get("foo");
  List<Object> l = t.exec();
  assertEquals(JedisDataException.class, l.get(1).getClass());
  try {
    error.get();
    fail("We expect exception here!");
  } catch (JedisDataException e) {
    // that is fine we should be here
  }
  assertEquals(r.get(), "bar");
}

@Test
public void execGetResponse() {
  Transaction t = jedis.multi();

  t.set("foo", "bar");
  t.smembers("foo");
  t.get("foo");

  List<Response<?>> lr = t.execGetResponse();
  try {
    lr.get(1).get();
    fail("We expect exception here!");
  } catch (JedisDataException e) {
    // that is fine we should be here
  }
  assertEquals("bar", lr.get(2).get());
}

@Test
public void select() {
  jedis.select(1);
  jedis.set("foo", "bar");
  jedis.watch("foo");
  Transaction t = jedis.multi();
  t.select(0);
  t.set("bar", "foo");

  Jedis jedis2 = createJedis();
  jedis2.select(1);
  jedis2.set("foo", "bar2");

  List<Object> results = t.exec();
  if(results.isEmpty()){
    results = null;
  }
  assertNull(results);
}

@Test
public void testResetStateWithFullyExecutedTransaction() {
  Jedis jedis2 = new Jedis(jedis.getClient().getHost(), jedis.getClient().getPort());
  jedis2.auth("foobared");

  Transaction t = jedis2.multi();
  t.set("mykey", "foo");
  t.get("mykey");

  List<Object> resp = t.exec();
  assertNotNull(resp);
  assertEquals(2, resp.size());

  jedis2.resetState();
  jedis2.close();
}

@Test
public void testCloseable() throws IOException {
  // we need to test with fresh instance of Jedis
  Jedis jedis2 = new Jedis(hnp.getHost(), hnp.getPort(), 500);
  jedis2.auth("foobared");

  Transaction transaction = jedis2.multi();
  transaction.set("a", "1");
  transaction.set("b", "2");

  transaction.close();

  try {
    transaction.exec();
    fail("close should discard transaction");
  } catch (JedisDataException e) {
    assertTrue(e.getMessage().contains("EXEC without MULTI"));
    // pass
  }
}

@Test
public void testExec() throws Exception {
  int rnd = ThreadLocalRandom.current().nextInt(0, presetElements.size());
  String key = presetElementKeys.get(rnd);
  String value = String.valueOf(presetElements.get(key));

  Transaction transaction = jedis.multi();

  Snapshot snapshot = commandTracker.snapshot();
  Snapshot discardSnapshot = discardTracker.snapshot();
  Snapshot txsnapshot = execTracker.snapshot();
  txsnapshot.increment();
  Response<Long> added = transaction.sadd(key, value);
  transaction.exec();
  assertEquals(1, (long) added.get());
  txsnapshot.validate();
  snapshot.validate();
  discardSnapshot.validate();
}

@Test
public void testDiscard() throws Exception {
  int rnd = ThreadLocalRandom.current().nextInt(0, presetElements.size());
  String key = presetElementKeys.get(rnd);
  String value = String.valueOf(presetElements.get(key));

  Transaction transaction = jedis.multi();

  Snapshot snapshot = commandTracker.snapshot();
  Snapshot discardSnapshot = discardTracker.snapshot();
  Snapshot txsnapshot = execTracker.snapshot();
  discardSnapshot.increment();
  Response<Long> added = transaction.sadd(key, value);
  transaction.discard();
  assertNull(jedis.get(key));
  txsnapshot.validate();
  snapshot.validate();
  discardSnapshot.validate();
}

@Test
public void testCloseDiscards() throws Exception {
  int rnd = ThreadLocalRandom.current().nextInt(0, presetElements.size());
  String key = presetElementKeys.get(rnd);
  String value = String.valueOf(presetElements.get(key));

  Transaction transaction = jedis.multi();

  Snapshot snapshot = commandTracker.snapshot();
  Snapshot discardSnapshot = discardTracker.snapshot();
  Snapshot txsnapshot = execTracker.snapshot();
  discardSnapshot.increment();
  Response<Long> added = transaction.sadd(key, value);
  transaction.close();
  assertNull(jedis.get(key));
  txsnapshot.validate();
  snapshot.validate();
  discardSnapshot.validate();
}

@Override
public boolean deleteMulti(String... keys) {
  Jedis jedis = null;
  try {
    jedis = pool.getResource();
    Transaction trans = jedis.multi();
    for (String key : keys) {
      trans.del(key.getBytes());
    }
    trans.exec();
    pool.returnResource(jedis);
  } catch (Exception e) {
    LOGGER.warn("Failed to delete key from cache {0}", keys, e);
    pool.returnBrokenResource(jedis);
    return false;
  }
  return true;
}

public Set<String> getAndRemZRange(String key, long score) {
  Jedis jedis = null;
  try {
    jedis = pool.getResource();
    Transaction trans = jedis.multi();
    trans.zrangeByScore(key.getBytes(), MIN_INF, SafeEncoder.encode(String.valueOf(score)));
    trans.zremrangeByScore(key.getBytes(), MIN_INF, SafeEncoder.encode(String.valueOf(score)));
    List<Object> response = trans.exec();
    Set<byte[]> data = (Set<byte[]>) response.get(0);
    Set<String> members = new LinkedHashSet<>(data.size());
    for (byte[] d : data) {
      members.add(new String(d));
    }
    pool.returnResource(jedis);
    return members;
  } catch (Exception e) {
    LOGGER.warn("Failed to get zrem keys from cache {0}:{1}", key, score, e);
    pool.returnBrokenResource(jedis);
    throw e;
  }
}

private void deleteItem0() {
    int selectedIndex = lstValues.getSelectionModel().getSelectedIndex();

    if (selectedIndex < 0) {
        return;
    }

    int listIndex = selectedIndex + currentFrom;

    try {
        JedisManager.withJedis(jedis -> {
            String toBeDeletedValue = "__TO_BE_DELETED/" + System.currentTimeMillis();
            jedis.watch(currentKey);
            Transaction transaction = jedis.multi();
            transaction.lset(currentKey, listIndex, toBeDeletedValue);
            transaction.lrem(currentKey, 1, toBeDeletedValue);
            transaction.exec();
        });
    } catch (Exception e) {
        LOG.error("delete failed", e);
        Fx.error(I18n.getString("title_op_fail"), I18n.getString("list_msg_op_failed"));
    }

    refreshList();
}

@Override
protected void doRelease(String token) {
    jedisClient.watch(lockName);

    String currentToken = jedisClient.get(lockName);
    if (currentToken == null){
        jedisClient.unwatch();
        return;
    }

    if (currentToken.equals(token)) {
        Transaction t = jedisClient.multi();
        t.del(lockName);
        t.exec();
    } else {
        jedisClient.unwatch();
    }
}

@Override
protected boolean doExtend(final String token, long additionalTime){
    jedisClient.watch(lockName);
    String currentToken = jedisClient.get(lockName);
    if (currentToken == null){
        jedisClient.unwatch();
        return false;
    }

    if (token.equals(currentToken)) {
        long expiration = jedisClient.pttl(lockName);
        if (expiration < 0) {
            jedisClient.unwatch();
            return false;
        }

        Transaction t = jedisClient.multi();
        t.pexpire(lockName, expiration + additionalTime);
        List response = t.exec();

        return (!response.isEmpty()) && ((Long)response.get(0) == 1);
    }

    jedisClient.unwatch();
    return false;
}

@Override
   public synchronized void updateQueuedJob(int jobId, JSONObject newJob) {
    try (Jedis jedis = jedisPool.getResource()) {
        jedis.select(redisDBIndex);
        if (jedis.zcard(queue) > 0) {
            Optional<Double> score = null;
            Set<Tuple> tuples = jedis.zrangeByScoreWithScores(queue, 0, Double.MAX_VALUE);
            String old = null;
            for(Tuple data : tuples) {
                JSONObject detail = JSONObject.parseObject(data.getElement());
                int id = detail.getIntValue(ScheduledConstants.JOB_ID);
                if(id == jobId) {
                    score = Optional.ofNullable(data.getScore());
                    old = data.getElement();
                    break;
                }
            }
            if(score.isPresent()){
                Transaction tx = jedis.multi();
                tx.zrem(queue, old);
                tx.zadd(queue, score.get(), newJob.toJSONString());
                tx.exec();
            }
        }
    }
}

@Override
public boolean append(final List<String> keyList, final List<String> valueList, final int seconds) {
    RedisUtil.checkList(keyList, valueList);

    return (Boolean) this.execute(new Invoker() {
        @Override
        public Object execute(Jedis jedis) {
            Transaction transaction = jedis.multi();
            for (int i = 0; i < keyList.size(); i++) {
                transaction.append(keyList.get(i), valueList.get(i));
                transaction.expire(keyList.get(i), seconds);
            }
            transaction.exec();
            return true;
        }
    });
}

@Override
public boolean set(final List<String> keyList, final List<String> valueList) {
    RedisUtil.checkList(keyList, valueList);

    return (Boolean) this.execute(new Invoker() {
        @Override
        public Object execute(Jedis jedis) {
            Transaction transaction = jedis.multi();
            for (int i = 0; i < keyList.size(); i++) {
                transaction.set(keyList.get(i), valueList.get(i));
            }
            transaction.exec();
            return true;
        }
    });

}

public void execute(TransactionCallBack rc) {
    Jedis jedis = null;
    Transaction transaction;
    List<Object> object = null;
    try {
        jedis = RedisPool.getJedis();
        transaction = jedis.multi();
        rc.execute(transaction);
        object = transaction.exec();
    } catch (Exception e) {
        log.error("执行redis操作异常" + e.getLocalizedMessage());
    }
    finally {
        RedisPool.returnJedis(jedis);
    }
}

/**
 * 方案一的坏处：
 * 假如在极端情况下，可能出现集群各个服务器同时执行到taskLockVersionOne，或者 doTask执行时间过长，
 * 在redis事物还没提交的时候，会出现同时有多台服务器执行doTask。
 * @param id
 */
private static void taskLockVersionOne(String id) {
    String key = "default_task_id";
    String value = jedis.get(key);
    // 用redis 事物是防止 在set成功之后 在执行doTask或者其他情况导致程序终止没有执行到transaction.expire（）
    // 导致单台机器一直占着锁会有单点事故
    Transaction transaction = null;
    try {
        transaction = jedis.multi();
        if (value == null) {
            transaction.set(key, id);
            doTask(id);
            // 设置过期是防止单点错误
            transaction.expire(key, 30);
        } else {
            if (value.equals(id)) {
                doTask(id);
            }
        }
    } catch (Exception e) {
        log.error("e" + e);
    } finally {
        transaction.exec();
    }
}

/**
 * 方案二：
 * 会重复两次回去jedis.get(key);
 * 同时如果expire时间小于doTask时间，也会出现同时有两个任务执行doTask情况。
 * @param id
 */
private static void taskLockVersionTwo(String id) {
    String key = "default_task_id";
    String value = jedis.get(key);
    // 用redis 事物是防止 在set成功之后 在执行doTask或者其他情况导致程序终止没有执行到transaction.expire（）
    // 导致单台机器一直占着锁会有单点事故
    Transaction transaction = null;
    try {
        transaction = jedis.multi();
        if (value == null) {
            transaction.set(key, id);
            transaction.expire(key, 30);
        }
    } catch (Exception e) {
        log.error("e" + e);
    } finally {
        transaction.exec();
    }
    value = jedis.get(key);
    if (value.equals(id)) {
        doTask(id);
    }
}

@Test
public void transactionResponseWithError() {
  Transaction t = jedis.multi();
  t.set("foo", "bar");
  Response<Set<String>> error = t.smembers("foo");
  Response<String> r = t.get("foo");
  List<Object> l = t.exec();
  assertEquals(JedisDataException.class, l.get(1).getClass());
  try {
    error.get();
    fail("We expect exception here!");
  } catch (JedisDataException e) {
    // that is fine we should be here
  }
  assertEquals(r.get(), "bar");
}

@Test
public void execGetResponse() {
  Transaction t = jedis.multi();

  t.set("foo", "bar");
  t.smembers("foo");
  t.get("foo");

  List<Response<?>> lr = t.execGetResponse();
  try {
    lr.get(1).get();
    fail("We expect exception here!");
  } catch (JedisDataException e) {
    // that is fine we should be here
  }
  assertEquals("bar", lr.get(2).get());
}

@Test
public void select() {
  jedis.select(1);
  jedis.set("foo", "bar");
  jedis.watch("foo");
  Transaction t = jedis.multi();
  t.select(0);
  t.set("bar", "foo");

  Jedis jedis2 = createJedis();
  jedis2.select(1);
  jedis2.set("foo", "bar2");

  List<Object> results = t.exec();

  assertNull(results);
}

@Test
public void testResetStateWhenInWatch() {
  jedis.watch("mykey", "somekey");

  // state reset : unwatch
  jedis.resetState();

  Transaction t = jedis.multi();

  nj.connect();
  nj.auth("foobared");
  nj.set("mykey", "bar");
  nj.disconnect();

  t.set("mykey", "foo");
  List<Object> resp = t.exec();
  assertNotNull(resp);
  assertEquals(1, resp.size());
  assertEquals("foo", jedis.get("mykey"));
}

@Test
public void testResetStateWithFullyExecutedTransaction() {
  Jedis jedis2 = new Jedis(jedis.getClient().getHost(), jedis.getClient().getPort());
  jedis2.auth("foobared");

  Transaction t = jedis2.multi();
  t.set("mykey", "foo");
  t.get("mykey");

  List<Object> resp = t.exec();
  assertNotNull(resp);
  assertEquals(2, resp.size());

  jedis2.resetState();
  jedis2.close();
}

@Test
public void testCloseable() throws IOException {
  // we need to test with fresh instance of Jedis
  Jedis jedis2 = new Jedis(hnp.getHost(), hnp.getPort(), 500);
  jedis2.auth("foobared");

  Transaction transaction = jedis2.multi();
  transaction.set("a", "1");
  transaction.set("b", "2");

  transaction.close();

  try {
    transaction.exec();
    fail("close should discard transaction");
  } catch (JedisDataException e) {
    assertTrue(e.getMessage().contains("EXEC without MULTI"));
    // pass
  }
}

public int nextId(String name) {
    String key = keyFor(name);
    Holder<Integer> holder = new Holder<>();
    redis.execute(jedis -> {
        jedis.sadd(ALL_SEQS, name);
        long pos = 0;
        do {
            jedis.watch(key);
            pos = jedis.bitpos(key, false);
            if (pos < 0) {
                pos = 0;
            }
            Transaction tx = jedis.multi();
            tx.setbit(key, pos, true);
            if (tx.exec() != null) {
                break;
            }
        } while (true);
        holder.set((int) pos);
    });
    return holder.value();
}

public static void main(String[] args) throws InterruptedException {
    ApplicationContext ac = new ClassPathXmlApplicationContext("root-context.xml");
    JedisPool pool = (JedisPool) ac.getBean("jedisPool");
    int count = 0;
    while (true) {
        Jedis jedis = pool.getResource();
        jedis.get("keyName");
        Transaction tx = jedis.multi();
        tx.incr("keyName");
        tx.expire("keyName", 10);
        tx.exec();  
        jedis.close();
        System.out.println("jedisPool active count:" + pool.getNumActive());
        System.out.println("jedisPool idle count:" + pool.getNumIdle());
        Thread.sleep(10);
    }

}

@Override
public Map<String, Response<Long>> setJedisMultiCommand(Transaction jedisMulti) {

    Response<Long> applicationDailyRateLimit =
            jedisMulti.hincrBy(Const.REDIS_APP_RATELIMIT_DAILY, this.requestInfo.getAppId(), -1);
    Response<Long> applicationMinutelyRateLimit =
            jedisMulti.hincrBy(Const.REDIS_APP_RATELIMIT_MINUTELY, this.requestInfo.getAppId(), -1);
    Response<Long> applicationDailyRateLimitTTL = jedisMulti.ttl(Const.REDIS_APP_RATELIMIT_DAILY);
    Response<Long> applicationMinutelyRateLimitTTL = jedisMulti.ttl(Const.REDIS_APP_RATELIMIT_MINUTELY);

    Map<String, Response<Long>> appRatelimit = Maps.newHashMap();
    appRatelimit.put(Const.REDIS_APP_RATELIMIT_DAILY, applicationDailyRateLimit);
    appRatelimit.put(Const.REDIS_APP_RATELIMIT_MINUTELY, applicationMinutelyRateLimit);
    appRatelimit.put(Const.REDIS_APP_RATELIMIT_DAILY_TTL, applicationDailyRateLimitTTL);
    appRatelimit.put(Const.REDIS_APP_RATELIMIT_MINUTELY_TTL, applicationMinutelyRateLimitTTL);

    return appRatelimit;

}

public void testTrans() {// 0.304秒
    Jedis jedis = new Jedis("120.25.241.144", 6379);
    jedis.auth("b840fc02d52404542994");

    long start = System.currentTimeMillis();
    Transaction tx = jedis.multi();
    for (int i = 0; i < 1000; i++) {
        tx.set("n" + i, "n" + i);
        System.out.println(i);
    }
    tx.exec();
    long end = System.currentTimeMillis();
    System.out.println("共花费：" + (end - start) / 1000.0 + "秒");

    jedis.disconnect();
    try {
        Closeables.close(jedis, true);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

@Override
public void storeModel(String modelName, String id, String object) throws Exception{

       try(Jedis jedis = pool.getResource()){

        Map<String, String> map = new HashMap<>();
        map.put(id, object);

        String hm_key = createKey(table_perfix, modelName);

        while(true){
            jedis.watch(hm_key);
            Transaction transaction = jedis.multi();
            transaction.hmset(hm_key, map);
            if(transaction.exec() != null)
                break;
        }

       }

}

public <T> int drainTo(String key, Collection<T> c, int maxElements, Class<T> classNm) {
    if (StringUtils.isEmpty(key)) {
        key = setKey;
    }
    String listKey = service.sPop(key, String.class);
    if (StringUtils.isEmpty(listKey)) {
        return 0;
    }
    INaviMultiRedis multi = service.multi(key);
    try {
        Transaction tran = multi.getTransaction();
        Response<List<byte[]>> response = tran.lrange(listKey.getBytes(), 0, maxElements - 1);
        tran.ltrim(listKey, maxElements, -1);
        tran.exec();
        List<byte[]> results = response.get();
        for (byte[] result : results) {
            c.add(jsonSerializer.getObjectFromBytes(result, classNm));
        }
        if (results.size() < maxElements) {
            removeKey(key, listKey);
        }
        return c.size();
    } finally {
        multi.returnObject();
    }
}

public <T> int drainTo(String key, Collection<T> c, int maxElements, Class<T> classNm) {
    INaviMultiRedis multi = service.multi(key);
    try {
        Transaction tran = multi.getTransaction();
        Response<List<byte[]>> response = tran.lrange(key.getBytes(), 0, maxElements - 1);
        tran.ltrim(key, maxElements, -1);
        tran.exec();
        List<byte[]> results = response.get();
        AlibabaJsonSerializer jsonSerializer = new AlibabaJsonSerializer();
        for (byte[] result : results) {
            c.add(jsonSerializer.getObjectFromBytes(result, classNm));
        }
        return c.size();
    } finally {
        multi.returnObject();
    }
}

/**
 * 设置set类型的值
 * 
 * @param key
 * @param member
 * @return 1 if the new element was added 0 if the element was already a
 *         member of the set
 */
public Long sSet(final String key, final String[] members) {
    if (RedisConstants.isInUse()) {
        return getTemplate().execute(new JedisCallBackHandler<Long>() {
            public Long invoke(Jedis jedis) {
                Transaction trans = jedis.multi();
                trans.del(key);
                Long ret = jedis.sadd(key, members);
                trans.exec();
                return ret;
            }
        });
    } else {
        return null;
    }
}

/**
 * 设置set类型的值
 * 
 * @param key
 * @param member
 * @return 1 if the new element was added 0 if the element was already a
 *         member of the set
 */
public Long sSet(final String key, final String[] members) {
    if (RedisConstants.isInUse()) {
        return getTemplate().executeWrite(new JedisCallBackHandler<Long>() {
            public Long invoke(Jedis jedis) {
                Transaction trans = jedis.multi();
                trans.del(key);
                Long ret = jedis.sadd(key, members);
                trans.exec();
                return ret;
            }
        });
    } else {
        return null;
    }
}

public Transaction multi(long logIndex) {
    String flag = getClassName() + ".multi";

    Transaction res = null;
    Jedis jds = null;
    try {
        jds = jedisPool.getResource();
        res = jds.multi();
        jedisPool.returnResource(jds);
        return res;
    } catch (Exception e) {
        jedisPool.returnBrokenResource(jds);
        FRCLogger.getInstance().warn(logIndex, flag, "exception", e);
        res = null;
        return res;
    }

}

public List<Object> exec(long logIndex, Transaction transaction) {
    String flag = getClassName() + ".exec";

    List<Object> res = null;
    Jedis jds = null;
    try {
        jds = jedisPool.getResource();
        res = transaction.exec();
        jedisPool.returnResource(jds);
        return res;
    } catch (Exception e) {
        jedisPool.returnBrokenResource(jds);
        FRCLogger.getInstance().warn(logIndex, flag, "exception:", e);
        res = null;
        return res;
    }
}

@Test
public void transactionResponseWithError() {
  Transaction t = jedis.multi();
  t.set("foo", "bar");
  Response<Set<String>> error = t.smembers("foo");
  Response<String> r = t.get("foo");
  List<Object> l = t.exec();
  assertEquals(JedisDataException.class, l.get(1).getClass());
  try {
    error.get();
    fail("We expect exception here!");
  } catch (JedisDataException e) {
    // that is fine we should be here
  }
  assertEquals(r.get(), "bar");
}

@Test
public void execGetResponse() {
  Transaction t = jedis.multi();

  t.set("foo", "bar");
  t.smembers("foo");
  t.get("foo");

  List<Response<?>> lr = t.execGetResponse();
  try {
    lr.get(1).get();
    fail("We expect exception here!");
  } catch (JedisDataException e) {
    // that is fine we should be here
  }
  assertEquals("bar", lr.get(2).get());
}

@Test
public void select() {
  jedis.select(1);
  jedis.set("foo", "bar");
  jedis.watch("foo");
  Transaction t = jedis.multi();
  t.select(0);
  t.set("bar", "foo");

  Jedis jedis2 = createJedis();
  jedis2.select(1);
  jedis2.set("foo", "bar2");

  List<Object> results = t.exec();

  assertNull(results);
}

@Test
public void testResetStateWhenInWatch() {
  jedis.watch("mykey", "somekey");

  // state reset : unwatch
  jedis.resetState();

  Transaction t = jedis.multi();

  nj.connect();
  nj.auth("foobared");
  nj.set("mykey", "bar");
  nj.disconnect();

  t.set("mykey", "foo");
  List<Object> resp = t.exec();
  assertNotNull(resp);
  assertEquals(1, resp.size());
  assertEquals("foo", jedis.get("mykey"));
}