分库分表会带来哪些问题如何解决

public class SnowflakeIdGenerator {
    // 起始时间戳 (2020-01-01)
    private static final long START_TIMESTAMP = 1577836800000L;
    // 机器ID位数
    private static final long WORKER_ID_BITS = 5L;
    // 数据中心ID位数
    private static final long DATACENTER_ID_BITS = 5L;
    // 序列号位数
    private static final long SEQUENCE_BITS = 12L;

    // 最大值
    private static final long MAX_WORKER_ID = ~(-1L << WORKER_ID_BITS);
    private static final long MAX_DATACENTER_ID = ~(-1L << DATACENTER_ID_BITS);

    // 偏移量
    private static final long WORKER_ID_SHIFT = SEQUENCE_BITS;
    private static final long DATACENTER_ID_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS;
    private static final long TIMESTAMP_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS + DATACENTER_ID_BITS;

    // 序列号掩码
    private static final long SEQUENCE_MASK = ~(-1L << SEQUENCE_BITS);

    private final long workerId;
    private final long datacenterId;
    private long sequence = 0L;
    private long lastTimestamp = -1L;

    public SnowflakeIdGenerator(long workerId, long datacenterId) {
        if (workerId > MAX_WORKER_ID || workerId < 0) {
            throw new IllegalArgumentException("Worker ID超出范围");
        }
        if (datacenterId > MAX_DATACENTER_ID || datacenterId < 0) {
            throw new IllegalArgumentException("Datacenter ID超出范围");
        }
        this.workerId = workerId;
        this.datacenterId = datacenterId;
    }

    public synchronized long nextId() {
        long currentTimestamp = System.currentTimeMillis();

        // 核心亮点：时钟回拨处理
        if (currentTimestamp < lastTimestamp) {
            long offset = lastTimestamp - currentTimestamp;
            // 回拨小于5ms，等待恢复
            if (offset <= 5) {
                try {
                    wait(offset << 1);
                    currentTimestamp = System.currentTimeMillis();
                    if (currentTimestamp < lastTimestamp) {
                        throw new RuntimeException("时钟回拨严重，无法生成ID");
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException("线程中断", e);
                }
            } else {
                throw new RuntimeException("时钟回拨超过5ms，无法生成ID");
            }
        }

        // 同一毫秒内序列号自增
        if (currentTimestamp == lastTimestamp) {
            sequence = (sequence + 1) & SEQUENCE_MASK;
            // 序列号溢出，等待下一毫秒
            if (sequence == 0) {
                currentTimestamp = waitNextMillis(lastTimestamp);
            }
        } else {
            sequence = 0L;
        }

        lastTimestamp = currentTimestamp;

        // 拼接64位ID
        return ((currentTimestamp - START_TIMESTAMP) << TIMESTAMP_SHIFT)
                | (datacenterId << DATACENTER_ID_SHIFT)
                | (workerId << WORKER_ID_SHIFT)
                | sequence;
    }

    private long waitNextMillis(long lastTimestamp) {
        long timestamp = System.currentTimeMillis();
        while (timestamp <= lastTimestamp) {
            timestamp = System.currentTimeMillis();
        }
        return timestamp;
    }
}

/**
 * 复合分片策略：按用户ID哈希分表 + 按创建时间分库
 * 优点：数据分布均匀，支持按时间快速查询历史数据
 */
public class UserOrderComplexShardingStrategy implements ComplexKeysShardingAlgorithm<Long> {
    // 分库数量
    private static final int DB_COUNT = 4;
    // 每个库的分表数量
    private static final int TABLE_COUNT_PER_DB = 8;

