文章目录


基于spring boot3 + lombok 1.18.30 + mybatis plus 3.5.5 +nacos

Spring Boot 3 整合 Sharding-JDBC 分库分表实现

根据需求，我们需要实现两个核心功能：

  1. 分库：将聊天记录库和用户业务库分离 （提高主业务吞吐量）
  2. 分表：聊天记录表按用户ID（deviceId）分为10个表 （保证单表不超过500万数据量）

下面是核心配置和业务代码实现：

Nacos配置中心配置

在Nacos中需要添加以下配置（dataId可以是application-dev.yml等）：

spring:
  shardingsphere:
    # 数据源配置
    datasource:
      # 定义所有数据源名称，多个用逗号分隔
      names: user_db,message_db
      
      # 用户数据库配置
      user_db:
        type: com.zaxxer.hikari.HikariDataSource  # 使用Hikari连接池
        driver-class-name: com.mysql.cj.jdbc.Driver  # MySQL驱动类
        url: jdbc:mysql://localhost:3306/user_db?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai  # 数据库连接地址
        username: root  # 数据库用户名
        password: root  # 数据库密码
        
      # 消息数据库配置
      message_db:
        type: com.zaxxer.hikari.HikariDataSource  # 使用Hikari连接池
        driver-class-name: com.mysql.cj.jdbc.Driver  # MySQL驱动类
        url: jdbc:mysql://localhost:3306/message_db?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai  # 数据库连接地址
        username: root  # 数据库用户名
        password: root  # 数据库密码
    
    # 分片规则配置
    rules:
      sharding:
        # 表级别的分片配置
        tables:
          # 逻辑表名称：sys_message
          sys_message:
            # 实际数据节点：由数据库名+表名表达式组成，这里表示message_db库中的sys_message_0到sys_message_9表
            actual-data-nodes: message_db.sys_message_${0..9}
            # 表分片策略
            table-strategy:
              # 标准分片策略
              standard:
                sharding-column: deviceId  # 分片键：使用deviceId字段进行分片
                sharding-algorithm-name: sys_message_inline  # 引用的分片算法名称
                
        # 分片算法配置
        sharding-algorithms:
          # 与上面引用的算法名称保持一致
          sys_message_inline:
            type: INLINE  # 算法类型：INLINE表示使用行表达式分片算法
            props:
              # 分片算法表达式：
              # 1. 先计算deviceId的哈希值对10取模
              # 2. 处理可能出现的负数情况（当哈希值为负数时，取模结果可能为负）
              # 3. 确保最终结果在0-9之间，对应sys_message_0到sys_message_9表
              algorithm-expression: sys_message_${deviceId.hashCode() % 10 + 10 >= 10 ? deviceId.hashCode() % 10 : deviceId.hashCode() % 10 + 10}

Sharding-JDBC 配置类

package com.zgb.config;

import org.apache.shardingsphere.driver.api.ShardingSphereDataSourceFactory;
import org.apache.shardingsphere.infra.config.algorithm.AlgorithmConfiguration;
import org.apache.shardingsphere.sharding.api.config.ShardingRuleConfiguration;
import org.apache.shardingsphere.sharding.api.config.rule.ShardingTableRuleConfiguration;
import org.apache.shardingsphere.sharding.api.config.strategy.sharding.StandardShardingStrategyConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.alibaba.nacos.api.config.annotation.NacosValue;

import javax.sql.DataSource;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * Sharding-JDBC 配置类，用于配置分库分表规则
 * 从Nacos配置中心获取数据源和分片规则配置
 */
@Configuration
public class ShardingJdbcConfig {

    /**
     * 用户业务库数据源配置，从Nacos获取
     */
    @NacosValue("${spring.shardingsphere.datasource.user.url}")
    private String userDbUrl;
    
    @NacosValue("${spring.shardingsphere.datasource.user.username}")
    private String userDbUsername;
    
    @NacosValue("${spring.shardingsphere.datasource.user.password}")
    private String userDbPassword;
    
    @NacosValue("${spring.shardingsphere.datasource.user.driver-class-name}")
    private String userDbDriverClassName;

    /**
     * 聊天记录库数据源配置，从Nacos获取
     */
    @NacosValue("${spring.shardingsphere.datasource.message.url}")
    private String messageDbUrl;
    
    @NacosValue("${spring.shardingsphere.datasource.message.username}")
    private String messageDbUsername;
    
    @NacosValue("${spring.shardingsphere.datasource.message.password}")
    private String messageDbPassword;
    
    @NacosValue("${spring.shardingsphere.datasource.message.driver-class-name}")
    private String messageDbDriverClassName;

