Un verrou distribué est une méthode permettant de contrôler l'accès synchronisé aux ressources partagées entre systèmes distribués.

Ce qui suit présente comment Zookeeper implémente les verrous distribués, en expliquant deux types de verrous distribués : les verrous exclusifs et les verrous partagés.

Verrou exclusif

Les verrous exclusifs, également appelés verrous d'écriture ou verrous exclusifs, si la transaction T1 ajoute un verrou exclusif à l'objet de données O1, alors pendant toute la période de verrouillage, seule la transaction T1 est autorisée à lire et mettre à jour O1 ; aucune autre transaction ne peut lire ou écrire.

Définir le verrou :

/exclusive_lock/lock

Implémentation :

En utilisant la caractéristique d'unicité des nœuds frères de Zookeeper, lorsqu'un verrou exclusif doit être acquis, tous les clients tentent d'appeler l'interface create() pour créer/exclusive_lockun nœud enfant temporaire sous le nœud/exclusive_lock/lock, finalement un seul client peut créer avec succès, et ce client obtient le verrou distribué. En même temps, tous les clients qui n'ont pas acquis le verrou peuvent enregistrer/exclusive_lockun événement d'écoute watcher pour les changements de nœuds enfants sur le nœud, afin de se disputer à nouveau le verrou.

Verrou partagé

Les verrous partagés, également appelés verrous de lecture. Si la transaction T1 ajoute un verrou partagé à l'objet de données O1, alors la transaction actuelle ne peut effectuer que des opérations de lecture sur O1, et les autres transactions ne peuvent ajouter que des verrous partagés à cet objet de données jusqu'à ce que tous les verrous partagés sur l'objet de données soient libérés.

Définir le verrou :

/shared_lock/[hostname]-请求类型W/R-序号

Implémentation :

1. Le client appelle la méthode create pour créer un nœud séquentiel temporaire similaire à la méthode de définition du verrou.

2. Le client appelle l'interface getChildren pour obtenir une liste de tous les nœuds enfants créés.

3. Déterminer si le verrou est obtenu. Pour les requêtes de lecture, si tous les nœuds enfants dont le numéro de séquence est plus petit que le sien sont des requêtes de lecture, ou s'il n'existe aucun nœud enfant avec un numéro de séquence plus petit que le sien, cela indique que le verrou partagé a été acquis avec succès et la logique de lecture commence à s'exécuter. Pour les requêtes d'écriture, s'il ne s'agit pas du nœud enfant avec le plus petit numéro de séquence, il entre dans un état d'attente.

4. Si le verrou partagé n'est pas acquis, les requêtes de lecture enregistrent un observateur sur le dernier nœud de requête d'écriture avec un numéro de séquence plus petit que le sien, et les requêtes d'écriture enregistrent un observateur sur le dernier nœud avec un numéro de séquence plus petit que le sien.

Dans le développement réel, nous pouvons utiliser les API encapsulées dans la boîte à outils Curator pour nous aider à implémenter des verrous distribués.

<dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>x.x.x</version> </dependency>

Plusieurs schémas de verrouillage de Curator :

  • 1, InterProcessMutex: Verrou exclusif réentrant distribué
  • 2, InterProcessSemaphoreMutex: Verrou exclusif distribué
  • 3, InterProcessReadWriteLock: Verrou de lecture-écriture distribué

L'exemple suivant simule 50 threads utilisant le verrou exclusif réentrant InterProcessMutex pour se disputer le verrou simultanément :

Exemple

public class InterprocessLock {
    public static void main(String[] args)  {
        CuratorFramework zkClient = getZkClient();
        String lockPath = "/lock";
        InterProcessMutex lock = new InterProcessMutex(zkClient, lockPath);
        //Simuler 50 threads se disputant le verrou
        for (int i = 0; i < 50; i++) {
            new Thread(new TestThread(i, lock)).start();
        }
    }


    static class TestThread implements Runnable {
        private Integer threadFlag;
        private InterProcessMutex lock;

        public TestThread(Integer threadFlag, InterProcessMutex lock) {
            this.threadFlag = threadFlag;
            this.lock = lock;
        }

        @Override
        public void run() {
            try {
                lock.acquire();
                System.out.println(Le+threadFlag+thread a acquis le verrou);
                //Attendre 1 seconde puis libérer le verrou
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                try {
                    lock.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private static CuratorFramework getZkClient() {
        String zkServerAddress = "192.168.3.39:2181";
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);
        CuratorFramework zkClient = CuratorFrameworkFactory.builder()
                .connectString(zkServerAddress)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(5000)
                .retryPolicy(retryPolicy)
                .build();
        zkClient.start();
        return zkClient;
    }
}

La console affiche un enregistrement chaque seconde :

Téléchargement du code source pour ce chapitre

Téléchargement