redis中使用事務保護數據完整性

事務是指一個執行過程，要么全部執行成功，要么失敗什么都不改變。不會存在一部分成功一部分失敗的情況，也就是事務的ACID四大特性（原子性、一致性、隔離性、持久性）。但是redis中的事務并不是嚴格意義上的事務，它只是保證了多個命令執行是按照順序執行，在執行過程中不會插入其他的命令，并不會保證所有命令都成功。也就是說在命令執行過程中，某些命令的失敗不會回滾前面已經執行成功的命令，也不會影響后面命令的執行。
redis中的事務跟pipeline很類似，但pipeline是批量提交命令到服務器，服務器執行命令過程中是一條一條執行的，在執行過程中是可以插入其他的命令。而事務是把這些命令批量提交到服務器，服務器執行命令過程也是一條一條執行的，但是在執行這一批命令時是不能插入執行其他的命令，必須等這一批命令執行完成后才能執行其他的命令。

1、事務基本結構

與數據庫事務執行過程類似，redis事務是由multi、exec、discard三個關鍵字組成，對比數據庫事務的begin、commit、rollback三個關鍵字。
命令行示例如下：

127.0.0.1:6379> set key1 value111
OK
127.0.0.1:6379> set key2 value222
OK
127.0.0.1:6379> mget key1 key2
1) "value111"
2) "value222"
127.0.0.1:6379> # 第一個事務
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> set key1 value-111
QUEUED
127.0.0.1:6379(TX)> setm key2 value-222
(error) ERR unknown command `setm`, with args beginning with: `key2`, `value-222`, 
127.0.0.1:6379(TX)> exec
(error) EXECABORT Transaction discarded because of previous errors.
127.0.0.1:6379> 
127.0.0.1:6379> mget key1 key2
1) "value111"
2) "value222"
127.0.0.1:6379> # 第二個事務
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> set key1 value-111
QUEUED
127.0.0.1:6379(TX)> set key2 value-222
QUEUED
127.0.0.1:6379(TX)> discard
OK
127.0.0.1:6379> 
127.0.0.1:6379> mget key1 key2
1) "value111"
2) "value222"
127.0.0.1:6379> # 第三個事務
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> set key1 value-111
QUEUED
127.0.0.1:6379(TX)> set key3 value-333 vddd
QUEUED
127.0.0.1:6379(TX)> set key2 value-222
QUEUED
127.0.0.1:6379(TX)> exec
1) OK
2) (error) ERR syntax error
3) OK
127.0.0.1:6379> 
127.0.0.1:6379> mget key1 key2 key3
1) "value-111"
2) "value-222"
3) (nil)
127.0.0.1:6379>

在上面的示例過程中，第一個事務執行時輸入了一個錯誤的命令，在提交事務時整個命令都沒有執行；第二個事務提交了多個命令，但是最后回滾了事務，整個事務也不會執行；第三個事務在提交命令時，故意設置一個執行失敗的命令，會發現這個失敗的命令并不會影響其他命令的成功。

2、事務的執行步驟

redis中的事務是分兩步執行的：第一步命令排隊，也就是將所有要執行的命令添加到一個隊列中，在這一步中命令不會真正執行；第二步命令執行或取消，在這一步中真正處理隊列中的命令，如果是exec命令，就執行這些命令；如果是discard命令，就取消執行命令。這里需要注意，如果在排隊中某些命令解析出錯，即使調用了exec命令，整個隊列中的命令也不會執行；但是如果在執行過程中出現了錯誤，它并不會影響其他命令的正常執行，一般使用封裝好的客戶端不會出現這種命令錯誤情況。

3、并發事務

多線程的項目就會有并發問題，并發問題就會存在數據不一致，數據庫中解決并發問題是通過鎖來實現的，在操作數據前加鎖，保證數據在整個執行過程中不被其他程序修改。這種方式加鎖是悲觀鎖，每次更新操作都認為數據會被其他程序修改，導致程序的并發性能不好；還有一種加鎖方式是樂觀鎖，每次直到真正更新數據時才判斷數據是否被更新了，如果數據被更新就放棄操作，對于讀多寫少的場景非常適合，一般實現樂觀鎖是通過版本號機制。
redis中就支持這種樂觀鎖機制，它的實現是通過watch命令，在開始執行事務前先通過watch監控被更新的key，如果在事務執行時發現這些key被修改了，那么就不執行事務中的命令。
下面演示的是扣費場景：在進行扣費前，先判斷用戶的余額，如果余額夠扣，就扣減用戶的賬號余額；如果余額不足，就不能扣減賬號余額。

watch某個key后，如果數據沒有被其他客戶端修改，事務將成功執行：

127.0.0.1:6379> set user:balance:1 100
OK
127.0.0.1:6379> watch user:balance:1
OK
127.0.0.1:6379> get user:balance:1
"100"
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> incrby user:balance:1 -100
QUEUED
127.0.0.1:6379(TX)> exec
1) (integer) 0
127.0.0.1:6379> get user:balance:1
"0"
127.0.0.1:6379>

