java 调用 Redis 的八种方式：普通、事务、管道、管道调用事务、分布式直连同步(异步)调用、分布式连接池同步(异步)调用

本是古典何须时尚 2022-02-16 01:21 213阅读 0赞

开始前需要准备两个jar包，当然运行前还需要开启redis服务（redis-server.exe redis.windows.conf）就不详细说了。

**[Apache Commons Pool][]**

**[Jedis][]**

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

### 一、普通同步 ###

基础的调用方式，每次`set`之后都可以返回结果，标记是否成功。

@Test
    public void NormalSyncTest() {
        Jedis jedis = new Jedis("localhost");
    
        long start = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) {
        String res = jedis.set("s" + i, "s" + i);
        }
        long end = System.currentTimeMillis();
        System.out.println("Simple SET: " + ((end - start) / 1000.0) + " seconds");
    
        jedis.disconnect();
    }

### 二、事务方式 ###

主要目的是保障，一个client发起的事务中的命令可以连续的执行，而中间不会插入其他client的命令。

调用jedis.watch(…)方法来监控key，如果调用后key值发生变化，则整个事务会执行失败.。

@Test
    public void TransactionsTest() {
        Jedis jedis = new Jedis("localhost");
    
        long start = System.currentTimeMillis();
        Transaction tx = jedis.multi(); // 创建事务
        for (int i = 0; i < 100000; i++) {
            tx.set("t" + i, "t" + i);
        }
        List<Object> res = tx.exec();
        long end = System.currentTimeMillis();
        System.out.println("Transaction SET: " + ((end - start) / 1000.0) + " seconds");
    
        jedis.disconnect();
    }

### 三、管道技术 ###

一次性发送多条命令并在执行完后一次性将结果返回，pipeline通过减少客户端与redis的通信次数来实现降低往返延时时间。

@Test
    public void PipeliningTest() {
        Jedis jedis = new Jedis("localhost");
    
        long start = System.currentTimeMillis();
        Pipeline pipeline = jedis.pipelined();// new了一个Pipeline对象，并设置了该对象的client属性
        for (int i = 0; i < 100000; i++) {
            pipeline.set("p" + i, "p" + i);// set方法实际调用的是Pipelinebase类（Pipeline父类的父类）的set方法
        }
        List<Object> res = pipeline.syncAndReturnAll();// 把每一次的请求操作加入到一个Queue里面
        long end = System.currentTimeMillis();
        System.out.println("Pipelined SET: " + ((end - start) / 1000.0) + " seconds");
    
        jedis.disconnect();
    
    }

### 四、管道中调用事务 ###

其效率和单独使用事务差不多，甚至还略微差点。 事务和管道都是异步的，在管道中再进行事务调用，没有必要，不如直接进行事务模式。

@Test
    public void PipeliningTranTest() {
        Jedis jedis = new Jedis("localhost");
    
        long start = System.currentTimeMillis();
        Pipeline pipeline = jedis.pipelined();// new了一个Pipeline对象，并设置了该对象的client属性。
        pipeline.multi();
        for (int i = 0; i < 100000; i++) {
            pipeline.set("px" + i, "px" + i);// set方法实际调用的是Pipelinebase类（Pipeline父类的父类）的set方法
        }
        pipeline.exec();
        List<Object> res = pipeline.syncAndReturnAll();// 把每一次的请求操作加入到一个Queue里面
        long end = System.currentTimeMillis();
        System.out.println("Pipelined transaction: " + ((end - start) / 1000.0) + " seconds");
        
        jedis.disconnect();
    }

### 五、分布式直连同步调用 ###

分布式直接连接，并且是同步调用，每步执行都返回执行结果。

@Test
    public void ShardNormalTest() {
        List<JedisShardInfo> shards = Arrays.asList(
                    new JedisShardInfo("localhost", 6379),
                    new JedisShardInfo("localhost", 6379));
    
        ShardedJedis sharding = new ShardedJedis(shards);
    
        long start = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) {
            String res = sharding.set("sn" + i, "sn" + i);
        }    
        long end = System.currentTimeMillis();
        System.out.println("Simple@Sharing SET: " + ((end - start) / 1000.0) + " seconds");
    
        sharding.disconnect();
    }

### 六、分布式直连异步调用 ###

