Mise en œuvre de redis par golang (10): transactions atomiques locales

Création continue,Accélérer la croissance！C'est ma participation「Nouveau plan de la Journée des Nuggets · 6 Le défi de la lune」De27Oh, mon Dieu.,Cliquez pour voir les détails de l'événement

Pour supporter l'exécution atomique de plusieurs commandes Redis Fournit un mécanisme de transaction. Redis Les documents officiels indiquent que la transaction comporte deux garanties importantes:：

• La transaction est une opération isolée distincte：Toutes les commandes de la transaction sont sérialisées、Exécution séquentielle.Transaction en cours d'exécution,Ne sera pas interrompu par une demande de commande envoyée par un autre client.
• Une transaction est une opération atomique：Toutes les commandes de la transaction sont exécutées,Soit ils ne le font pas tous.

redis.io/docs/manual…

Nous pouvons rencontrer deux types d'erreurs lors de l'utilisation des transactions:

1. Erreur de syntaxe lors de la mise en file d'attente des commandes
2. Une erreur d'exécution s'est produite pendant l'exécution de la commande,Oui, par exemple. string Type key En cours lpush Fonctionnement

En cas d'erreur de syntaxe Redis Les commandes sont interrompues et les transactions sont abandonnées . En cas d'erreur d'exécution Redis Seules les erreurs sont signalées et les commandes restantes de la transaction sont exécutées. , Ne fait pas reculer les transactions comme la plupart des bases de données .C'est,Redis L'explication officielle est：

Redis La commande ne peut échouer qu'en raison d'une syntaxe incorrecte（Et ces problèmes ne peuvent pas être détectés en entrant dans l'équipe）,Ou la commande a été utilisée sur une clé du mauvais type：Ce qui veut dire,Du point de vue pratique,La commande échouée a été causée par une erreur de programmation,Et ces erreurs devraient être détectées pendant le développement,Ne devrait pas apparaître dans l'environnement de production.

Parce qu'il n'est pas nécessaire de soutenir le ROLLBACK,Alors... Redis L'intérieur du. Il y a un point de vue selon lequel Redis Les pratiques transactionnelles produisent bug , Il convient toutefois de noter que:, Dans des circonstances normales, Le retour en arrière ne résout pas les problèmes causés par les erreurs de programmation. Par exemple,, Si tu voulais passer INCR La commande ajoute la valeur de la clé 1 , Et j'ai accidentellement ajouté 2 , Ou une clé du mauvais type a été exécutée INCR , Il n'y a aucun moyen de faire reculer ces situations.Étant donné qu'il n'existe aucun mécanisme permettant d'éviter les erreurs commises par les programmeurs eux - mêmes, Et ce type d'erreur ne se produit généralement pas dans l'environnement de production, Alors... Redis Plus simple、Plus rapide sans retour en arrière pour traiter les transactions.

emmmm, Ensuite, nous essayons de Godis La mise en œuvre est atomique 、 Des affaires isolées .

L'atomicité d'une transaction a deux caractéristiques ：1. Le processus d'exécution de la transaction ne peut être effectué par aucune autre transaction (Thread)Insérer 2. La transaction a été entièrement réussie ou n'a pas été exécutée du tout , Il n'y a pas d'état de réussite partielle L'isolement d'une transaction est la question de savoir si le résultat d'une opération dans une transaction est visible pour d'autres transactions simultanées. .Parce queKV Il n'y a pas de problème de lecture fantôme dans la base de données , Nous devons donc éviter les problèmes de lecture sale et de non - répétabilité .

Analyse du mécanisme de transaction

Verrouillage

Avec Redis Les moteurs monothreadés sont différents godis Le moteur de stockage est parallèle , Il est donc nécessaire de concevoir un mécanisme de verrouillage pour assurer l'atomicité et l'isolement lors de l'exécution de plusieurs commandes. .

On est là. Implémenter une base de données de mémoire Il est mentionné dans un article que:

La mise en œuvre d'une commande générale nécessite 3Fonctions:

• ExecFunc Est la fonction qui exécute réellement la commande
• PrepareFunc In ExecFunc Exécution avant, Responsable de l'analyse de ce que la ligne de commande lit et écrit key Facile à verrouiller
• UndoFunc Utilisé uniquement dans les transactions , Responsable de la préparation undo logs Au cas où une erreur serait rencontrée pendant l'exécution de la transaction .

Dont: PrepareFunc La ligne de commande sera analysée et retournée pour la lecture et l'écriture key, Par prepareMSet Par exemple:

