Spring Boot ShardingSphere Integration: Sharding Strategies, Global IDs & Cross-Shard Query Pitfalls

When to Shard

Single-table MySQL performance degrades beyond 20 million rows or 500 GB: B+ tree depth increases, Buffer Pool hit rate drops, replication lag becomes visible. Sharding distributes write bottlenecks, trading space for time. Apache ShardingSphere-JDBC uses a lightweight fat-client approach: add the JAR, write configuration, and business code remains largely unchanged while the framework handles SQL routing, rewriting, and result merging.

Core Configuration & Logical Mapping (ShardingSphere 5.x)

ShardingSphere-JDBC processes SQL in four steps: parse → route → rewrite → merge. Developers operate on logical tables; the framework maps to physical nodes.

Dependencies

<dependency>
  <groupId>org.apache.shardingsphere</groupId>
  <artifactId>shardingsphere-jdbc-core-spring-boot-starter</artifactId>
  <version>5.5.0</version>
</dependency>
<!-- Persistence layer: JPA or MyBatis -->
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>

YAML Configuration (Read-Write Splitting + Sharding)

spring:
  shardingsphere:
    datasource:
      names: ds-write,ds-read-01,ds-read-02
      ds-write:
        jdbc-url: jdbc:mysql://primary-host:3306/order_db?useSSL=false&serverTimezone=Asia/Shanghai
        driver-class-name: com.mysql.cj.jdbc.Driver
        username: root
        password: ${DB_WRITE_PWD}
      ds-read-01:
        jdbc-url: jdbc:mysql://replica01-host:3306/order_db?useSSL=false&serverTimezone=Asia/Shanghai
        driver-class-name: com.mysql.cj.jdbc.Driver
        username: reader
        password: ${DB_READ_PWD}
      ds-read-02:
        jdbc-url: jdbc:mysql://replica02-host:3306/order_db?useSSL=false&serverTimezone=Asia/Shanghai
        driver-class-name: com.mysql.cj.jdbc.Driver
        username: reader
        password: ${DB_READ_PWD}
    rules:
      sharding:
        tables:
          t_order:
            actual-data-nodes: ds$->{0..1}.t_order_$->{0..1}
            table-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: t_order_inline
            database-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: db_inline
        sharding-algorithms:
          t_order_inline:
            type: INLINE
            props:
              algorithm-expression: t_order_$->{user_id % 2}
          db_inline:
            type: INLINE
            props:
              algorithm-expression: ds$->{user_id % 2}
        binding-tables:
          - t_order,t_order_item
      readwrite-splitting:
        data-sources:
          readwrite_ds:
            write-data-source-name: ds-write
            read-data-source-names: ds-read-01,ds-read-02
            load-balancer-name: round_robin
        load-balancers:
          round_robin:
            type: ROUND_ROBIN

Key concepts:

Logical table : e.g., t_order in code, transparent to ORM.

Actual data nodes : physical tables like ds0.t_order_0, dynamically assembled by the algorithm.

Binding tables : parent-child tables (order & order item) with same sharding key; JOIN routes to a single physical database, avoiding Cartesian explosion.

Broadcast tables : small, rarely changing tables (dictionaries, region codes) replicated to every database for local JOINs.

Choosing a Sharding Algorithm

Wrong algorithm choice makes later scaling a disaster. Prioritize query patterns over perfect uniformity.

Hash Modulo

Most common: user_id % N. Simple, even distribution. Drawback: scaling changes all remainders, requiring data migration. Range scans ( WHERE create_time BETWEEN ...) become full-shard broadcasts. Recommendation: start with 32 or 64 shards, later merge logically or use dual-write transition.

Range Sharding

Suitable for naturally increasing fields (serial numbers, billing cycles). Implement SPI StandardShardingAlgorithm<Long>. Example: split every 1 million IDs. Excellent range query performance, high index hit rate. Write hotspot is severe: latest data hits one table, causing page splits and lock contention. Compromise: composite range+hash routing or isolate hot tables.

// 5.x Range Sharding SPI Example
public class RangeShardingAlgorithm implements StandardShardingAlgorithm<Long> {
  @Override
  public String doSharding(Collection<String> availableTargetNames,
      PreciseShardingValue<Long> preciseShardingValue) {
    long id = preciseShardingValue.getValue();
    long index = (id / 1000000L) % availableTargetNames.size();
    return availableTargetNames.stream()
        .filter(n -> n.endsWith(String.valueOf(index)))
        .findFirst().orElseThrow(() -> new IllegalArgumentException("No valid target"));
  }
}

Time-Based Sharding

Partition by year/month/day for logs and flow archives. Old data can be dropped or moved to cold storage, clean lifecycle management. Cross-year/month queries require multi-table UNION; combine with secondary indexes or Elasticsearch for fallback.

Scaling procedure (four steps): enable dual-write at application layer → full migration + incremental Binlog catch-up → data sampling verification (MD5/Checksum) → gray-scale read cutover, then write cutover. Keep old topology for two weeks as rollback window.