    @Override
    public Collection<String> doSharding(Collection<String> availableTargetNames, ComplexKeysShardingValue<Long> shardingValue) {
        // 获取分片键值
        Map<String, Collection<Long>> valueMap = shardingValue.getColumnNameAndShardingValuesMap();
        Long userId = valueMap.get("user_id").iterator().next();
        Long createTime = valueMap.get("create_time").iterator().next();

        // 按时间分库：每半年一个库
        LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(createTime), ZoneId.systemDefault());
        int year = time.getYear() - 2020;
        int month = time.getMonthValue();
        int dbIndex = (year * 2 + (month > 6 ? 1 : 0)) % DB_COUNT;

        // 按用户ID哈希分表
        int tableIndex = Math.abs(userId.hashCode()) % TABLE_COUNT_PER_DB;

        // 拼接实际表名
        String actualTableName = String.format("order_db_%d.order_%d", dbIndex, tableIndex);
        return Collections.singletonList(actualTableName);
    }

    @Override
    public void init(Properties properties) {}

    @Override
    public String getType() {
        return "USER_ORDER_COMPLEX_SHARDING";
    }
}

@Service
public class OrderPaginationService {
    @Autowired
    private OrderMapper orderMapper;

    /**
     * 游标分页查询（性能比offset分页高100倍以上）
     * 对比：offset分页需要扫描前N条数据，游标分页直接定位到起始位置
     */
    public PageResult<Order> getOrdersByCursor(Long userId, Long cursor, int pageSize) {
        List<Order> orders;
        if (cursor == null) {
            // 第一页，没有游标
            orders = orderMapper.selectFirstPage(userId, pageSize);
        } else {
            // 后续页，使用游标（上一页最后一条记录的ID）
            orders = orderMapper.selectByCursor(userId, cursor, pageSize);
        }

        // 计算是否有下一页
        boolean hasNext = orders.size() == pageSize;
        // 获取下一页的游标
        Long nextCursor = hasNext ? orders.get(orders.size() - 1).getId() : null;

        return new PageResult<>(orders, hasNext, nextCursor);
    }
}

// Mapper接口
public interface OrderMapper {
    @Select("SELECT * FROM order WHERE user_id = #{userId} ORDER BY id DESC LIMIT #{pageSize}")
    List<Order> selectFirstPage(@Param("userId") Long userId, @Param("pageSize") int pageSize);

    @Select("SELECT * FROM order WHERE user_id = #{userId} AND id < #{cursor} ORDER BY id DESC LIMIT #{pageSize}")
    List<Order> selectByCursor(@Param("userId") Long userId, @Param("cursor") Long cursor, @Param("pageSize") int pageSize);
}

// 分页结果类
@Data
@AllArgsConstructor
public class PageResult<T> {
    private List<T> data;
    private boolean hasNext;
    private Long nextCursor;
}

public class SnowflakeIdGenerator {
    private final long epoch = 1577836800000L; // 2020-01-01
    private final long workerIdBits = 10L;
    private final long sequenceBits = 12L;
    private final long maxWorkerId = ~(-1L << workerIdBits);
    private final long sequenceMask = ~(-1L << sequenceBits);
    private final long workerIdShift = sequenceBits;
    private final long timestampShift = sequenceBits + workerIdBits;

    private long workerId;
    private long sequence = 0L;
    private long lastTimestamp = -1L;
    private final long maxBackwardMs = 500; // 允许最大回拨 500ms

    public SnowflakeIdGenerator(long workerId) {
        if (workerId > maxWorkerId || workerId < 0) throw new IllegalArgumentException();
        this.workerId = workerId;
    }

    public synchronized long nextId() {
        long timestamp = System.currentTimeMillis();
        // 🧨 时钟回拨防御
        if (timestamp < lastTimestamp) {
            long offset = lastTimestamp - timestamp;
            if (offset <= maxBackwardMs) {
                try { Thread.sleep(offset << 1); } // 等两倍差值
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
                timestamp = System.currentTimeMillis();
                if (timestamp < lastTimestamp) throw new RuntimeException("Clock moved backwards!");
            } else {
                throw new RuntimeException("Clock moved backwards too much!");
            }
        }
        if (timestamp == lastTimestamp) {
            sequence = (sequence + 1) & sequenceMask;
            if (sequence == 0) timestamp = tilNextMillis(lastTimestamp);
        } else {
            sequence = 0L;
        }
        lastTimestamp = timestamp;
        return ((timestamp - epoch) << timestampShift) | (workerId << workerIdShift) | sequence;
    }

