NoSQL之Redis---事务（transaction）Java实现

[不忘初心]

前文中，我们介绍了通过命令行的方式调用redis命令实现事务。本文我们来介绍一下使用Java来实现redis。惯例，我们先看看准备工作有哪些：

a.操作系统：windows 7 x64

b.其他软件：eclipse mars，jdk7，redis 2.8.19，jedis 2.8

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

1.通过maven建立redis-transaction工程，结构图如下：



2.修改pom.xml文件，具体内容如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion>

<groupId>com.csdn.ingo</groupId>
<artifactId>redis-transaction</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>

<name>redis-transaction</name>
<url>http://maven.apache.org</url>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>

<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.8.0</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
</dependencies>
</project>

3.在工程中，我们使用输出流代替日志，各位看官可以根据需要加入日志配置文件即可。具体请参考前文订阅与发布的java实现。

4.启动redis服务器。【对于Redis多端口，可以进入redis目录，根据help命令提示开启多个Redis服务器】

5.新建JunitTest.java文件，具体内容如下：

package com.csdn.ingo.redis_transaction;
import org.junit.Test;

import redis.clients.jedis.Jedis;
public class JunitTest {
//本例为不使用redis事务的jedis命令调用方式
@Test
public void normalTest() {
Jedis jedis = new Jedis("localhost");
long startTime = System.currentTimeMillis();
for (int i = 0; i < 1000; i++) {
jedis.set("key-nor-" + i, "value-nor" + i);
System.out.println(jedis.get("key-nor-" + i));
}
long endTime = System.currentTimeMillis();
System.out.println("normal cost:" + (endTime - startTime) + "ms");
jedis.disconnect();
}
}

6.通过调用事务命令执行操作，单元测试用例如下：

@Test
public void transactionTest() {
Jedis jedis = new Jedis("localhost");
long startTime = System.currentTimeMillis();
Transaction tx = jedis.multi();

for (int i = 0; i < 1000; i++) {
tx.set("key-" + i, "value-" + i);
}
for (int i = 0; i < 1000; i++) {
tx.get("key-" + i);
}
List<Object> resultList = tx.exec();
for (int i = 0; i < resultList.size(); i++) {
System.out.println(resultList.get(i).toString());
}
long endTime = System.currentTimeMillis();
System.out.println("transaction cost:" + (endTime - startTime) + "ms");
jedis.disconnect();
System.out.println("------------------------------------------");
}

7.在上面两个的单元测试中，我们循环调用了1000次set命令，1000次get命令。在redis底层实现中，每次命令的调用都会执行一次redis的server与cli的链接与释放。因此，对于批量执行的命令，最好使用pipeline方法进行使用。所谓的pipeline方法可以理解为：多条命令只执行一次链接的建立与释放，其接受命令和返回结果都是集合的形式进行传输。具体的性能提升请参考文章末尾的对比。

@Test
public void pipelineTest() {
Jedis jedis = new Jedis("localhost");
Pipeline pipeline = jedis.pipelined();
long startTime = System.currentTimeMillis();
for (int i = 0; i < 1000; i++) {
pipeline.set("pipeline-key-" + i, "value-" + i);
}
List<Object> resultList = pipeline.syncAndReturnAll();
long endTime = System.currentTimeMillis();
System.err.println("pipeline cost:" + (endTime - startTime) + "ms");
jedis.disconnect();
}

8.在第七步中，我们使用pipeline的方式执行了多条redis命令，与此同时，我们可能还需要通过事务的方式进行。因此，我们修改上文的单元测试方法，如下：

@Test
public void transactionInPipelineTest() {
Jedis jedis = new Jedis("localhost");
Pipeline pipeline = jedis.pipelined();
long startTime = System.currentTimeMillis();
pipeline.multi();
for (int i = 0; i < 1000; i++) {
pipeline.set("pipeline-key-" + i, "value-" + i);
}
pipeline.exec();
List<Object> resultList = pipeline.syncAndReturnAll();
long endTime = System.currentTimeMillis();
System.out.println("transaction in pipeline cost:" + (endTime - startTime) + "ms");
jedis.disconnect();

}

9.分布式的直连调用：在某些情况下，我们可能需要在多个redis上同时执行命令。此时，我们通过如下的方式进行使用：

