#数据分区策略

public class HashPartitioner {
    private final int numPartitions;

    public HashPartitioner(int numPartitions) {
        this.numPartitions = numPartitions;
    }

    public int getPartition(String key) {
        // 字符串哈希取模
        int hash = key.hashCode();
        // 解决负数问题
        return (hash & 0x7FFFFFFF) % numPartitions;
    }

    public int getPartition(long key) {
        // 长整型取模
        return ((int) (key ^ (key >>> 32))) & 0x7FFFFFFF) % numPartitions;
    }
}

public class RangePartitioner {
    private final List<Range> ranges;

    public RangePartitioner(List<Range> ranges) {
        this.ranges = ranges;
    }

    public int getPartition(long key) {
        for (int i = 0; i < ranges.size(); i++) {
            if (key >= ranges.get(i).start && key < ranges.get(i).end) {
                return i;
            }
        }
        throw new IllegalArgumentException("Key out of range: " + key);
    }

    // 按时间范围分区示例
    public int getPartitionByTime(LocalDateTime time) {
        int year = time.getYear();
        int month = time.getMonthValue();
        // 按年月分区：2024-01 -> 分区0, 2024-02 -> 分区1
        return (year - 2020) * 12 + month - 1;
    }
}

// 热点检测与告警
public class HotPartitionDetector {
    public void detectAndAlert() {
        Map<String, Long> partitionSizes = getPartitionSizes();

        long avgSize = calculateAverage(partitionSizes.values());
        long threshold = avgSize * 3; // 超过平均值3倍视为热点

        partitionSizes.entrySet().stream()
            .filter(e -> e.getValue() > threshold)
            .forEach(e -> alert("Hot partition: " + e.getKey() +
                               ", size: " + formatSize(e.getValue())));
    }
}

public class ConsistentHashRing<T> {
    private final SortedMap<Long, T> ring = new TreeMap<>();
    private final int virtualNodes; // 虚拟节点数

    public ConsistentHashRing(Collection<T> nodes, int virtualNodes) {
        this.virtualNodes = virtualNodes;
        for (T node : nodes) {
            addNode(node);
        }
    }

    public void addNode(T node) {
        // 为每个物理节点创建多个虚拟节点
        for (int i = 0; i < virtualNodes; i++) {
            long hash = hash(node.toString() + "_vn_" + i);
            ring.put(hash, node);
        }
    }

    public T getNode(String key) {
        if (ring.isEmpty()) {
            throw new IllegalStateException("Ring is empty");
        }

        long hash = hash(key);
        // 找到第一个大于等于 hash 的节点
        Map.Entry<Long, T> entry = ring.ceilingEntry(hash);

        // 如果没有，找到环的第一个节点
        if (entry == null) {
            entry = ring.firstEntry();
        }

        return entry.getValue();
    }

    private long hash(String key) {
        // MurmurHash 或其他均匀分布的哈希算法
        return MurmurHash.hash(key);
    }
}

// 虚拟节点 vs 物理节点
public class VirtualNodeDemo {
    // 物理节点只有 3 个
    // 但虚拟节点有 150 个（每个物理节点 50 个）
    // 虚拟节点均匀分布在环上 → 数据均匀分布

    public static void main(String[] args) {
        // 物理节点
        List<String> physicalNodes = Arrays.asList(
            "192.168.1.10",
            "192.168.1.11",
            "192.168.1.12"
        );

        // 每个物理节点 50 个虚拟节点
        ConsistentHashRing<String> ring = new ConsistentHashRing<>(
            physicalNodes, 50
        );

        // 测试数据分布
        Map<String, Integer> distribution = new HashMap<>();
        for (int i = 0; i < 10000; i++) {
            String key = "user_" + i;
            String node = ring.getNode(key);
            distribution.merge(node, 1, Integer::sum);
        }

        // 打印分布（应该接近均匀：3300, 3300, 3400）
        distribution.forEach((node, count) ->
            System.out.println(node + ": " + count + " (" +
                String.format("%.1f", count * 100.0 / 10000) + "%)")
        );
    }
}

