Multi : 開啟事務(wù)
Exec [rb]：執(zhí)行事務(wù)
Watch key [key...]：監(jiān)視key在事務(wù)執(zhí)行之前不能發(fā)生變更,如果事務(wù)執(zhí)行時發(fā)現(xiàn)有變更，事務(wù)終止執(zhí)行
Unwatch:取消監(jiān)視
Discard: 退出事務(wù)

func (db *DB) Exec(c abstract.Connection, redisCommand [][]byte) protocol.Reply {
	cmdName := strings.ToLower(string(redisCommand[0]))
	if cmdName == "multi" {
		iflen(redisCommand) != 1 {
			return protocol.NewArgNumErrReply(cmdName)
		}
		return StartMulti(c) // 開啟事務(wù)
	} elseif cmdName == "discard" {
		iflen(redisCommand) != 1 {
			return protocol.NewArgNumErrReply(cmdName)
		}
		return DiscardMulti(c) // 取消事務(wù)
	} elseif cmdName == "watch" {
		return Watch(db, c, redisCommand[1:]) // 監(jiān)視watch key [key...]
	} elseif cmdName == "unwatch" {
		iflen(redisCommand) != 1 {
			return protocol.NewArgNumErrReply(cmdName)
		}
		return UnWatch(db, c) // 取消監(jiān)視
	} elseif cmdName == "exec" {
		return ExecMulti(db, c, redisCommand[1:]) // 執(zhí)行事務(wù)
	}
	// **事務(wù)模式** 將命令入隊到命令緩沖隊列中
	if c != nil && c.IsTransaction() {
		return EnqueueCmd(c, redisCommand)
	}
	// ** 普通模式 **
	return db.execNormalCommand(c, redisCommand)
}

// 開啟事務(wù)
func StartMulti(c abstract.Connection) protocol.Reply {
	if c.IsTransaction() {
		return protocol.NewGenericErrReply("multi is already start,do not repeat it")
	}
	// 設(shè)定開啟
	c.SetTransaction(true)
	return protocol.NewOkReply()
}

// socket連接
func (k *KeepConnection) SetTransaction(val bool) {
	if !val { // 取消事務(wù)模式，清空隊列和watch key
		k.queue = nil
		k.watchKey = nil
		k.txErrors = nil
	}
	// 開啟事務(wù)狀態(tài)
	k.trx.Store(val)
}

//... 省略
// **事務(wù)模式** 將命令入隊到命令緩沖隊列中
	if c != nil && c.IsTransaction() {
		return EnqueueCmd(c, redisCommand)
	}
//... 省略

// 取消事務(wù)
func DiscardMulti(c abstract.Connection) protocol.Reply {
	if !c.IsTransaction() {
		return protocol.NewGenericErrReply("DISCARD without MULTI")
	}
	// 取消開啟
	c.SetTransaction(false)
	return protocol.NewOkReply()
}

// 監(jiān)視 key [key...]
func Watch(db *DB, conn abstract.Connection, args [][]byte) protocol.Reply {
	iflen(args) < 1 {
		return protocol.NewArgNumErrReply("WATCH")
	}
	if conn.IsTransaction() {
		return protocol.NewGenericErrReply("WATCH inside MULTI is not allowed")
	}
	watching := conn.GetWatchKey()
	for _, bkey := range args {
		key := string(bkey)
		watching[key] = db.GetVersion(key) // 保存當(dāng)前key的版本號（利用版本號機制判斷key是否有變化）
	}
	return protocol.NewOkReply()
}

// 清空watch key
func UnWatch(db *DB, conn abstract.Connection) protocol.Reply {
	conn.CleanWatchKey()
	return protocol.NewOkReply()
}

