SQL Distributed Lock¶

SQL connector enables distributed locking on a relational databases. It was tested on PostgreSQL v11 and MySQL v8.

Read and write from the same DB node

Make sure that DB connection passed to Sherlock reads and writes to the same DB node so every lock change is visible to all of your services.

Usage¶

Add dependencies to build.gradle.kts:

SyncCoroutinesReactorRxJava

dependencies {
    implementation("com.coditory.sherlock:sherlock-sql:0.4.19")
    implementation("com.zaxxer:HikariCP:$versions.hikaricp")
    implementation("org.postgresql:postgresql:$versions.postgresql")
}

dependencies {
    implementation("com.coditory.sherlock:sherlock-sql-coroutines:0.4.19")
    implementation("org.postgresql:postgresql:$versions.postgres")
    implementation("com.zaxxer:HikariCP:$versions.hikaricp")
    implementation("org.postgresql:r2dbc-postgresql:$versions.r2dbc")
}

dependencies {
    implementation("com.coditory.sherlock:sherlock-sql-reactor:0.4.19")
    implementation("org.postgresql:postgresql:$versions.postgres")
    implementation("com.zaxxer:HikariCP:$versions.hikaricp")
    implementation("org.postgresql:r2dbc-postgresql:$versions.r2dbc")
}

dependencies {
    implementation("com.coditory.sherlock:sherlock-sql-rxjava:0.4.19")
    implementation("org.postgresql:postgresql:$versions.postgres")
    implementation("com.zaxxer:HikariCP:$versions.hikaricp")
    implementation("org.postgresql:r2dbc-postgresql:$versions.r2dbc")
}

Create sherlock instance and distributed lock:

SyncCoroutinesReactorRxJava

import com.coditory.sherlock.DistributedLock;
import com.coditory.sherlock.Sherlock;
import com.coditory.sherlock.sql.SqlSherlock;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;

public class PostgresSyncLockSample {
    private static final Logger logger = LoggerFactory.getLogger(PostgresSyncLockSample.class);

    private static DataSource dataSource() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:postgresql://localhost:5432/test");
        config.setUsername("postgres");
        config.setPassword("postgres");
        return new HikariDataSource(config);
    }

    public static void main(String[] args) {
        Sherlock sherlock = SqlSherlock.create(dataSource());
        DistributedLock lock = sherlock.createLock("sample-lock");
        lock.runLocked(() -> logger.info("Lock acquired!"));
    }
}

import com.coditory.sherlock.sql.BindingMapper.POSTGRES_MAPPER
import com.coditory.sherlock.sql.coroutines.SqlSherlock
import io.r2dbc.spi.ConnectionFactories
import io.r2dbc.spi.ConnectionFactory
import io.r2dbc.spi.ConnectionFactoryOptions
import kotlinx.coroutines.runBlocking
import org.slf4j.Logger
import org.slf4j.LoggerFactory

object PostgresKtLockSample {
    private val logger: Logger = LoggerFactory.getLogger(this.javaClass)

    private fun getConnectionFactory(): ConnectionFactory {
        val database = "test"
        val options =
            ConnectionFactoryOptions
                .parse("r2dbc:postgresql://localhost:5432/$database")
                .mutate()
                .option(ConnectionFactoryOptions.USER, "postgres")
                .option(ConnectionFactoryOptions.PASSWORD, "postgres")
                .option(ConnectionFactoryOptions.DATABASE, database)
                .build()
        return ConnectionFactories.get(options)
    }

    private suspend fun sample() {
        val sherlock = SqlSherlock.create(getConnectionFactory(), POSTGRES_MAPPER)
        val lock = sherlock.createLock("sample-lock")
        lock
            .runLocked { logger.info("Lock acquired!") }
    }

    @JvmStatic
    fun main(args: Array<String>) {
        runBlocking { sample() }
    }
}

import com.coditory.sherlock.reactor.DistributedLock;
import com.coditory.sherlock.reactor.Sherlock;
import com.coditory.sherlock.sql.reactor.SqlSherlock;
import io.r2dbc.spi.ConnectionFactories;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.ConnectionFactoryOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.coditory.sherlock.sql.BindingMapper.POSTGRES_MAPPER;

public class PostgresReactorLockSample {
    private static final Logger logger = LoggerFactory.getLogger(PostgresReactorLockSample.class);

    private static ConnectionFactory getConnectionFactory() {
        String database = "test";
        ConnectionFactoryOptions options = ConnectionFactoryOptions
                .parse("r2dbc:postgresql://localhost:5432/" + database)
                .mutate()
                .option(ConnectionFactoryOptions.USER, "postgres")
                .option(ConnectionFactoryOptions.PASSWORD, "postgres")
                .option(ConnectionFactoryOptions.DATABASE, database)
                .build();
        return ConnectionFactories.get(options);
    }