分布式调用中不支持事务。 因为事务是在服务器端实现，而在分布式中，每批次的调用对象都可能访问不同的机器，所以，没法进行事务。

@Test
    public void ShardSyncPipelinedTest() {
        List<JedisShardInfo> shards = Arrays.asList(
                    new JedisShardInfo("localhost", 6379),
                    new JedisShardInfo("localhost", 6379)); // 使用相同的ip:port,仅作测试
    
        ShardedJedis sharding = new ShardedJedis(shards);
    
        ShardedJedisPipeline pipeline = sharding.pipelined();
    
        long start = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) {
            pipeline.set("sp" + i, "sp" + i);
        }
        List<Object> res = pipeline.syncAndReturnAll();
        long end = System.currentTimeMillis();
        System.out.println("Pipelined@Sharing SET: " + ((end - start) / 1000.0) + " seconds");
    
        sharding.disconnect();
    }

### 七、分布式连接池同步调用 ###

分布式调用代码是运行在线程中，那么上面两个直连调用方式就不合适了，因为直连方式是非线程安全的，这个时候，你就必须选择连接池调用。

@Test
    public void ShardSimplePoolTest() {
        List<JedisShardInfo> shards = Arrays.asList(
                    new JedisShardInfo("localhost", 6379),
                    new JedisShardInfo("localhost", 6379)); // 使用相同的ip:port,仅作测试
    
        ShardedJedisPool pool = new ShardedJedisPool(new JedisPoolConfig(), shards);
    
        ShardedJedis sJ = pool.getResource();
        
        long start = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) {
            String res = sJ.set("spn" + i, "spn" + i);
        }
        long end = System.currentTimeMillis();
        pool.returnResource(sJ);
        System.out.println("Simple@Pool SET: " + ((end - start) / 1000.0) + " seconds");
        
        pool.destroy();
    }

### 八、分布式连接池异步调用 ###

分布式中，连接池方式调用不但线程安全外，根据上面的测试数据，也可以看出连接池比直连的效率更好。

@Test
    public void ShardPipelinedPoolTest() {
        List<JedisShardInfo> shards = Arrays.asList(
                    new JedisShardInfo("localhost", 6379),
                    new JedisShardInfo("localhost", 6379)); // 使用相同的ip:port,仅作测试
    
        ShardedJedisPool pool = new ShardedJedisPool(new JedisPoolConfig(), shards);
        
        ShardedJedis sJ = pool.getResource();
    
        ShardedJedisPipeline Pipeline = sJ.pipelined();
    
        long start = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) {
            Pipeline.set("sppn" + i, "sppn" + i);
        }
        List<Object> res = Pipeline.syncAndReturnAll();
        long end = System.currentTimeMillis();
        pool.returnResource(sJ);
        System.out.println("Pipelined@Pool SET: " + ((end - start) / 1000.0) + " seconds");
    
        pool.destroy();
    }

### 测试结果 ###

Simple SET: 5.227 seconds
     
    Transaction SET: 0.5 seconds
    Pipelined SET: 0.353 seconds
    Pipelined transaction: 0.509 seconds
     
    Simple@Sharing SET: 5.289 seconds
    Pipelined@Sharing SET: 0.348 seconds
     
    Simple@Pool SET: 5.039 seconds
    Pipelined@Pool SET: 0.401 seconds

经测试分布式中用到的机器越多，调用会越慢。分布式中，连接池方式调用不但线程安全外，根据上面的测试数据，也可以看出连接池比直连的效率更好。

### 完整测试代码 ###

