redis中pipeline詳解

一、pipeline出現的背景：

redis執行一條指令有四個程序：傳送指令、指令排隊、指令執行、傳回結果；

這個過程稱為Round trip time(簡稱RTT, 往返時間)，mget mset有效節約了RTT，但大部分命令（如hgetall，並沒有mhgetall）不支援批次操作，需要消耗N次RTT ，這時候需要pipeline來解決這個問題。

二、pepeline的效能

1、未使用pipeline執行N條指令

2、使用了pipeline執行N條指令

3、兩者效能比較

#小結：這是一組統計資料出來的數據，使用Pipeline執行速度比逐條執行快，特別是客戶端與服務端的網路延遲越大，效能體能越明顯。

下面貼出測試程式碼分析兩者的效能差異：

	@Test
	public void pipeCompare() {
		Jedis redis = new Jedis("192.168.1.111", 6379);
		redis.auth("12345678");//授权密码 对应redis.conf的requirepass密码
		Map<String, String> data = new HashMap<String, String>();
		redis.select(8);//使用第8个库
		redis.flushDB();//清空第8个库所有数据
		// hmset
		long start = System.currentTimeMillis();
		// 直接hmset
		for (int i = 0; i < 10000; i++) {
			data.clear();  //清空map
			data.put("k_" + i, "v_" + i);
			redis.hmset("key_" + i, data); //循环执行10000条数据插入redis
		}
		long end = System.currentTimeMillis();
		System.out.println("    共插入:[" + redis.dbSize() + "]条 .. ");
		System.out.println("1,未使用PIPE批量设值耗时" + (end - start) / 1000 + "秒..");
		redis.select(8);
		redis.flushDB();
		// 使用pipeline hmset
		Pipeline pipe = redis.pipelined();
		start = System.currentTimeMillis();
		//
		for (int i = 0; i < 10000; i++) {
			data.clear();
			data.put("k_" + i, "v_" + i);
			pipe.hmset("key_" + i, data); //将值封装到PIPE对象，此时并未执行，还停留在客户端
		}
		pipe.sync(); //将封装后的PIPE一次性发给redis
		end = System.currentTimeMillis();
		System.out.println("    PIPE共插入:[" + redis.dbSize() + "]条 .. ");
		System.out.println("2,使用PIPE批量设值耗时" + (end - start) / 1000 + "秒 ..");
//--------------------------------------------------------------------------------------------------
		// hmget
		Set<String> keys = redis.keys("key_*"); //将上面设值所有结果键查询出来
		// 直接使用Jedis hgetall
		start = System.currentTimeMillis();
		Map<String, Map<String, String>> result = new HashMap<String, Map<String, String>>();
		for (String key : keys) {
			//此处keys根据以上的设值结果，共有10000个，循环10000次
			result.put(key, redis.hgetAll(key)); //使用redis对象根据键值去取值，将结果放入result对象
		}
		end = System.currentTimeMillis();
		System.out.println("    共取值:[" + redis.dbSize() + "]条 .. ");
		System.out.println("3,未使用PIPE批量取值耗时 " + (end - start) / 1000 + "秒 ..");

		// 使用pipeline hgetall
		result.clear();
		start = System.currentTimeMillis();
		for (String key : keys) {
			pipe.hgetAll(key); //使用PIPE封装需要取值的key,此时还停留在客户端，并未真正执行查询请求
		}
		pipe.sync();  //提交到redis进行查询
		
		end = System.currentTimeMillis();
		System.out.println("    PIPE共取值:[" + redis.dbSize() + "]条 .. ");
		System.out.println("4,使用PIPE批量取值耗时" + (end - start) / 1000 + "秒 ..");

		redis.disconnect();
	}

#三、原生批次指令(mset, mget)與Pipeline比較

1、原生批命令是原子性，pipeline是非原子性

(原子性概念:一個事務是一個不可分割的最小工作單位,要么都成功要么都失敗。原子操作是指你的一個業務邏輯必須是不可拆分的. 處理一件事情要么都成功，要么都失敗，原子不可拆分)