// 執(zhí)行事務(wù)  exec rb
func ExecMulti(db *DB, conn abstract.Connection, args [][]byte) protocol.Reply {
	// 說明當(dāng)前不是【事務(wù)模式】
	if !conn.IsTransaction() {
		return protocol.NewGenericErrReply("EXEC without MULTI")
	}
	// 執(zhí)行完，自動退出事務(wù)模式
	defer conn.SetTransaction(false)
	// 如果在入隊的時候，就有格式錯誤，直接返回
	iflen(conn.GetTxErrors()) > 0 {
		return protocol.NewGenericErrReply("EXECABORT Transaction discarded because of previous errors.")
	}
	// 是否自動回滾（這里是自定義的一個參數(shù)，標(biāo)準(zhǔn)redis中沒有）
	isRollBack := false
	iflen(args) > 0 && strings.ToUpper(string(args[0])) == "RB" { // 有rb參數(shù)，說明要自動回滾
		isRollBack = true
	}
	// 獲取所有的待執(zhí)行命令
	cmdLines := conn.GetQueuedCmdLine()
	return db.execMulti(conn, cmdLines, isRollBack)
}

// 執(zhí)行事務(wù)：本質(zhì)就是一堆命令一起執(zhí)行， isRollback 表示出錯是否回滾
func (db *DB) execMulti(conn abstract.Connection, cmdLines []CmdLine, isRollback bool) protocol.Reply {
	// 命令的執(zhí)行結(jié)果
	results := make([]protocol.Reply, len(cmdLines))
	versionKeys := make([][]string, len(cmdLines))
	var writeKeys []string
	var readKeys []string
	for idx, cmdLine := range cmdLines {
		cmdName := strings.ToLower(string(cmdLine[0]))
		cmd, ok := commandCenter[cmdName]
		if !ok {
			// 這里正常不會執(zhí)行
			continue
		}
		keyFunc := cmd.keyFunc
		readKs, writeKs := keyFunc(cmdLine[1:])
		// 讀寫key
		readKeys = append(readKeys, readKs...)
		writeKeys = append(writeKeys, writeKs...)
		// 寫key需要 變更版本號
		versionKeys[idx] = append(versionKeys[idx], writeKs...)
	}
	watchingKey := conn.GetWatchKey()
	if isWatchingChanged(db, watchingKey) { // 判斷watch key是否發(fā)生了變更
		return protocol.NewEmptyMultiBulkReply()
	}
	// 所有key上鎖（原子性）
	db.RWLock(readKeys, writeKeys)
	defer db.RWUnLock(readKeys, writeKeys)
	undoCmdLines := [][]CmdLine{}
	aborted := false
	for idx, cmdLine := range cmdLines {
		// 生成回滾命令
		if isRollback {
			undoCmdLines = append(undoCmdLines, db.GetUndoLog(cmdLine))
		}
		// 執(zhí)行命令
		reply := db.execWithLock(cmdLine)
		if protocol.IsErrReply(reply) { // 執(zhí)行出錯
			if isRollback { // 需要回滾
				undoCmdLines = undoCmdLines[:len(undoCmdLines)-1] // 命令執(zhí)行失?。ú挥没貪L），剔除最后一個回滾命令
				aborted = true
				break
			}
		}
		// 執(zhí)行結(jié)果
		results[idx] = reply
	}
	// 中斷，執(zhí)行回滾
	if aborted {
		size := len(undoCmdLines)
		// 倒序執(zhí)行回滾指令（完成回滾）
		for i := size - 1; i >= 0; i-- {
			curCmdLines := undoCmdLines[i]
			iflen(curCmdLines) == 0 {
				continue
			}
			for _, cmdLine := range curCmdLines {
				db.execWithLock(cmdLine)
			}
		}
		return protocol.NewGenericErrReply("EXECABORT Transaction discarded because of previous errors.")
	}
	// 執(zhí)行到這里，說明命令執(zhí)行完成（可能全部成功，也可能部分成功）
	for idx, keys := range versionKeys {
		if !protocol.IsErrReply(results[idx]) { // 針對執(zhí)行成功的命令（寫命令），變更版本號
			db.addVersion(keys...)
		}
	}
	// 將多個命令執(zhí)行的結(jié)果，進行合并返回
	mixReply := protocol.NewMixReply()
	mixReply.Append(results...)
	return mixReply
}