    /**
     * 配置ShardingSphere数据源
     * 
     * @return 配置好的ShardingSphere数据源
     * @throws SQLException 数据库配置异常
     */
    @Bean
    public DataSource dataSource() throws SQLException {
        // 1. 配置数据源
        Map<String, DataSource> dataSources = new HashMap<>(2);
        dataSources.put("user_db", createDataSource(userDbUrl, userDbUsername, userDbPassword, userDbDriverClassName));
        dataSources.put("message_db", createDataSource(messageDbUrl, messageDbUsername, messageDbPassword, messageDbDriverClassName));
        
        // 2. 配置分片规则
        ShardingRuleConfiguration shardingRuleConfig = new ShardingRuleConfiguration();
        
        // 2.1 配置消息表分表规则
        shardingRuleConfig.getTables().add(getSysMessageTableRuleConfiguration());
        
        // 2.2 配置分库策略（按业务类型）
        shardingRuleConfig.setDefaultDatabaseShardingStrategy(new StandardShardingStrategyConfiguration(
                "database_type", "db_inline"));
        
        // 2.3 配置分片算法
        shardingRuleConfig.getShardingAlgorithms().put("table_inline", 
                new AlgorithmConfiguration("INLINE", getTableShardingAlgorithmProperties()));
        shardingRuleConfig.getShardingAlgorithms().put("db_inline", 
                new AlgorithmConfiguration("INLINE", getDbShardingAlgorithmProperties()));
        
        // 3. 创建并返回ShardingSphere数据源
        return ShardingSphereDataSourceFactory.createDataSource(dataSources, shardingRuleConfig, new Properties());
    }

    /**
     * 配置sys_message表的分表规则
     * 
     * @return 消息表的分片规则配置
     */
    private ShardingTableRuleConfiguration getSysMessageTableRuleConfiguration() {
        ShardingTableRuleConfiguration tableRuleConfig = new ShardingTableRuleConfiguration("sys_message", "message_db.sys_message_${0..9}");
        
        // 按deviceId进行分表
        tableRuleConfig.setTableShardingStrategy(new StandardShardingStrategyConfiguration(
                "deviceId", "table_inline"));
        
        return tableRuleConfig;
    }

    /**
     * 创建数据源
     * 
     * @param url 数据库连接地址
     * @param username 数据库用户名
     * @param password 数据库密码
     * @param driverClassName 数据库驱动类名
     * @return 配置好的数据源
     */
    private DataSource createDataSource(String url, String username, String password, String driverClassName) {
        com.zaxxer.hikari.HikariConfig config = new com.zaxxer.hikari.HikariConfig();
        config.setJdbcUrl(url);
        config.setUsername(username);
        config.setPassword(password);
        config.setDriverClassName(driverClassName);
        return new com.zaxxer.hikari.HikariDataSource(config);
    }

    /**
     * 获取表分片算法属性配置
     * 按deviceId哈希后取模10，分为10个表
     * 
     * @return 表分片算法属性
     */
    private Properties getTableShardingAlgorithmProperties() {
        Properties props = new Properties();
        props.setProperty("algorithm-expression", "sys_message_${deviceId.hashCode() % 10 + 10 >= 10 ? deviceId.hashCode() % 10 : deviceId.hashCode() % 10 + 10}");
        return props;
    }

    /**
     * 获取库分片算法属性配置
     * 按业务类型区分用户库和消息库
     * 
     * @return 库分片算法属性
     */
    private Properties getDbShardingAlgorithmProperties() {
        Properties props = new Properties();
        props.setProperty("algorithm-expression", "${database_type == 'message' ? 'message_db' : 'user_db'}");
        return props;
    }
}

SysMessage 实体类

package com.zgb.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import java.time.LocalDateTime;

/**
 * 人与AI对话消息实体类
 * 对应分表后的sys_message_0到sys_message_9表
 */
@Data
@TableName(value = "sys_message", autoResultMap = true)
public class SysMessage {

    /**
     * 消息ID，主键，自增
     */
    @TableId(type = IdType.AUTO)
    private Long messageId;

    /**
     * 设备ID，用于分表的关键字段
     */
    @TableField("deviceId")
    private String deviceId;

    /**
     * 会话ID
     */
    @TableField("sessionId")
    private String sessionId;

    /**
     * 消息发送方：user-用户，assistant-人工智能
     */
    private String sender;

    /**
     * AI扮演的角色ID
     */
    private Long roleId;

    /**
     * 消息内容
     */
    private String message;

    /**
     * 消息类型
     */
    private String messageType;

    /**
     * 语音文件路径
     */
    private String audioPath;

    /**
     * 状态：1-有效，0-删除
     */
    private String state;

    /**
     * 消息发送时间
     */
    private LocalDateTime createTime;

    /**
     * 数据库类型标识，用于分库的关键字段
     * 不在数据库表中，仅用于分库路由
     */
    @TableField(exist = false)
    private String databaseType = "message";
}

mapper类

package com.zgb.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.zgb.entity.SysMessage;
import org.apache.ibatis.annotations.Mapper;
import java.util.List;

/**
 * 消息表数据访问层接口
 * 继承MyBatis Plus的BaseMapper，提供基本CRUD操作
 */
@Mapper
public interface SysMessageMapper extends BaseMapper<SysMessage> {

    /**
     * 根据设备ID查询消息列表
     * 
     * @param deviceId 设备ID
     * @return 该设备的消息列表
     */
    List<SysMessage> selectByDeviceId(String deviceId);

    /**
     * 根据会话ID查询消息列表
     * 
     * @param sessionId 会话ID
     * @return 该会话的消息列表
     */
    List<SysMessage> selectBySessionId(String sessionId);