// return writtenKeys, readKeys
func prepareMSet(args [][]byte) ([]string, []string) {
	size := len(args) / 2
	keys := make([]string, size)
	for i := 0; i < size; i++ {
		keys[i] = string(args[2*i])
	}
	return keys, nil
}

Union Implémenter une base de données de mémoire Mentionné dans LockMap Le verrouillage est terminé . Impossible d'obtenir la corrélation en raison d'autres programmes key Il n'est donc pas possible d'insérer la serrure dans la transaction , Nous avons donc réalisé la propriété non inséparable de l'atomicité .

La transaction exige que tous les key Verrouillage terminé une fois , Ne peut être déverrouillé que lorsque la transaction est engagée ou reportée . On ne peut pas utiliser un key Il suffit d'ajouter une serrure et de la déverrouiller , Cette méthode peut entraîner une lecture sale ：

Temps	Services1	Services2
t1	Verrouillé.key A
t2	Modifierkey A
t3	Déverrouillerkey A
t4		Verrouillé.key A
t4		Lirekey A
t5		Déverrouillerkey A
t6	Soumettre

Comme le montre la figure ci - dessus t4 Le moment, Services 2 J'ai lu la transaction. 1Données non soumises, Une anomalie de lecture sale s'est produite .

Retour en arrière

Afin que les transactions puissent être reportées en cas d'erreur d'exécution （Atomicité）, Il y a deux façons de reculer ：

• Enregistrer avant de modifier value, Utilisez l'ancien valueCouverture
• Utiliser la commande ROLLBACK pour annuler l'effet de la commande originale .Par exemple,：CléALa valeur originale était1,AppeléIncr A Et ça devient 2, On peut recommencer Set A 1Ordre d'annuler incr Les ordres.

Pour économiser de la mémoire, nous avons finalement choisi la deuxième option .Par exemple, HSet L'ordre n'a besoin que d'un autre HSet Oui. field Il suffit de revenir à la valeur originale , Si Save est utilisé value Nous devons sauver tout HashMap. Dans le même ordre d'idées LPushRPop Attendez les ordres..

Certaines commandes peuvent nécessiter plusieurs commandes pour revenir en arrière ,Comme le ROLLBACK. Del Il n'est pas seulement nécessaire de restaurer key-value Il faut récupérer TTL Données.Ou Del La commande a supprimé plusieurs key Heure, Plusieurs commandes sont également nécessaires pour faire reculer . En résumé, nous donnons UndoFunc Définitions：

// UndoFunc returns undo logs for the given command line
// execute from head to tail when undo
type UndoFunc func(db *DB, args [][]byte) []CmdLine

Nous pouvons faire reculer n'importe quelle opération rollbackGivenKeysÀ titre d'exemple,Bien sûr.rollbackGivenKeysLe coût de, Cibler autant que possible undo log.

func rollbackGivenKeys(db *DB, keys ...string) []CmdLine {
	var undoCmdLines [][][]byte
	for _, key := range keys {
		entity, ok := db.GetEntity(key)
		if !ok {
			// Il n'existe pas. key Supprimer
			undoCmdLines = append(undoCmdLines,
				utils.ToCmdLine("DEL", key),
			)
		} else {
			undoCmdLines = append(undoCmdLines,
				utils.ToCmdLine("DEL", key), //  D'abord le nouveau  key Supprimer
				aof.EntityToCmd(key, entity).Args, // Prends ça. DataEntity  Séquençage en ligne de commande 
				toTTLCmd(db, key).Args,
			)
		}
	}
	return undoCmdLines
}

Ensuite, regardez EntityToCmd, Très facile à comprendre：

func EntityToCmd(key string, entity *database.DataEntity) *protocol.MultiBulkReply {
	if entity == nil {
		return nil
	}
	var cmd *protocol.MultiBulkReply
	switch val := entity.Data.(type) {
	case []byte:
		cmd = stringToCmd(key, val)
	case *List.LinkedList:
		cmd = listToCmd(key, val)
	case *set.Set:
		cmd = setToCmd(key, val)
	case dict.Dict:
		cmd = hashToCmd(key, val)
	case *SortedSet.SortedSet:
		cmd = zSetToCmd(key, val)
	}
	return cmd
}

var hMSetCmd = []byte("HMSET")