package redis;
    
    import java.util.Arrays;
    import java.util.List;
    
    import org.junit.AfterClass;
    import org.junit.BeforeClass;
    import org.junit.FixMethodOrder;
    import org.junit.Test;
    import org.junit.runners.MethodSorters;
    
    import redis.clients.jedis.Jedis;
    import redis.clients.jedis.JedisPoolConfig;
    import redis.clients.jedis.JedisShardInfo;
    import redis.clients.jedis.Pipeline;
    import redis.clients.jedis.ShardedJedis;
    import redis.clients.jedis.ShardedJedisPipeline;
    import redis.clients.jedis.ShardedJedisPool;
    import redis.clients.jedis.Transaction;
    
    /**
     * @ClassName: RedisTest
     * @Description: java调用redis的八种方式
     * @author: XHChen
     * @date: 2019年4月21日 上午9:57:21
     */
    @FixMethodOrder(MethodSorters.NAME_ASCENDING)
    public class RedisTest {
    	
    	private static List<JedisShardInfo> shards;
    	private static Jedis jedis;
    	private static ShardedJedis sharding;
    	private static ShardedJedisPool pool;
    	
    	@BeforeClass
    	public static void setUpBeforeClass() {
    		// 分布式多线程
    		shards = Arrays.asList(
    				new JedisShardInfo("localhost", 6379),
    				new JedisShardInfo("localhost", 6379)); // 使用相同的ip:port,仅作测试
    		
    		// 普通连接
    		jedis = new Jedis("localhost");
    		
    		// 分布式
    		sharding = new ShardedJedis(shards);
    		
    		// 创建连接池
    		pool = new ShardedJedisPool(new JedisPoolConfig(), shards);
    	}
    	
    	@AfterClass
    	public static void tearDownAfterClass() {
    		
    		jedis.disconnect();
    		sharding.disconnect();
    		pool.destroy();
    	}
    
    	/**
    	 * 
    	 * @Title: NormalSyncTest
    	 * @Description: 普通同步
    	 * @author XHChen
    	 * @date 2019年4月21日上午9:58:46
    	 */
    	@Test
    	public void NormalSyncTest() {
    		jedis = new Jedis("localhost");
    
    		long start = System.currentTimeMillis();
    		for (int i = 0; i < 100000; i++) {
    			String res = jedis.set("s" + i, "s" + i);
    		}
    		long end = System.currentTimeMillis();
    		System.out.println("Simple SET: " + ((end - start) / 1000.0) + " seconds");
    
    	}
    
    	/**
    	 * 
    	 * @Title: TransactionsTest
    	 * @Description: 事务方式
    	 * @author XHChen
    	 * @date 2019年4月21日上午10:14:39
    	 */
    	@Test
    	public void TransactionsTest() {
    
    		jedis = new Jedis("localhost");
    
    		long start = System.currentTimeMillis();
    		Transaction tx = jedis.multi(); // 创建事务
    		for (int i = 0; i < 100000; i++) {
    			tx.set("t" + i, "t" + i);
    		}
    		List<Object> res = tx.exec();
    		long end = System.currentTimeMillis();
    		System.out.println("Transaction SET: " + ((end - start) / 1000.0) + " seconds");
    
    	}
    
    	/**
    	 * 
    	 * @Title: PipeliningTest
    	 * @Description: 管道技术
    	 * @author XHChen
    	 * @date 2019年4月21日上午10:26:34
    	 */
    	@Test
    	public void PipeliningTest() {
    
    		jedis = new Jedis("localhost");
    
    		long start = System.currentTimeMillis();
    		Pipeline pipeline = jedis.pipelined();// new了一个Pipeline对象，并设置了该对象的client属性。
    		for (int i = 0; i < 100000; i++) {
    			pipeline.set("p" + i, "p" + i);// set方法实际调用的是Pipelinebase类（Pipeline父类的父类）的set方法
    		}
    		List<Object> res = pipeline.syncAndReturnAll();// 把每一次的请求操作加入到一个Queue里面
    		long end = System.currentTimeMillis();
    		System.out.println("Pipelined SET: " + ((end - start) / 1000.0) + " seconds");
    
    	}
    
    	/**
    	 * 
    	 * @Title: PipeliningTranTest
    	 * @Description: 管道中调用事务
    	 * @author XHChen
    	 * @date 2019年4月21日上午10:26:34
    	 */
    	@Test
    	public void PipeliningTranTest() {
    
    		jedis = new Jedis("localhost");
    
    		long start = System.currentTimeMillis();
    		Pipeline pipeline = jedis.pipelined();// new了一个Pipeline对象，并设置了该对象的client属性。
    		pipeline.multi();
    		for (int i = 0; i < 100000; i++) {
    			pipeline.set("px" + i, "px" + i);// set方法实际调用的是Pipelinebase类（Pipeline父类的父类）的set方法
    		}
    		pipeline.exec();
    		List<Object> res = pipeline.syncAndReturnAll();// 把每一次的请求操作加入到一个Queue里面
    		long end = System.currentTimeMillis();
    		System.out.println("Pipelined transaction: " + ((end - start) / 1000.0) + " seconds");
    
    	}
    
    	/**
    	 * 
    	 * @Title: ShardNormalTest
    	 * @Description: 分布式直连同步调用
    	 * @author XHChen
    	 * @date 2019年4月21日上午10:26:34
    	 */
    	@Test
    	public void ShardNormalTest() {
    
    		shards = Arrays.asList(
    				new JedisShardInfo("localhost", 6379),
    				new JedisShardInfo("localhost", 6379));
    
    		sharding = new ShardedJedis(shards);
    
    		long start = System.currentTimeMillis();
    		for (int i = 0; i < 100000; i++) {
    			String res = sharding.set("sn" + i, "sn" + i);
    		}
    		long end = System.currentTimeMillis();
    		System.out.println("Simple@Sharing SET: " + ((end - start) / 1000.0) + " seconds");
    
    	}
    
    	/**
    	 * 
    	 * @Title: ShardSyncPipelinedTest
    	 * @Description: 分布式直连异步调用
    	 * @author XHChen
    	 * @date 2019年4月21日上午10:26:34
    	 */
    	@Test
    	public void ShardSyncPipelinedTest() {
    
    		shards = Arrays.asList(
    				new JedisShardInfo("localhost", 6379),
    				new JedisShardInfo("localhost", 6379)); // 使用相同的ip:port,仅作测试
    
    		sharding = new ShardedJedis(shards);
    
    		ShardedJedisPipeline pipeline = sharding.pipelined();
    
    		long start = System.currentTimeMillis();
    		for (int i = 0; i < 100000; i++) {
    			pipeline.set("sp" + i, "sp" + i);
    		}
    		List<Object> res = pipeline.syncAndReturnAll();
    		long end = System.currentTimeMillis();
    		System.out.println("Pipelined@Sharing SET: " + ((end - start) / 1000.0) + " seconds");
    
    	}
    
    	/**
    	 * 
    	 * @Title: ShardSimplePoolTest
    	 * @Description: 分布式连接池同步调用
    	 * @author XHChen
    	 * @date 2019年4月21日上午10:26:34
    	 */
    	@Test
    	public void ShardSimplePoolTest() {
    
    		shards = Arrays.asList(
    				new JedisShardInfo("localhost", 6379),
    				new JedisShardInfo("localhost", 6379)); // 使用相同的ip:port,仅作测试
    		
    		pool = new ShardedJedisPool(new JedisPoolConfig(), shards);
    		
    		ShardedJedis sJ = pool.getResource();
    
    		long start = System.currentTimeMillis();
    		for (int i = 0; i < 100000; i++) {
    			String res = sJ.set("spn" + i, "spn" + i);
    		}
    		long end = System.currentTimeMillis();
    		pool.returnResource(sJ);
    		System.out.println("Simple@Pool SET: " + ((end - start) / 1000.0) + " seconds");
    
    	}
    	
    	/**
    	 * 
    	 * @Title: ShardPipelinedPoolTest
    	 * @Description: 分布式连接池异步调用
    	 * @author XHChen
    	 * @date 2019年4月21日上午10:26:34
    	 */
    	@Test
    	public void ShardPipelinedPoolTest() {
    
    		shards = Arrays.asList(
    				new JedisShardInfo("localhost", 6379),
    				new JedisShardInfo("localhost", 6379)); // 使用相同的ip:port,仅作测试
    
    		pool = new ShardedJedisPool(new JedisPoolConfig(), shards);
    		
    		ShardedJedis sJ = pool.getResource();
    		
    		ShardedJedisPipeline Pipeline = sJ.pipelined();
    
    		long start = System.currentTimeMillis();
    		for (int i = 0; i < 100000; i++) {
    			Pipeline.set("sppn" + i, "sppn" + i);
    		}
    		List<Object> res = Pipeline.syncAndReturnAll();
    		long end = System.currentTimeMillis();
    		pool.returnResource(sJ);
    		System.out.println("Pipelined@Pool SET: " + ((end - start) / 1000.0) + " seconds");
    
    	}
    
    }

[Apache Commons Pool]: https://mvnrepository.com/artifact/org.apache.commons/commons-pool2
[Jedis]: https://mvnrepository.com/artifact/redis.clients/jedis