    /**
     * 批量插入消息
     * 
     * @param messages 消息列表
     * @return 插入成功的记录数
     */
    int batchInsert(List<SysMessage> messages);
}

服务接口类

package com.zgb.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.zgb.entity.SysMessage;
import com.zgb.mapper.SysMessageMapper;
import com.zgb.service.SysMessageService;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.List;

/**
 * 消息服务实现类
 * 实现消息的CRUD及相关业务操作
 */
@Service
@RequiredArgsConstructor
public class SysMessageServiceImpl extends ServiceImpl<SysMessageMapper, SysMessage> implements SysMessageService {

    private final SysMessageMapper sysMessageMapper;

    /**
     * 保存消息
     * 
     * @param message 消息实体，必须包含deviceId用于分表路由
     * @return 保存成功返回true，否则返回false
     */
    @Override
    public boolean saveMessage(SysMessage message) {
        // 设置数据库类型为message，确保路由到消息库
        message.setDatabaseType("message");
        return save(message);
    }

    /**
     * 根据设备ID查询消息
     * 
     * @param deviceId 设备ID，用于分表路由
     * @return 该设备的消息列表
     */
    @Override
    public List<SysMessage> getMessagesByDeviceId(String deviceId) {
        return sysMessageMapper.selectByDeviceId(deviceId);
    }

    /**
     * 根据会话ID查询消息
     * Sharding-JDBC会根据SQL中的deviceId进行路由
     * 
     * @param sessionId 会话ID
     * @return 该会话的消息列表
     */
    @Override
    public List<SysMessage> getMessagesBySessionId(String sessionId) {
        return sysMessageMapper.selectBySessionId(sessionId);
    }

    /**
     * 批量保存消息
     * 
     * @param messages 消息列表，每条消息必须包含deviceId用于分表路由
     * @return 保存成功的数量
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public int batchSaveMessages(List<SysMessage> messages) {
        // 设置数据库类型为message，确保路由到消息库
        messages.forEach(message -> message.setDatabaseType("message"));
        return sysMessageMapper.batchInsert(messages);
    }
}

服务接口

package com.zgb.service;

import com.baomidou.mybatisplus.extension.service.IService;
import com.zgb.entity.SysMessage;
import java.util.List;

/**
 * 消息服务接口
 * 提供消息的CRUD及相关业务操作
 */
public interface SysMessageService extends IService<SysMessage> {

    /**
     * 保存消息
     * 
     * @param message 消息实体
     * @return 保存成功返回true，否则返回false
     */
    boolean saveMessage(SysMessage message);

    /**
     * 根据设备ID查询消息
     * 
     * @param deviceId 设备ID
     * @return 消息列表
     */
    List<SysMessage> getMessagesByDeviceId(String deviceId);

    /**
     * 根据会话ID查询消息
     * 
     * @param sessionId 会话ID
     * @return 消息列表
     */
    List<SysMessage> getMessagesBySessionId(String sessionId);

    /**
     * 批量保存消息
     * 
     * @param messages 消息列表
     * @return 保存成功的数量
     */
    int batchSaveMessages(List<SysMessage> messages);
}

实现说明

  1. 分库策略：

    • 通过databaseType字段区分，聊天记录相关表路由到message_db，用户业务表路由到user_db
    • 在消息实体类中默认设置databaseType = "message"，确保消息表操作都路由到消息库

  2. 分表策略：

    • 聊天记录表sys_message按deviceId的哈希值取模10分为10个表（sys_message_0到sys_message_9）
    • 使用INLINE算法实现分片，确保相同设备的消息保存在同一个表中，便于查询

  3. 核心优势：

    • 分片规则配置在Nacos中，可动态调整，无需重启服务
    • 使用MyBatis Plus简化数据库操作，开发者无需关心具体操作的是哪个分表
    • 分库分表逻辑对业务代码透明，降低开发复杂度

以上实现满足了将聊天记录与用户业务分离存储，并按用户ID水平拆分聊天记录表的需求，适合高并发、大数据量的聊天场景。

# oracle # 数据库
Logo
乐奇 Rokid 开放社区

更多推荐

cover

【记账大师 AIUI】购物记账与交互可靠性实践

avatar 乐奇 Rokid 开放社区

从对话到行动：用 AIUI 构建「意图驱动」的隐形界面——「言值」AI 表达教练开发实录

我观察到一个现象：身边准备面试和答辩的人，练表达的方式几乎只有三种——看面经、背稿子、对着手机录像回看。练习的时候，你可以低头看屏幕。面试时，你要看着对方的眼睛；答辩时，你要看着评委；路演时，你要站定、看着观众；即兴发言时，你连手机都来不及掏。只要一低头，眼神、体态、气场全散了；而双手被手机占住，练习本身也就失真了。更麻烦的是反馈。练完之后，大多数人只知道「我好像说得不太好」，但不知道下一句该怎么

avatar 乐奇 Rokid 开放社区
cover

Rokid_AIUI_成都沙龙_深度观察

avatar 乐奇 Rokid 开放社区