func hashToCmd(key string, hash dict.Dict) *protocol.MultiBulkReply {
	args := make([][]byte, 2+hash.Len()*2)
	args[0] = hMSetCmd
	args[1] = []byte(key)
	i := 0
	hash.ForEach(func(field string, val interface{}) bool {
		bytes, _ := val.([]byte)
		args[2+i*2] = []byte(field)
		args[3+i*2] = bytes
		i++
		return true
	})
	return protocol.MakeMultiBulkReply(args)
}

Watch

Redis Watch La commande est utilisée pour surveiller un(Ou plus) key ,Si c'est avant l'exécution de la transaction(Ou ceux - ci) key Modifié par d'autres commandes, Alors la transaction sera abandonnée .

Réalisation Watch Au cœur de l'ordre se trouve la découverte key Si elle a été modifiée , Nous utilisons un schéma de numéro de version simple et fiable ：Pour chaque key Stocker un numéro de version , Description du changement de numéro de version key Modifié:

// database/single_db.go
func (db *DB) GetVersion(key string) uint32 {
	entity, ok := db.versionMap.Get(key)
	if !ok {
		return 0
	}
	return entity.(uint32)
}

// database/transaciton.go
func Watch(db *DB, conn redis.Connection, args [][]byte) redis.Reply {
	watching := conn.GetWatching()
	for _, bkey := range args {
		key := string(bkey)
		watching[key] = db.GetVersion(key) //  Le numéro de version actuel existe  conn Dans l'objet
	}
	return protocol.MakeOkReply()
}

Comparer les numéros de version avant d'effectuer une transaction ：

// database/transaciton.go
func isWatchingChanged(db *DB, watching map[string]uint32) bool {
	for key, ver := range watching {
		currentVersion := db.GetVersion(key)
		if ver != currentVersion {
			return true
		}
	}
	return false
}

Guide de lecture du code source

Après avoir compris les mécanismes liés aux transactions , Regardons le Code de base pour l'exécution des transactions ExecMulti

func (db *DB) ExecMulti(conn redis.Connection, watching map[string]uint32, cmdLines []CmdLine) redis.Reply {
	// Phase préparatoire
	// Utiliser prepareFunc  Obtenir la transaction pour lire et écrire  key
	writeKeys := make([]string, 0) // may contains duplicate
	readKeys := make([]string, 0)
	for _, cmdLine := range cmdLines {
		cmdName := strings.ToLower(string(cmdLine[0]))
		cmd := cmdTable[cmdName]
		prepare := cmd.prepare
		write, read := prepare(cmdLine[1:])
		writeKeys = append(writeKeys, write...)
		readKeys = append(readKeys, read...)
	}
	watchingKeys := make([]string, 0, len(watching))
	for key := range watching {
		watchingKeys = append(watchingKeys, key)
	}
	readKeys = append(readKeys, watchingKeys...)
	//  À lire et à écrire  key Et par watch De key  Verrouiller ensemble 
	db.RWLocks(writeKeys, readKeys)
	defer db.RWUnLocks(writeKeys, readKeys)

	//  Vérifier par  watch De key Y a - t - il eu un changement
	if isWatchingChanged(db, watching) { // watching keys changed, abort
		return protocol.MakeEmptyMultiBulkReply()
	}

	// Phase de mise en œuvre
	results := make([]redis.Reply, 0, len(cmdLines))
	aborted := false
	undoCmdLines := make([][]CmdLine, 0, len(cmdLines))
	for _, cmdLine := range cmdLines {
		//  Préparez - vous avant l'exécution de la commande  undo log,  Cela garantit, par exemple, l'utilisation de  decr Retour en arrière incr  L'implémentation de la commande peut fonctionner correctement 
		undoCmdLines = append(undoCmdLines, db.GetUndoLogs(cmdLine))
		result := db.execWithLock(cmdLine)
		if protocol.IsErrorReply(result) {
			aborted = true
			// don't rollback failed commands
			undoCmdLines = undoCmdLines[:len(undoCmdLines)-1]
			break
		}
		results = append(results, result)
	}
	// Exécution réussie
	if !aborted { 
		db.addVersion(writeKeys...)
		return protocol.MakeMultiRawReply(results)
	}
	//  La transaction a échoué à reculer 
	size := len(undoCmdLines)
	for i := size - 1; i >= 0; i-- {
		curCmdLines := undoCmdLines[i]
		if len(curCmdLines) == 0 {
			continue
		}
		for _, cmdLine := range curCmdLines {
			db.execWithLock(cmdLine)
		}
	}
	return protocol.MakeErrReply("EXECABORT Transaction discarded because of previous errors.")
}