CRDTs - Conflict-Free Replicated Data Types

Giriş

Conflict-free replicated data types( CRDTs) are a family of data structures for sets, maps, ordered lists, counters, etc that can be concurrently edited by multiple users. It uses two-way merges.

İlgili bir yazı burada

Örnek

Şöyle olsun

import java.util.Objects;

public record ReplicaId(String value) {
    public ReplicaId {
        if (value == null || value.isBlank()) {
            throw new IllegalArgumentException("Replica ID must not be blank");
        }
    }
}  

Şöyle olsun

public sealed interface CounterCommand
        permits CounterCommand.Increment, CounterCommand.Decrement {
    ReplicaId replica();
    record Increment(ReplicaId replica) implements CounterCommand {
        public Increment {
            java.util.Objects.requireNonNull(replica);
        }
    }
    record Decrement(ReplicaId replica) implements CounterCommand {
        public Decrement {
            java.util.Objects.requireNonNull(replica);
        }
    }
}     

Gerçekleştirim şöyle.

Instead of storing one mutable total, it stores two maps:

- positive: the cumulative increments made by each replica.

- negative: the cumulative decrements made by each replica.

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

public record PnCounter(
        Map<ReplicaId, Long> positive,
        Map<ReplicaId, Long> negative) {

    public PnCounter {
        positive = immutableNonNegative(positive);
        negative = immutableNonNegative(negative);
    }

    public static PnCounter empty() {
        return new PnCounter(Map.of(), Map.of());
    }

    public PnCounter apply(CounterCommand command) {
        Objects.requireNonNull(command);

        return switch (command) {
            case CounterCommand.Increment inc ->
                    new PnCounter(bump(positive, inc.replica()), negative);
            case CounterCommand.Decrement dec ->
                    new PnCounter(positive, bump(negative, dec.replica()));
        };
    }

    public PnCounter merge(PnCounter other) {
        Objects.requireNonNull(other);
        return new PnCounter(
                joinByMaximum(positive, other.positive),
                joinByMaximum(negative, other.negative));
    }

    public long value() {
        return Math.subtractExact(sum(positive), sum(negative));
    }

    private static Map<ReplicaId, Long> bump(
            Map<ReplicaId, Long> current,
            ReplicaId replica) {

        var next = new HashMap<>(current);
        next.merge(replica, 1L, Math::addExact);
        return Map.copyOf(next);
    }

    private static Map<ReplicaId, Long> joinByMaximum(
            Map<ReplicaId, Long> left,
            Map<ReplicaId, Long> right) {

        var merged = new HashMap<>(left);
        right.forEach((replica, count) ->
                merged.merge(replica, count, Math::max));
        return Map.copyOf(merged);
    }

    private static long sum(Map<ReplicaId, Long> components) {
        return components.values().stream()
                .mapToLong(Long::longValue)
                .reduce(0L, Math::addExact);
    }

    private static Map<ReplicaId, Long> immutableNonNegative(
            Map<ReplicaId, Long> source) {

        Objects.requireNonNull(source);
        Map<ReplicaId, Long> copy = Map.copyOf(source);
        copy.forEach((replica, count) -> {
            Objects.requireNonNull(replica);
            if (count < 0) {
                throw new IllegalArgumentException("Components must be non-negative");
            }
        });
        return copy;
    }
}