    private long tilNextMillis(long lastTimestamp) {
        long ts = System.currentTimeMillis();
        while (ts <= lastTimestamp) ts = System.currentTimeMillis();
        return ts;
    }
}

public List<Order> scrollOrders(long lastId, int limit, List<Shard> shards) {
    int shardCount = shards.size();
    List<CompletableFuture<List<Order>>> futures = new ArrayList<>();
    
    // 并行查询每个分片
    for (Shard shard : shards) {
        futures.add(CompletableFuture.supplyAsync(() ->
            shard.query("SELECT * FROM order WHERE id > ? ORDER BY id LIMIT ?", lastId, limit)
        ));
    }
    
    // 内存归并，使用最小堆取全局最小的 N 条
    PriorityQueue<Order> minHeap = new PriorityQueue<>(Comparator.comparingLong(Order::getId));
    for (CompletableFuture<List<Order>> future : futures) {
        List<Order> part = future.join();
        if (part.size() > limit) part = part.subList(0, limit); // 防爆
        minHeap.addAll(part);
    }
    
    List<Order> result = new ArrayList<>(limit);
    while (!minHeap.isEmpty() && result.size() < limit) {
        result.add(minHeap.poll());
    }
    return result;
}

// Try 阶段
@Transactional
public boolean tryFreeze(String orderId, long skuId, int count) {
    // 幂等：先查流水
    if (frozenRecordDao.exists(orderId)) return true;
    // Redis 预占库存，原子操作
    Long remain = redis.decrBy("stock:sku:" + skuId, count);
    if (remain == null || remain < 0) {
        redis.incrBy("stock:sku:" + skuId, count); // 回滚 Redis
        return false;
    }
    // 写冻结流水（和订单同库，本地事务）
    frozenRecordDao.insert(new FrozenRecord(orderId, skuId, count));
    return true;
}

// Confirm 阶段
@Transactional
public void confirm(String orderId) {
    FrozenRecord record = frozenRecordDao.selectForUpdate(orderId);
    if (record == null || record.getStatus() == CONFIRMED) return; // 幂等
    // 正式扣减数据库库存
    stockDao.decrease(record.getSkuId(), record.getCount());
    // 标记完成
    record.setStatus(CONFIRMED);
    frozenRecordDao.update(record);
    // Redis 不需要管，预占时已经扣了
}

// Cancel 阶段
@Transactional
public void cancel(String orderId) {
    FrozenRecord record = frozenRecordDao.selectForUpdate(orderId);
    if (record == null || record.getStatus() == CANCELLED) return; // 幂等
    // 释放 Redis 库存
    redis.incrBy("stock:sku:" + record.getSkuId(), record.getCount());
    record.setStatus(CANCELLED);
    frozenRecordDao.update(record);
}

public class ConsistentHashRouter {
    private final TreeMap<Integer, Shard> ring = new TreeMap<>();
    private final int virtualNodeCnt;
    private final List<Shard> newShardGroup = null; // 扩容时临时持有新集群

    public Shard route(long shardKey) {
        int hash = hash(shardKey);
        if (ring.ceilingEntry(hash) != null) return ring.ceilingEntry(hash).getValue();
        return ring.firstEntry().getValue();
    }

    // 双写：同时写入新旧集群
    public void insertWithDualWrite(Order order) {
        Shard oldShard = route(order.getUserId());
        oldShard.insert(order);
        if (newShardGroup != null) {
            Shard newShard = newConsistentHashRoute(order.getUserId());
            newShard.insert(order);
        }
    }
}