文章目录Spring Boot 3 整合 Sharding-JDBC 分库分表实现Nacos配置中心配置Sharding-JDBC 配置类SysMessage 实体类mapper类服务接口类服务接口实现说明基于spring boot3 lombok 1.18.30 mybatis plus 3.5.5 nacosSpring Boot 3 整合 Sharding-JDBC 分库分表实现根据需求我们需要实现两个核心功能分库将聊天记录库和用户业务库分离 提高主业务吞吐量分表聊天记录表按用户IDdeviceId分为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?useUnicodetruecharacterEncodingutf8useSSLfalseserverTimezoneAsia/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?useUnicodetruecharacterEncodingutf8useSSLfalseserverTimezoneAsia/Shanghai# 数据库连接地址username:root# 数据库用户名password:root# 数据库密码# 分片规则配置rules:sharding:# 表级别的分片配置tables:# 逻辑表名称sys_messagesys_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 配置类packagecom.zgb.config;importorg.apache.shardingsphere.driver.api.ShardingSphereDataSourceFactory;importorg.apache.shardingsphere.infra.config.algorithm.AlgorithmConfiguration;importorg.apache.shardingsphere.sharding.api.config.ShardingRuleConfiguration;importorg.apache.shardingsphere.sharding.api.config.rule.ShardingTableRuleConfiguration;importorg.apache.shardingsphere.sharding.api.config.strategy.sharding.StandardShardingStrategyConfiguration;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importcom.alibaba.nacos.api.config.annotation.NacosValue;importjavax.sql.DataSource;importjava.sql.SQLException;importjava.util.HashMap;importjava.util.Map;importjava.util.Properties;/** * Sharding-JDBC 配置类用于配置分库分表规则 * 从Nacos配置中心获取数据源和分片规则配置 */ConfigurationpublicclassShardingJdbcConfig{/** * 用户业务库数据源配置从Nacos获取 */NacosValue(${spring.shardingsphere.datasource.user.url})privateStringuserDbUrl;NacosValue(${spring.shardingsphere.datasource.user.username})privateStringuserDbUsername;NacosValue(${spring.shardingsphere.datasource.user.password})privateStringuserDbPassword;NacosValue(${spring.shardingsphere.datasource.user.driver-class-name})privateStringuserDbDriverClassName;/** * 聊天记录库数据源配置从Nacos获取 */NacosValue(${spring.shardingsphere.datasource.message.url})privateStringmessageDbUrl;NacosValue(${spring.shardingsphere.datasource.message.username})privateStringmessageDbUsername;NacosValue(${spring.shardingsphere.datasource.message.password})privateStringmessageDbPassword;NacosValue(${spring.shardingsphere.datasource.message.driver-class-name})privateStringmessageDbDriverClassName;/** * 配置ShardingSphere数据源 * * return 配置好的ShardingSphere数据源 * throws SQLException 数据库配置异常 */BeanpublicDataSourcedataSource()throwsSQLException{// 1. 配置数据源MapString,DataSourcedataSourcesnewHashMap(2);dataSources.put(user_db,createDataSource(userDbUrl,userDbUsername,userDbPassword,userDbDriverClassName));dataSources.put(message_db,createDataSource(messageDbUrl,messageDbUsername,messageDbPassword,messageDbDriverClassName));// 2. 配置分片规则ShardingRuleConfigurationshardingRuleConfignewShardingRuleConfiguration();// 2.1 配置消息表分表规则shardingRuleConfig.getTables().add(getSysMessageTableRuleConfiguration());// 2.2 配置分库策略按业务类型shardingRuleConfig.setDefaultDatabaseShardingStrategy(newStandardShardingStrategyConfiguration(database_type,db_inline));// 2.3 配置分片算法shardingRuleConfig.getShardingAlgorithms().put(table_inline,newAlgorithmConfiguration(INLINE,getTableShardingAlgorithmProperties()));shardingRuleConfig.getShardingAlgorithms().put(db_inline,newAlgorithmConfiguration(INLINE,getDbShardingAlgorithmProperties()));// 3. 创建并返回ShardingSphere数据源returnShardingSphereDataSourceFactory.createDataSource(dataSources,shardingRuleConfig,newProperties());}/** * 配置sys_message表的分表规则 * * return 消息表的分片规则配置 */privateShardingTableRuleConfigurationgetSysMessageTableRuleConfiguration(){ShardingTableRuleConfigurationtableRuleConfignewShardingTableRuleConfiguration(sys_message,message_db.sys_message_${0..9});// 按deviceId进行分表tableRuleConfig.setTableShardingStrategy(newStandardShardingStrategyConfiguration(deviceId,table_inline));returntableRuleConfig;}/** * 创建数据源 * * param url 数据库连接地址 * param username 数据库用户名 * param password 数据库密码 * param driverClassName 数据库驱动类名 * return 配置好的数据源 */privateDataSourcecreateDataSource(Stringurl,Stringusername,Stringpassword,StringdriverClassName){com.zaxxer.hikari.HikariConfigconfignewcom.zaxxer.hikari.HikariConfig();config.setJdbcUrl(url);config.setUsername(username);config.setPassword(password);config.setDriverClassName(driverClassName);returnnewcom.zaxxer.hikari.HikariDataSource(config);}/** * 获取表分片算法属性配置 * 按deviceId哈希后取模10分为10个表 * * return 表分片算法属性 */privatePropertiesgetTableShardingAlgorithmProperties(){PropertiespropsnewProperties();props.setProperty(algorithm-expression,sys_message_${deviceId.hashCode() % 10 10 10 ? deviceId.hashCode() % 10 : deviceId.hashCode() % 10 10});returnprops;}/** * 获取库分片算法属性配置 * 按业务类型区分用户库和消息库 * * return 库分片算法属性 */privatePropertiesgetDbShardingAlgorithmProperties(){PropertiespropsnewProperties();props.setProperty(algorithm-expression,${database_type message ? message_db : user_db});returnprops;}}SysMessage 实体类package com.zgb.entity;importcom.baomidou.mybatisplus.annotation.IdType;importcom.baomidou.mybatisplus.annotation.TableId;importcom.baomidou.mybatisplus.annotation.TableField;importcom.baomidou.mybatisplus.annotation.TableName;importlombok.Data;importjava.time.LocalDateTime;/** * 人与AI对话消息实体类 * 对应分表后的sys_message_0到sys_message_9表 */Data TableName(valuesys_message,autoResultMaptrue)publicclassSysMessage{/** * 消息ID主键自增 */TableId(typeIdType.AUTO)privateLong messageId;/** * 设备ID用于分表的关键字段 */TableField(deviceId)privateString deviceId;/** * 会话ID */TableField(sessionId)privateString sessionId;/** * 消息发送方user-用户assistant-人工智能 */privateString sender;/** * AI扮演的角色ID */privateLong roleId;/** * 消息内容 */privateString message;/** * 消息类型 */privateString messageType;/** * 语音文件路径 */privateString audioPath;/** * 状态1-有效0-删除 */privateString state;/** * 消息发送时间 */privateLocalDateTime createTime;/** * 数据库类型标识用于分库的关键字段 * 不在数据库表中仅用于分库路由 */TableField(existfalse)privateString databaseTypemessage;}mapper类package com.zgb.mapper;importcom.baomidou.mybatisplus.core.mapper.BaseMapper;importcom.zgb.entity.SysMessage;importorg.apache.ibatis.annotations.Mapper;importjava.util.List;/** * 消息表数据访问层接口 * 继承MyBatis Plus的BaseMapper提供基本CRUD操作 */Mapperpublicinterface SysMessageMapper extends BaseMapperSysMessage{/** * 根据设备ID查询消息列表 * * param deviceId 设备ID * return 该设备的消息列表 */ListSysMessageselectByDeviceId(String deviceId);/** * 根据会话ID查询消息列表 * * param sessionId 会话ID * return 该会话的消息列表 */ListSysMessageselectBySessionId(String sessionId);/** * 批量插入消息 * * param messages 消息列表 * return 插入成功的记录数 */intbatchInsert(ListSysMessagemessages);}服务接口类package com.zgb.service.impl;importcom.baomidou.mybatisplus.extension.service.impl.ServiceImpl;importcom.zgb.entity.SysMessage;importcom.zgb.mapper.SysMessageMapper;importcom.zgb.service.SysMessageService;importlombok.RequiredArgsConstructor;importorg.springframework.stereotype.Service;importorg.springframework.transaction.annotation.Transactional;importjava.util.List;/** * 消息服务实现类 * 实现消息的CRUD及相关业务操作 */Service RequiredArgsConstructorpublicclassSysMessageServiceImplextends ServiceImplSysMessageMapper,SysMessageimplements SysMessageService{privatefinalSysMessageMapper sysMessageMapper;/** * 保存消息 * * param message 消息实体必须包含deviceId用于分表路由 * return 保存成功返回true否则返回false */OverridepublicbooleansaveMessage(SysMessage message){// 设置数据库类型为message确保路由到消息库message.setDatabaseType(message);returnsave(message);}/** * 根据设备ID查询消息 * * param deviceId 设备ID用于分表路由 * return 该设备的消息列表 */OverridepublicListSysMessagegetMessagesByDeviceId(String deviceId){returnsysMessageMapper.selectByDeviceId(deviceId);}/** * 根据会话ID查询消息 * Sharding-JDBC会根据SQL中的deviceId进行路由 * * param sessionId 会话ID * return 该会话的消息列表 */OverridepublicListSysMessagegetMessagesBySessionId(String sessionId){returnsysMessageMapper.selectBySessionId(sessionId);}/** * 批量保存消息 * * param messages 消息列表每条消息必须包含deviceId用于分表路由 * return 保存成功的数量 */Override Transactional(rollbackForException.class)publicintbatchSaveMessages(ListSysMessagemessages){// 设置数据库类型为message确保路由到消息库messages.forEach(message-message.setDatabaseType(message));returnsysMessageMapper.batchInsert(messages);}}服务接口packagecom.zgb.service;importcom.baomidou.mybatisplus.extension.service.IService;importcom.zgb.entity.SysMessage;importjava.util.List;/** * 消息服务接口 * 提供消息的CRUD及相关业务操作 */publicinterfaceSysMessageServiceextendsIServiceSysMessage{/** * 保存消息 * * param message 消息实体 * return 保存成功返回true否则返回false */booleansaveMessage(SysMessagemessage);/** * 根据设备ID查询消息 * * param deviceId 设备ID * return 消息列表 */ListSysMessagegetMessagesByDeviceId(StringdeviceId);/** * 根据会话ID查询消息 * * param sessionId 会话ID * return 消息列表 */ListSysMessagegetMessagesBySessionId(StringsessionId);/** * 批量保存消息 * * param messages 消息列表 * return 保存成功的数量 */intbatchSaveMessages(ListSysMessagemessages);}实现说明分库策略通过databaseType字段区分聊天记录相关表路由到message_db用户业务表路由到user_db在消息实体类中默认设置databaseType message确保消息表操作都路由到消息库分表策略聊天记录表sys_message按deviceId的哈希值取模10分为10个表sys_message_0到sys_message_9使用INLINE算法实现分片确保相同设备的消息保存在同一个表中便于查询核心优势分片规则配置在Nacos中可动态调整无需重启服务使用MyBatis Plus简化数据库操作开发者无需关心具体操作的是哪个分表分库分表逻辑对业务代码透明降低开发复杂度以上实现满足了将聊天记录与用户业务分离存储并按用户ID水平拆分聊天记录表的需求适合高并发、大数据量的聊天场景。