El bloqueo distribuido es una forma de controlar el acceso sincronizado a recursos compartidos entre sistemas distribuidos.

A continuación se explica cómo Zookeeper implementa bloqueos distribuidos, y se describen dos tipos de bloqueos distribuidos: el bloqueo exclusivo y el bloqueo compartido.

Bloqueo exclusivo

El bloqueo exclusivo (Exclusive Locks), también llamado bloqueo de escritura o bloqueo de exclusividad. Si la transacción T1 añade un bloqueo exclusivo al objeto de datos O1, durante todo el período de bloqueo solo se permite que la transacción T1 realice operaciones de lectura y actualización sobre O1; ninguna otra transacción puede realizar lecturas o escrituras.

Definición del bloqueo:

/exclusive_lock/lock

Método de implementación:

Aprovechando la característica de unicidad de los nodos del mismo nivel en Zookeeper, cuando se necesita adquirir un bloqueo exclusivo, todos los clientes intentan, mediante la llamada a la interfaz create(),/exclusive_lockcrear un nodo temporal hijo bajo el nodo,/exclusive_lock/lockfinalmente solo un cliente puede crearlo con éxito, y ese cliente obtiene el bloqueo distribuido. Al mismo tiempo, todos los clientes que no obtuvieron el bloqueo pueden/exclusive_lockregistrar en el nodo un evento de escucha watcher para cambios de nodos secundarios, con el fin de volver a intentar obtener el bloqueo.

Bloqueo compartido

El bloqueo compartido (Shared Locks), también llamado bloqueo de lectura. Si la transacción T1 agrega un bloqueo compartido al objeto de datos O1, entonces la transacción actual solo puede realizar operaciones de lectura sobre O1; otras transacciones solo pueden agregar bloqueos compartidos a este objeto de datos, hasta que todos los bloqueos compartidos sobre el objeto de datos sean liberados.

Definición del bloqueo:

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

Método de implementación:

1. El cliente llama al método create para crear un nodo secuencial temporal similar al modo de definición de bloqueo.

2. El cliente llama a la interfaz getChildren para obtener la lista de todos los nodos secundarios creados.

3. Determinar si se ha obtenido el bloqueo. Para las solicitudes de lectura, si todos los nodos secundarios con número de secuencia menor que el propio son solicitudes de lectura, o no hay nodos secundarios con número de secuencia menor que el propio, indica que se ha adquirido con éxito el bloqueo compartido y se comienza a ejecutar la lógica de lectura. Para las solicitudes de escritura, si uno no es el nodo secundario con el número de secuencia más pequeño, entonces entra en espera.

4. Si no se ha obtenido el bloqueo compartido, la solicitud de lectura registra un watcher de escucha en el último nodo de solicitud de escritura con número de secuencia menor que el propio; la solicitud de escritura registra un watcher de escucha en el último nodo con número de secuencia menor que el propio.

En el proceso de desarrollo real, podemos usar la API encapsulada en el kit de herramientas curator para implementar bloqueos distribuidos.

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

Varios esquemas de bloqueo de curator:

  • 1, InterProcessMutex: bloqueo exclusivo reentrante distribuido
  • 2, InterProcessSemaphoreMutex: bloqueo exclusivo distribuido
  • 3, InterProcessReadWriteLock: bloqueo de lectura-escritura distribuido

El siguiente ejemplo simula 50 hilos que usan el bloqueo exclusivo reentrante InterProcessMutex para competir por el bloqueo al mismo tiempo:

Ejemplo

public class InterprocessLock {
    public static void main(String[] args)  {
        CuratorFramework zkClient = getZkClient();
        String lockPath = "/lock";
        InterProcessMutex lock = new InterProcessMutex(zkClient, lockPath);
        // Simular 50 hilos compitiendo por el bloqueo
        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("El hilo "+threadFlag+" ha adquirido el bloqueo");
                // Esperar 1 segundo y luego liberar el bloqueo
                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 consola emite un registro cada segundo:

Descargar el código fuente de este capítulo

Download