// 跨分区分页的坑
public class CrossPartitionPagination {
    // 错误示例：跨分区分页导致数据重复或遗漏
    public PageResult<User> badPagination(List<Shard> shards, int page, int size) {
        List<User> allUsers = new ArrayList<>();

        // 第一步：从每个分片获取第一页
        for (Shard shard : shards) {
            List<User> users = shard.query(0, size);  // 每页 size 条
            allUsers.addAll(users);
        }

        // 第二步：本地分页
        int fromIndex = (page - 1) * size;
        return new PageResult<>(
            allUsers.subList(fromIndex, Math.min(fromIndex + size, allUsers.size())),
            allUsers.size()
        );
    }

    // 问题：第二页可能包含第一页已经返回过的数据！
    // 因为各分片的"第一页"是独立的
}

// 游标分页解决方案
public class CursorPagination {
    public List<Order> paginateByCursor(Long userId, String cursor, int size) {
        if (cursor == null) {
            // 首次查询：定位分区，然后按时间排序
            int shardId = getShardIdByUserId(userId);
            return queryShard(shardId, null, size);
        }

        // 后续查询：使用游标直接定位
        Order lastOrder = decodeCursor(cursor);  // 解析游标
        int shardId = getShardIdByUserId(userId);
        return queryShard(shardId, lastOrder.getCreatedAt(), size);
    }

    private String encodeCursor(Order order) {
        // 游标 = 分片ID + 时间戳 + 订单ID（确保唯一性）
        return String.format("%d_%d_%d",
            order.getShardId(),
            order.getCreatedAt().getTime(),
            order.getId()
        );
    }
}

public class ConsistentHashRing<T> {
    private final TreeMap<Long, T> ring = new TreeMap<>();
    private final HashFunction hashFunction;
    private final int virtualNodes;

    public ConsistentHashRing(Collection<T> nodes, int virtualNodes) {
        this(nodes, virtualNodes, MurmurHash::hash);
    }

    public ConsistentHashRing(Collection<T> nodes, int virtualNodes,
                            HashFunction hashFunction) {
        this.virtualNodes = virtualNodes;
        this.hashFunction = hashFunction;

        for (T node : nodes) {
            addNode(node);
        }
    }

    public void addNode(T node) {
        for (int i = 0; i < virtualNodes; i++) {
            long hash = hashFunction.hash(node.toString() + "_vn_" + i);
            ring.put(hash, node);
        }
    }

    public void removeNode(T node) {
        for (int i = 0; i < virtualNodes; i++) {
            long hash = hashFunction.hash(node.toString() + "_vn_" + i);
            ring.remove(hash);
        }
    }

    public T getNode(String key) {
        if (ring.isEmpty()) {
            throw new IllegalStateException("Ring is empty");
        }

        long hash = hashFunction.hash(key);
        Map.Entry<Long, T> entry = ring.ceilingEntry(hash);

        if (entry == null) {
            entry = ring.firstEntry();
        }

        return entry.getValue();
    }

    @FunctionalInterface
    public interface HashFunction {
        long hash(String key);
    }

    // 测试
    public static void main(String[] args) {
        List<String> nodes = Arrays.asList("NodeA", "NodeB", "NodeC");
        ConsistentHashRing<String> ring = new ConsistentHashRing<>(nodes, 100);

        Map<String, Integer> distribution = new HashMap<>();
        for (int i = 0; i < 10000; i++) {
            String key = "user_" + i;
            String node = ring.getNode(key);
            distribution.merge(node, 1, Integer::sum);
        }

        distribution.forEach((node, count) ->
            System.out.printf("%s: %d (%.1f%%)%n", node, count, count * 100.0 / 10000)
        );
    }
}