@Test
public void multiJedisShardTest() {
List<JedisShardInfo> shards = Arrays.asList(new JedisShardInfo("localhost", 6379),
new JedisShardInfo("localhost", 6380));
ShardedJedis sharding = new ShardedJedis(shards);
long startTime = System.currentTimeMillis();
for (int i = 0; i < 1000; i++) {
sharding.set("multiJedis-key-" + i, "value-" + i);
}
long endTime = System.currentTimeMillis();
System.out.println("multi JedisShardTest cost:" + (endTime - startTime) + "ms");
sharding.disconnect();
}

其执行结果如下：【两个redis服务器中都出现了我们set的值。其中，测试时6379中共计520条数据，6380种共计480条数据】





10.与上面对应的，分布式的异步调用如下：即通过pipeline进行：

@Test
public void multiJedisShardInPipelineTest() {
List<JedisShardInfo> shards = Arrays.asList(new JedisShardInfo("localhost", 6379),
new JedisShardInfo("localhost", 6380));
ShardedJedis sharding = new ShardedJedis(shards);
ShardedJedisPipeline pipeline = sharding.pipelined();
long startTime = System.currentTimeMillis();
for (int i = 0; i < 1000; i++) {
pipeline.set("multiJedis-key-" + i, "value-" + i);
}
List<Object> resultList = pipeline.syncAndReturnAll();
long endTime = System.currentTimeMillis();
System.out.println("multi JedisShardTest In Pipeline cost:" + (endTime - startTime) + "ms");
sharding.disconnect();
}

11.线程池同步调用：上面直连方式是非线程安全的，因此，对于分布式线程中调用的情景，就需要使用连接池进行调用。具体如下：

@Test
public void multiJedisInThreadTest(){
List<JedisShardInfo> shards = Arrays.asList(new JedisShardInfo("localhost", 6379),
new JedisShardInfo("localhost", 6380));
ShardedJedisPool pool = new ShardedJedisPool(new JedisPoolConfig(), shards);
ShardedJedis jedis = pool.getResource();
long startTime = System.currentTimeMillis();
for (int i = 0; i < 1000; i++) {
jedis.set("multiJedis-key-" + i, "value-" + i);
}
long endTime = System.currentTimeMillis();
pool.close();
System.out.println("multi JedisShardTest In Thread cost:" + (endTime - startTime) + "ms");
pool.destroy();
}

12.线程池异步调用：与上例对应的，使用pipeline实现异步执行，具体如下：

@Test
public void multiJedisInThreadPipelineTest(){
List<JedisShardInfo> shards = Arrays.asList(new JedisShardInfo("localhost", 6379),
new JedisShardInfo("localhost", 6380));
ShardedJedisPool pool = new ShardedJedisPool(new JedisPoolConfig(), shards);
ShardedJedis jedis = pool.getResource();
ShardedJedisPipeline pipeline = jedis.pipelined();
long startTime = System.currentTimeMillis();
for (int i = 0; i < 1000; i++) {
pipeline.set("multiJedis-key-" + i, "value-" + i);
}
List<Object> results = pipeline.syncAndReturnAll();
long endTime = System.currentTimeMillis();
pool.close();
System.out.println("multi JedisShardTest In Thread Pipeline cost:" + (endTime - startTime) + "ms");
pool.destroy();
}

13.性能对比：【特别声明，为了尽量减少差异项，以下数据都是我们移除了工程中所有的system.out后的运行时间】

normal cost:197ms

transaction cost:73ms

pipeline cost:49ms

transaction in pipeline cost:55ms

multi JedisShardTest cost:217ms

multi JedisShardTest In Pipeline cost:67ms

multi JedisShardTest In Thread cost:197ms

multi JedisShardTest In Thread Pipeline cost:63ms

14.注意事项：

a.对于异步的调用，不同使用同步的方法。如下面的代码，其在具体使用某个服务时，一定是错误的。



其对应的执行结果如下：



我们看到，命令实际返回的结果不是命令正常的返回结果。因此，如果在此时我们使用同步的方式进行调用时，一定会导致错误出现。同样的，在pipeline中，也是禁止这种使用方式的。

b.通过上面的性能对比，在实际情境中，最好能够选择适当的方式来优化redis的性能。

c.在分布式的redis连接池情景下，redis不支持食物，原因是每次执行命令的服务器可能不同，因此无法保证其事务执行。

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

至此，NoSQL之Redis---事务（transaction）Java实现 结束

参考资料：

至此，NoSQL之Redis---事务（transaction）命令

参考资料：

redis官网：redis.io
其他资料： http://doc.redisfans.com/ http://my.oschina.net/sphl520/blog/312514#navbar-header