    public static void main(String[] args) {
        Sherlock sherlock = SqlSherlock.create(getConnectionFactory(), POSTGRES_MAPPER);
        DistributedLock lock = sherlock.createLock("sample-lock");
        lock.runLocked(() -> logger.info("Lock acquired!"))
                .block();
    }
}

import com.coditory.sherlock.rxjava.DistributedLock;
import com.coditory.sherlock.rxjava.Sherlock;
import com.coditory.sherlock.sql.rxjava.SqlSherlock;
import io.r2dbc.spi.ConnectionFactories;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.ConnectionFactoryOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.coditory.sherlock.sql.BindingMapper.POSTGRES_MAPPER;

public class PostgresRxLockSample {
    private static final Logger logger = LoggerFactory.getLogger(PostgresRxLockSample.class);

    private static ConnectionFactory getConnectionFactory() {
        String database = "test";
        ConnectionFactoryOptions options = ConnectionFactoryOptions
            .parse("r2dbc:postgresql://localhost:5432/" + database)
            .mutate()
            .option(ConnectionFactoryOptions.USER, "postgres")
            .option(ConnectionFactoryOptions.PASSWORD, "postgres")
            .option(ConnectionFactoryOptions.DATABASE, database)
            .build();
        return ConnectionFactories.get(options);
    }

    public static void main(String[] args) {
        Sherlock sherlock = SqlSherlock.create(getConnectionFactory(), POSTGRES_MAPPER);
        DistributedLock lock = sherlock.createLock("sample-lock");
        lock.runLocked(() -> logger.info("Lock acquired!"))
            .blockingGet();
    }
}

These examples use Hikari Connection Pool, but any implementation of java.sql.DataSource will suffice.

Learn more

See full source code example on Github.

Configuration¶

Configuration is available via sherlock builder:

SyncCoroutinesReactorRxJava

SqlSherlock.builder()
    .withClock(Clock.systemUTC())
    .withLockDuration(Duration.ofMinutes(5))
    .withOwnerIdPolicy(OwnerIdPolicy.uniqueOwnerId())
    .withDataSource(dataSource())
    .withLocksTable("LOCKS")
    .build();

SqlSherlock.builder()
    .withClock(Clock.systemUTC())
    .withLockDuration(Duration.ofMinutes(5))
    .withOwnerIdPolicy(OwnerIdPolicy.uniqueOwnerId())
    .withBindingMapper(BindingMapper.POSTGRES_MAPPER)
    .withConnectionFactory(getConnectionFactory())
    .withLocksTable("LOCKS")
    .build()

SqlSherlock.builder()
    .withClock(Clock.systemUTC())
    .withLockDuration(Duration.ofMinutes(5))
    .withOwnerIdPolicy(OwnerIdPolicy.uniqueOwnerId())
    .withBindingMapper(BindingMapper.POSTGRES_MAPPER)
    .withConnectionFactory(getConnectionFactory())
    .withLocksTable("LOCKS")
    .build();

SqlSherlock.builder()
    .withClock(Clock.systemUTC())
    .withLockDuration(Duration.ofMinutes(5))
    .withOwnerIdPolicy(OwnerIdPolicy.uniqueOwnerId())
    .withBindingMapper(BindingMapper.POSTGRES_MAPPER)
    .withConnectionFactory(getConnectionFactory())
    .withLocksTable("LOCKS")
    .build();

Parameters:

clock (default: Clock.systemUTC()) - used to generate acquisition and expiration timestamps.
lockDuration (default: Duration.ofMinutes(5)) - a default lock expiration time. If lock is not released and expiration time passes, the lock is treated as released.
ownerIdPolicy (default: uniqueOwnerId()) - used to generate lock owner id. It's executed once for every lock, during lock creation. There are different policies available for generating lock ownerIds.
locksTable (default: "LOCKS") - locks database name
connectionFactory (only in: Reactor, RxJava, Coroutines) - database connection factory
bindingMapper (only in: Reactor, RxJava, Coroutines) - specifies DB type for proper SQL query mapping
dataSource (only in synchronous) - database data source

Locks Table¶

Locks table is automatically created if it does not already exist. Table is created with a following SQL:

CREATE TABLE LOCKS (
  -- Lock id
  ID VARCHAR(100) NOT NULL,
  -- Owner id
  ACQUIRED_BY VARCHAR(100) NOT NULL,
  -- Lock acquisition moment
  ACQUIRED_AT TIMESTAMP(3) NOT NULL,
  -- Lock expiration time
  -- Might be null for locks that do not expire
  EXPIRES_AT TIMESTAMP(3),
  PRIMARY KEY (ID)
)
CREATE INDEX LOCKS_IDX ON LOCKS (ID, ACQUIRED_BY, EXPIRES_AT)