Scaling MySQL at JD Logistics with ShardingSphere: Refactoring Sharding-JDBC

Background

JD Logistics' delivery order fulfillment centers (cp‑eofc) and logistics platform fulfillment center (jdl‑uep‑ofc) use ShardingSphere's sharding‑jdbc as a sharding middleware. The cluster consists of 16 MySQL instances, each with 32 databases, totaling 512 databases.

When a client host is added, each MySQL instance requires at least 32 connections; connection pools can multiply this by 5‑10×. Multiple applications (web, provider, worker) share the same data source, quickly reaching the MySQL connection limit and preventing horizontal scaling.

Rapid business growth makes this a critical bottleneck.

Sharding concepts

Why sharding

Database sharding

As data volume grows, a single database’s QPS and read/write latency increase, becoming a performance bottleneck. Splitting a database into multiple instances distributes load and improves availability.

Table sharding

When a single table becomes large, query and update performance degrades. Horizontal splitting of the table into multiple tables reduces per‑table size and improves performance.

Sharding‑JDBC overview

ShardingSphere is an open‑source distributed database middleware ecosystem composed of Sharding‑JDBC, Sharding‑Proxy, and (planned) Sharding‑Sidecar. It provides standardized data sharding, distributed transactions, and governance, suitable for Java and other environments.

Sharding‑JDBC is a lightweight Java framework that acts as an enhanced JDBC driver, requiring no extra deployment. It works with any Java ORM (JPA, Hibernate, MyBatis, Spring JDBC Template) and any JDBC‑compatible connection pool (DBCP, C3P0, Druid, HikariCP, etc.). It supports MySQL, Oracle, SQL Server, and PostgreSQL.

Example Spring XML configuration:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:sharding="http://shardingsphere.io/schema/shardingsphere/sharding"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
                           http://www.springframework.org/schema/beans/spring-beans.xsd
                           http://shardingsphere.io/schema/shardingsphere/sharding
                           http://shardingsphere.io/schema/shardingsphere/sharding/sharding.xsd">
    <!-- Data source ds0 -->
    <bean id="ds0" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
        <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
        <property name="url" value="jdbc:mysql://localhost:3306/ds0"/>
        <property name="username" value="root"/>
        <property name="password" value=""/>
    </bean>
    <!-- Data source ds1 -->
    <bean id="ds1" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
        <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
        <property name="url" value="jdbc:mysql://localhost:3306/ds1"/>
        <property name="username" value="root"/>
        <property name="password" value=""/>
    </bean>
    <sharding:inline-strategy id="databaseStrategy" sharding-column="user_id"
                               algorithm-expression="ds$->{user_id % 2}"/>
    <sharding:inline-strategy id="orderTableStrategy" sharding-column="order_id"
                               algorithm-expression="t_order$->{order_id % 2}"/>
    <sharding:inline-strategy id="orderItemTableStrategy" sharding-column="order_id"
                               algorithm-expression="t_order_item$->{order_id % 2}"/>
    <sharding:data-source id="shardingDataSource">
        <sharding:sharding-rule data-source-names="ds0,ds1">
            <sharding:table-rules>
                <sharding:table-rule logic-table="t_order"
                                   actual-data-nodes="ds$->{0..1}.t_order$->{0..1}"
                                   database-strategy-ref="databaseStrategy"
                                   table-strategy-ref="orderTableStrategy"/>
                <sharding:table-rule logic-table="t_order_item"
                                   actual-data-nodes="ds$->{0..1}.t_order_item$->{0..1}"
                                   database-strategy-ref="databaseStrategy"
                                   table-strategy-ref="orderItemTableStrategy"/>
            </sharding:table-rules>
        </sharding:sharding-rule>
    </sharding:data-source>
</beans>

Configuration summary:

Define multiple data sources (ds0, ds1).

Specify sharding column and Groovy‑style algorithm expression.

In <sharding:table-rule>, set logical table name, database and table strategies, and compose actual data nodes.

Problem analysis and solution

Analysis

The original MySQL cluster has 16 instances × 32 databases = 512 databases. Each client creates 32 connections per instance; with a pool max of 5, a single client can open up to 160 connections per instance, quickly hitting the limit during large‑scale promotions.

Current connection topology is shown below:

Feasible options

Single‑instance, no database sharding, only table sharding – reduces connections but requires full data migration and risky cut‑over.

Adopt elastic databases such as JD’s JED or TiDB – offloads connection management to the database layer.

Deploy Sharding‑Proxy – clients connect to the proxy, which then manages MySQL connections, decoupling client count from MySQL.

Refactor Sharding‑JDBC – customize routing to allow a client to access multiple databases through a single connection.

Considering cost and risk, the team chose to refactor Sharding‑JDBC.

Sharding‑JDBC investigation

Workflow

SQL parsing (lexical and syntactic).

SQL routing – match sharding strategy and generate route.

SQL rewrite – transform logical SQL to executable SQL.

SQL execution – multi‑threaded execution.

Result merging – combine results from multiple nodes.

Source code analysis

The ShardingStandardRoutingEngine class’s route method is the entry point, delegating to routeDataSources (database routing) and routeTables (table routing). The actual sharding calculation occurs in StandardShardingStrategy.doSharding, which uses either a precise or range algorithm. Implementing a custom PreciseShardingAlgorithm enables bespoke sharding logic.

Refactoring steps

Database sharding

Reduce data sources from ds_0‑ds_511 to ds_0‑ds_15 (16 sources), each pointing to the first database on an instance. Update the Groovy expression for the sharding key order_code from (Math.abs(order_code.hashCode()) % 512) to (Math.abs(order_code.hashCode()) % 512).intdiv(32) to map to 0‑15.

Table sharding

Implement PreciseShardingAlgorithm to return “actualDatabase.table”. For example, querying t_order with user_id=35711 on DB_31 yields data source “DB_0” and table “DB_31.t_order”.

Define <sharding:standard-strategy> in XML and reference the custom algorithm.

Connection pool adjustment

Before refactoring, each database had its own pool; after refactoring, 32 databases share a single pool per instance. Pool size parameters must be tuned based on traffic and load testing.

Lesson learned

When adjusting the Groovy expression, using “/32” produced fractional results (e.g., ds_14.6857) leading to “no database route info” errors. The correct integer division operator is .intdiv(32).

Conclusion

The article presented sharding concepts, advantages, and a deep dive into Sharding‑JDBC routing. It emphasized that the decision to shard databases or tables depends on system characteristics such as QPS, data volume, storage, retention, and business growth.