2、原生批命令一命令多個key, 但pipeline支持多命令（存在事務），非原子性

3、原生批次指令是服務端實現，而pipeline需要服務端與客戶端共同完成

四、Pipeline正確使用方式

使用pipeline組裝的指令個數不能太多，不然資料量過大，增加客戶端的等待時間，還可能造成網路阻塞，可以將大量指令的拆分多個小的pipeline指令完成。

1、Jedis中的pipeline使用方式

大家知道redis提供了mset、mget方法，但沒有提供mdel方法，如果想實現，可以藉助pipeline實現。

2、Jedis中的pipeline使用步驟：

• 取得jedis物件（一般從連接池取得）

取得jedis物件的pipeline物件

新增指令

#執行指令

##測試類別方法：

	 @Test
	public void testCommond() {
		// 工具类初始化
		JedisUtils jedis = new JedisUtils("192.168.1.111", 6379, "12345678");

		for (int i = 0; i < 100; i++) {
			// 设值
			jedis.set("n" + i, String.valueOf(i));
		}
		System.out.println("keys from redis return =======" + jedis.keys("*"));

	}

	// 使用pipeline批量删除
	 @Test
	public void testPipelineMdel() {
		// 工具类初始化
		JedisUtils jedis = new JedisUtils("192.168.1.111", 6379, "12345678");
		List<String> keys = new ArrayList<String>();
		for (int i = 0; i < 100; i++) {
			keys.add("n" + i);
		}
		jedis.mdel(keys);
		System.out.println("after mdel the redis return ---------" + jedis.keys("*"));
	}

JedisUtils下的mdel方法：

	/**
	 * 删除多个字符串key 并释放连接
	 * 
	 * @param keys*
	 * @return 成功返回value 失败返回null
	 */
	public boolean mdel(List<String> keys) {
		Jedis jedis = null;
		boolean flag = false;
		try {
			jedis = pool.getResource();//从连接借用Jedis对象
			Pipeline pipe = jedis.pipelined();//获取jedis对象的pipeline对象
			for(String key:keys){
				pipe.del(key); //将多个key放入pipe删除指令中
			}
			pipe.sync(); //执行命令，完全此时pipeline对象的远程调用 
			flag = true;
		} catch (Exception e) {
			pool.returnBrokenResource(jedis);
			e.printStackTrace();
		} finally {
			returnResource(pool, jedis);
		}
		return flag;
	}

使用pipeline提交所有操作並傳回執行結果：

@Test
	public void testPipelineSyncAll() {
		// 工具类初始化
		Jedis jedis = new Jedis("192.168.1.111", 6379);
		jedis.auth("12345678");
		// 获取pipeline对象
		Pipeline pipe = jedis.pipelined();
		pipe.multi();
		pipe.set("name", "james"); // 调值
		pipe.incr("age");// 自增
		pipe.get("name");
		pipe.discard();
		// 将不同类型的操作命令合并提交，并将操作操作以list返回
		List<Object> list = pipe.syncAndReturnAll();

		for (Object obj : list) {
			// 将操作结果打印出来
			System.out.println(obj);
		}
		// 断开连接，释放资源
		jedis.disconnect();
	}

五、redis交易

pipeline是多條命令的組合，為了保證它的原子性，redis提供了簡單的事務。

1、redis的簡單事務，

一組需要一起執行的指令放到multi和exec兩個指令之間，其中multi代表事務開始，exec代表事務結束。

2、停止交易discard

#3、指令錯誤，語法不正確，導致交易無法正常結束

######4、執行錯誤，文法正確，但型別錯誤，交易可以正常結束###############5、watch指令： ######使用watch後， multi失效，交易失效################WATCH的機制是：當交易EXEC指令執行時，Redis會檢查被WATCH的key，只有被WATCH的key從WATCH起始時至今沒有發生過變更，EXEC才會被執行。如果WATCH的key在WATCH指令到EXEC指令之間發生過變化，則EXEC指令會回傳失敗。 ######更多redis知識請關注###redis入門教學###欄位。 ###

以上是redis中pipeline詳解的詳細內容。更多資訊請關注PHP中文網其他相關文章！