watch某個key后，如果key對應的值被其他程序修改了，執行事務將不成功；如果不用watch命令，事務會成功執行。
下圖展示了在兩個客戶端驗證事務：
1、首先在下面的客戶端設置鍵的值為100；
2、然后設置 watch 該值，并且開啟事務；
3、執行減100的命令；
4、在上面的客戶端中修改這個鍵的值，減3；
5、下面的客戶端執行 exec 命令提交事務。
這幾個步驟執行完成后，發現數據沒有修改成功，表示 watch 命令監控到數據變動沒有執行事務中的命令。

下面演示步驟與上面一樣，只不過在事務前沒有 watch 命令，發現數據被修改了。
watch命令只對當前客戶端中的 multi / exec 之間的命令有影響，不在它們之間的命令不受影響，可以正常執行：

127.0.0.1:6379> set user:balance:1 100
OK
127.0.0.1:6379> watch user:balance:1
OK
127.0.0.1:6379> incrby user:balance:1 -3
(integer) 97
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> incrby user:balance:1 -100
QUEUED
127.0.0.1:6379(TX)> set watchkey aaa
QUEUED
127.0.0.1:6379(TX)> exec
(nil)
127.0.0.1:6379> mget user:balance:1 watchkey
1) "97"
2) (nil)
127.0.0.1:6379>

上面這段代碼在watch命令后對鍵的值進行了修改，發現更新成功；watch的key對應值被修改了，導致事務內的命令不執行，所以后面mget命令沒有獲取到新的值。

與watch對應有一個unwatch命令，它表示watch某個key后可以通過unwatch取消監控；如果watch某個key后有 exec 或 discard 命令執行，程序會自動取消監控，不必再使用unwatch取消監控：

127.0.0.1:6379> watch user:balance:1
OK
127.0.0.1:6379> unwatch
OK
127.0.0.1:6379>

4、代碼中使用

使用jedis實現扣費邏輯

首先還是先使用jedis測試上面提出扣費場景：
引入依賴：

<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>4.3.0</version>
</dependency>

主要代碼邏輯如下：