Global ID Generation: Don't Rely on Auto-Increment

After sharding, AUTO_INCREMENT fails; a distributed ID generator is mandatory.

Snowflake Algorithm

64-bit: timestamp + machine bits + sequence. Handles millions of QPS. Main risk: clock drift. ShardingSphere's built-in implementation adds max-vibration-offset and a tolerance window; configure ~500 ms. If NTP drift is severe, wrap a layer that throws exceptions or falls back on drift.

Segment Mode

Fetch ID ranges from database (e.g., [1001, 2000]), increment in memory. Meituan Leaf's DB mode uses this. Advantages: strictly monotonic IDs (friendly to MySQL page splits), immune to clock issues. Cost: maintain a segment table, but DB pressure is tiny because one fetch serves thousands of IDs.

Production integration with Leaf: use official Starter; manual @Bean wiring risks missing initialization hooks. High availability: multi-node deployment + proactive local segment refresh (fetch new segment at 20% remaining). On network/DB failure, degrade to Snowflake (may lose monotonicity but keeps business running).

Read-Write Splitting & Consistency Traps

Read-heavy scenarios use read-write splitting. ShardingSphere routes by SQL type: INSERT/UPDATE/DELETE → primary; SELECT → replicas. Load balancers: round-robin, random, weight. TRANSACTION_RANDOM maintains session stickiness within a transaction.

Replica lag (50 ms to seconds) causes "write then read, but read stale data" complaints. Solutions:

Force primary for critical paths : use HintManager.setWriteRouteOnly(); must call close() in finally to clear ThreadLocal, else subsequent requests all hit primary and overload it.

HintManager hintManager = HintManager.getInstance();
try {
  hintManager.setWriteRouteOnly();
  orderService.create(dto);
} finally {
  hintManager.close();
}

Tiered latency tolerance : payment callbacks, order status changes → FORCE_MASTER; product listings, reviews → replica lag acceptable.

Session stickiness : route same user to fixed replica via Nginx cookie or gateway session, reducing read-your-writes inconsistency probability.

Cross-Shard Queries & Slow SQL Governance

Sharding turns single-SQL operations into multi-shard merges. GROUP BY, ORDER BY, LIMIT without sharding key trigger full routing, risking OOM.

Framework uses stream merge and memory merge. Aggregations ( COUNT/SUM/MAX) push down; AVG rewrites to SUM()/COUNT(). Worst case: ORDER BY ... LIMIT offset, size with large offset forces each shard to full scan and sort, performance collapses.

Practical Optimizations

Binding tables mandatory : same sharding key on parent/child tables routes JOIN to one database, halving physical connections.

Ruthless sharding key selection : reject SQL missing sharding key. SaaS multi-tenant often uses tenant_id + user_id composite routing; isolate large tenants to dedicated databases to avoid skew.

Deep pagination alternatives : avoid LIMIT. Use cursor pagination WHERE id > last_max_id LIMIT 20 or fetch ID list then IN query. Offload list pages to Elasticsearch; primary handles only authoritative writes and transactions, cutting DB load by 60%+.

Slow SQL interception : enable sql-show: true to log routing; alert on full-shard scans. Set max-connections-size-per-query: 16 to cap per-query threads, preventing one bad SQL from exhausting the pool.

Connection Pool & Distributed Transactions

Sharding inflates connection counts: 10 shards × 20 connections = 200 connections. MySQL max_connections is typically a few hundred; Too many connections errors appear easily.

HikariCP tuning: don't follow formulas, rely on load tests. Per-shard maximum-pool-size 15–25, minimum-idle: 5, sensible idle-timeout. Startup connection storms are common; increase initialization-fail-timeout or set spring.datasource.hikari.initializationFailTimeout: -1 to avoid health checks blowing up the pool.

Avoid distributed transactions when possible. XA (strong consistency) costs 30%+ performance; two-phase commit hangs are frequent. Prefer local transactions + MQ eventual consistency: order persists, emit reliable message, inventory/points consume asynchronously with compensation. If cross-shard strong consistency is unavoidable, use Seata-AT; configure undo_log table, set timeout to business-acceptable upper bound, don't rely on defaults.

Production Checklist

[ ] Does sharding key cover 90%+ of query conditions?

[ ] Are binding table configs strictly consistent with physical sharding rules?

[ ] Is every HintManager usage closed in a finally block? ThreadLocal leaks are silent bombs.

[ ] Is replica lag monitoring in place? Any fallback for read-your-writes inconsistency?

[ ] Is (connection pool size × shard count) < 80% of DB max connections?

[ ] Does global ID generator have fallback for clock drift / segment exhaustion?

Sharding is not technical showmanship; it's a compromise forced by data scale. Early-stage business should not rush to split: single-table index optimization, cache layering, archiving strategies handle 80% of cases. Wait until write TPS truly hits a wall and I/O bottlenecks are visible, then evolve smoothly using these standards. Architecture has no optimal solution, only the best balance for current business scale and team ops capability.