import redis.clients.jedis.*;
import java.time.Duration;
import java.util.List;public class JedisUtil {/*** 連接池*/private JedisPool jedisPool;/*** 連接初始化* @param host* @param port* @param password*/public JedisUtil(String host, int port, String password) {JedisPoolConfig config = new JedisPoolConfig();config.setMaxTotal(256);config.setMaxIdle(256);config.setMinIdle(1);config.setMaxWait(Duration.ofMillis(300));if(password != null && !"".equals(password)) {jedisPool = new JedisPool(config, host, port, 500, password);} else {jedisPool = new JedisPool(config, host, port, 500);}}/*** 關閉連接池*/public void close() {if(jedisPool != null && !jedisPool.isClosed()) {jedisPool.clear();jedisPool.close();}}/*** 獲取連接* @return*/public Jedis getJedis() {if(jedisPool != null && !jedisPool.isClosed()) {return jedisPool.getResource();}return null;}/*** 歸還jedis對象* @param jedis*/public void returnJedis(Jedis jedis) {if(jedis != null) {jedis.close();}}public static void main(String[] args) {// 獲取jedis連接JedisUtil util = new JedisUtil("192.168.56.101", 6379, "");// 鍵String key = "user:balance:1";util.deduct(key, 100);}/*** 扣減金額*/public void deduct(String key, int money) {Jedis jedis = this.getJedis();// 監控鍵對應值的變化jedis.watch(key);// 獲取賬戶余額，當余額足夠時扣減金額String val = jedis.get(key);if(val != null && Integer.parseInt(val) >= money) {// 開啟事務Transaction multi = jedis.multi();try {// 事務中的命令multi.incrBy(key, -money);System.out.println("run in multi!");// 執行事務List<Object> exec = multi.exec();System.out.println("run exec : " + exec);} catch (Exception e) {// 放棄事務multi.discard();e.printStackTrace();} finally {this.returnJedis(jedis);}} else {// 取消監控jedis.unwatch();System.out.println("余額不足...");}}
}

在上面代碼中執行事務部分添加斷點，并通過其他客戶端更新watch對應key的值，發現事務并不執行，這就達到了數據邏輯的一致性，不會因為其他客戶端扣減金額后，該客戶端繼續扣減余額導致剩余金額為負數的情況。

redisTemplate使用事務

在redisTemplate中使用事務，有三種方式，下面的代碼是實現上面扣費邏輯的過程：
（1）使用SessionCallback實現：

public void runTransaction(final String key, final String value) {List<Object> exec = redisTemplate.execute(new SessionCallback<List<Object>>() {@Overridepublic <K, V> List<Object> execute(RedisOperations<K, V> operations) throws DataAccessException {List<Object> exec = null;// 監控鍵對應值的變化operations.watch((K) key);ValueOperations<String, String> op1 = (ValueOperations<String, String>) operations.opsForValue();String val = op1.get(key);int num = Integer.parseInt(value);if(val != null && Integer.parseInt(val) >= num) {try {// 開啟事務operations.multi();// 事務中的命令op1.increment(key, -num);// 執行事務exec = operations.exec();} catch (NumberFormatException e) {// 放棄事務operations.discard();e.printStackTrace();}} else {// 取消監控operations.unwatch();System.out.println("余額不足...");}return exec;}});System.out.println(exec);
}

（2）使用RedisCallback實現：

public void runTransaction(final String key, final String value) {List<Object> exec = redisTemplate.execute(new RedisCallback<List<Object>>() {@Overridepublic List<Object> doInRedis(RedisConnection connection) throws DataAccessException {List<Object> exec = null;// 監控鍵對應值的變化connection.watch(key.getBytes(StandardCharsets.UTF_8));byte[] val = connection.get(key.getBytes(StandardCharsets.UTF_8));int num = Integer.parseInt(value);if(val != null && Integer.parseInt(new String(val)) >= num) {try {// 開啟事務connection.multi();// 事務中的命令connection.incrBy(key.getBytes(StandardCharsets.UTF_8), -num);// 執行事務exec = connection.exec();} catch (NumberFormatException e) {// 放棄事務connection.discard();e.printStackTrace();}} else {// 取消監控connection.unwatch();System.out.println("余額不足...");}return exec;}});System.out.println(exec);
}

（3）使用@Transactional注解實現：

@Transactional
public void runTransaction(final String key, final String value) {// 監控鍵對應值的變化redisTemplate.watch(key);String val = redisTemplate.opsForValue().get(key);int num = Integer.parseInt(value);if(val != null && Integer.parseInt(val) >= num) {// 開啟事務支持// 開啟這個值后，所有的命令都會在exec執行完才返回結果，所以需要返回值的命令要在這個方法前執行redisTemplate.setEnableTransactionSupport(true);try {// 開啟事務redisTemplate.multi();// 事務中的命令redisTemplate.opsForValue().increment(key, -num);// 執行事務List<Object> exec = redisTemplate.exec();System.out.println(exec);} catch (Exception e) {// 放棄事務redisTemplate.discard();e.printStackTrace();} finally {// 關閉事務支持redisTemplate.setEnableTransactionSupport(false);}} else {// 取消監控redisTemplate.unwatch();System.out.println("余額不足...");}
}

事務只能保證每一條命令的原子性，并不能保證事務內所有命令的原子性，上面的示例代碼已經驗證了這個結論，其實redis中已經提供了一些多值指令，如：mset、mget、hmset、hmget。但是這些只能是一種數據類型的多鍵值對操作，這些命令是原子操作。
上面這種扣費邏輯，除了使用redis的事務支持，還可以使用lua腳本實現，lua腳本在服務端執行與事務中的命令類似，是不可分割的整體，下面演示lua腳本內容，可以實現上面一樣的處理結果：
lua腳本如下：

local b = redis.call('get', KEYS[1]);
if tonumber(b) >= tonumber(ARGV[1]) thenlocal rs = redis.call('incrby', KEYS[1], 0 - tonumber(ARGV[1]));return rs;
else return nil;
end;

測試過程：

127.0.0.1:6379> set user:balance:1 100
OK
127.0.0.1:6379> get user:balance:1
"100"
127.0.0.1:6379> eval "local b = redis.call('get', KEYS[1]); if tonumber(b) >= tonumber(ARGV[1]) then local rs = redis.call('incrby', KEYS[1], 0 - tonumber(ARGV[1])); return rs; else return nil; end;" 1 user:balance:1 100
(integer) 0
127.0.0.1:6379> get user:balance:1
"0"
127.0.0.1:6379> set user:balance:1 100
OK
127.0.0.1:6379> incrby user:balance:1 -3
(integer) 97
127.0.0.1:6379> eval "local b = redis.call('get', KEYS[1]); if tonumber(b) >= tonumber(ARGV[1]) then local rs = redis.call('incrby', KEYS[1], 0 - tonumber(ARGV[1])); return rs; else return nil; end;" 1 user:balance:1 100
(nil)
127.0.0.1:6379> get user:balance:1
"97"
127.0.0.1:6379>

第一次執行余額正常夠扣場景，第二次設置余額不足，會發現扣減邏輯并沒有執行。
以上內容就是redis中事務的全部內容，要記住幾點：
（1）redis中的事務跟我們平時用的數據庫中的事務有一些差異的，它能保證多條命令執行時中間不會插入其他的命令，但并不會保證所有命令都執行成功，單條redis命令能保證原子性，但事務中的多條命令并不是原子性。
（2）redis中事務分兩步完成：第一步將所有命令添加到命令隊列中，這一步并不會執行命令；第二步執行隊列中的命令。如果第一步中的命令有錯誤，第二步并不會執行；如果第二步已經開始執行了，那么部分失敗的命令并不會影響其他正確命令的結果，這樣就會導致一部分命令成功而另外一部分命令失敗的場景。
（3）事務中不宜執行過多的命令或非常耗時的命令，因為redis底層執行命令是單線程，如果單個事務中執行過多的命令會導致其他客戶端的請求被阻塞。