通過租戶 id 實現的 SaaS 方案

概況

項目開發到一半，用戶突然提出需要多個分公司共同使用，這種需要將系統設計成 SaaS 架構，將各個分公司的數據進行隔離。

SaaS 實現的方案

獨立數據庫

每個企業獨立的物理數據庫，隔離性好，成本高。
共享數據庫、獨立 schema

就是一臺物理機，多個邏輯數據庫，oracle 叫做 schema，mysql 叫做 database，每個企業獨立的 schema。
共享數據庫、數據庫表 (本次採用)：

在表中添加 “企業” 或者 “租戶” 字段區分是哪個企業的數據。操作的時候根據 “租戶” 字段去查詢相應的數據。

優點：所有租戶使用同一數據庫，所以成本低廉。

缺點：隔離級別低，安全性低，需要在開發時加大對安全的開發量，數據備份和恢復最困難。

改造思路

本次採用共享數據庫、數據庫表的 SaaS 方案。改造時需要做以下工作：

創建租戶信息表。
先要將所有的表添加租戶 id 字段tenant_id。用於關聯租戶信息表。
將tenant_id和原始表 id 創建聯合主鍵。注意主鍵的順序，原表主鍵必須在左邊。
將表修改爲分區表。

改造後，在添加租戶信息的時候，同時在所有表中添加該租戶的分區，分區用於保存該租戶的數據。
在後續增加記錄時，需要tenant_id字段的值，在刪改查中，都需要在 where 條件中以tenant_id爲條件來操作某個租戶的數據。

測試環境介紹

測試庫中有 5 張表，我下文使用sys_log表進行測試。

sys_log的建表語句爲：

CREATE TABLE `sys_log` (
  `log_id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主鍵',
  `type` TINYINT(1) DEFAULT NULL COMMENT '類型',
  `content` VARCHAR(255) DEFAULT NULL COMMENT '內容',
  `create_id` BIGINT(18) DEFAULT NULL COMMENT '創建人ID',
  `create_time` DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '創建時間',
  `tenant_id` INT NOT NULL,
  PRIMARY KEY (`log_id`,`tenant_id`) USING BTREE
) ENGINE=INNODB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT='系統日誌'

表添加租戶 id 字段

找出未添加租戶 id(tenant_id) 字段的表。

SELECT 
    table_name 
  FROM
    INFORMATION_SCHEMA.TABLES
  WHERE table_schema = 'my'   
    AND table_name NOT IN 
    (SELECT 
      t.table_name 
    FROM
      (SELECT 
        table_name,
        column_name 
      FROM
        information_schema.columns 
      WHERE table_name IN 
        (SELECT 
          table_name 
        FROM
          INFORMATION_SCHEMA.TABLES 
        WHERE table_schema = 'my')) t 
    WHERE t.column_name = 'tenant_id') ;

執行，找到兩個符合條件的表，在數據庫進行確認，確實表中沒tenant_id字段。

創建租戶信息表

僅供參考，用於保存租戶信息

CREATE TABLE `t_tenant` (
  `tenant_id` varchar(40) NOT NULL DEFAULT 'c12dee54f652452b88142a0267ec74b7' COMMENT '租戶id',
  `tenant_code` varchar(100) DEFAULT NULL COMMENT '租戶編碼',
  `name` varchar(50) DEFAULT NULL COMMENT '租戶名稱',
  `desc` varchar(500) DEFAULT NULL COMMENT '租戶描述',
  `logo` varchar(255) DEFAULT NULL COMMENT '公司logo地址',
  `status` smallint(6) DEFAULT NULL COMMENT '狀態1有效0無效',
  `create_by` varchar(100) DEFAULT NULL COMMENT '創建者',
  `create_time` datetime DEFAULT NULL COMMENT '創建時間',
  `last_update_by` varchar(100) DEFAULT NULL COMMENT '最後修改人',
  `last_update_time` datetime DEFAULT NULL COMMENT '最後修改時間',
  `street_address` varchar(200) DEFAULT NULL COMMENT '街道樓號地址',
  `province` varchar(20) DEFAULT NULL COMMENT '一級行政單位,如廣東省,上海市等',
  `city` varchar(20) DEFAULT NULL COMMENT '城市, 如廣州市,佛山市等',
  `district` varchar(20) DEFAULT NULL COMMENT '行政區,如番禺區,天河區等',
  `link_man` varchar(50) DEFAULT NULL COMMENT '聯繫人',
  `link_phone` varchar(50) DEFAULT NULL COMMENT '聯繫電話',
  `longitude` decimal(10,6) DEFAULT NULL COMMENT '經度',
  `latitude` decimal(10,6) DEFAULT NULL COMMENT '緯度',
  `adcode` varchar(8) DEFAULT NULL COMMENT '區域編碼,用於通過區域id快速匹配後展示, 如廣州是440100',
  PRIMARY KEY (`tenant_id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='租戶的基本信息表';

將所有表添加 tenant_id 字段

DROP PROCEDURE IF EXISTS addColumn ;
DELIMITER ?
CREATE PROCEDURE addColumn () 
BEGIN
  -- 定義表名變量
  DECLARE s_tablename VARCHAR (100) ;
  /*顯示錶的數據庫中的所有表
 SELECT table_name FROM information_schema.tables WHERE table_schema='databasename' Order by table_name ;
 */
  #顯示所有
  DECLARE cur_table_structure CURSOR FOR 
  SELECT 
    table_name 
  FROM
    INFORMATION_SCHEMA.TABLES
  WHERE table_schema = 'my'     -- my = 我的測試數據庫名稱
    AND table_name NOT IN 
    (SELECT 
      t.table_name 
    FROM
      (SELECT 
        table_name,
        column_name 
      FROM
        information_schema.columns 
      WHERE table_name IN 
        (SELECT 
          table_name 
        FROM
          INFORMATION_SCHEMA.TABLES 
        WHERE table_schema = 'my')) t 
    WHERE t.column_name = 'tenant_id') ;
  DECLARE CONTINUE HANDLER FOR SQLSTATE '02000' SET s_tablename = NULL ;
  OPEN cur_table_structure ;
  FETCH cur_table_structure INTO s_tablename ;
  WHILE
    (s_tablename IS NOT NULL) DO SET @MyQuery = CONCAT(
      "alter table `",
      s_tablename,
      "` add COLUMN `tenant_id` INT not null COMMENT '租戶id'"
    ) ;
    PREPARE msql FROM @MyQuery ;
    EXECUTE msql ;
    #USING @c; 
    FETCH cur_table_structure INTO s_tablename ;
  END WHILE ;
  CLOSE cur_table_structure ;
END ?
DELIMITER ;
#執行存儲過程
CALL addColumn () ;

實現表分區

實現的目標：在添加租戶的時候實現對所有表添加分區

需要的條件：

表必須是分區表，如果不是分區表，那麼需要改成分區表。
tenant_id必須和原表log_id主鍵組成聯合主鍵。

將表修改成分區表

表中添加分區有三種方式：

創建臨時分區表sys_log_copy，copy 數據過來後，刪除舊的sys_log，再將sys_log_copy修改爲sys_log（本次採用，詳見下文）
直接將表修改爲分區表，需要原表中無數據，否則無法成功：

ALTER TABLE sys_log PARTITION BY LIST COLUMNS (tenant_id)
(
    PARTITION a1 VALUES IN (1) ENGINE = INNODB,
    PARTITION a2 VALUES IN (2) ENGINE = INNODB,
    PARTITION a3 VALUES IN (3) ENGINE = INNODB
);

在分區表中添加新分區，需要表已經是分區表，否則無法成功：

ALTER TABLE sys_log_copy ADD PARTITION
(
    PARTITION a4 VALUES IN (4) ENGINE = INNODB,
    PARTITION a5 VALUES IN (5) ENGINE = INNODB,
    PARTITION a6 VALUES IN (6) ENGINE = INNODB
);

通過`創建臨時分區表`的方式將原錶轉換成分區表

查看錶建表語句：

SHOW CREATE TABLE `sys_log`;

2、參考建表語句，創建 copy 表：

CREATE TABLE `sys_log_copy` (
  `log_id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主鍵',
  `type` TINYINT(1) DEFAULT NULL COMMENT '類型',
  `content` VARCHAR(255) DEFAULT NULL COMMENT '內容',
  `create_id` BIGINT(18) DEFAULT NULL COMMENT '創建人ID',
  `create_time` DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '創建時間',
  `tenant_id` INT NOT NULL,
  PRIMARY KEY (`log_id`,`tenant_id`) USING BTREE
) ENGINE=INNODB DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC COMMENT='系統日誌'
PARTITION BY LIST COLUMNS (tenant_id)
(
    PARTITION a1 VALUES IN (1) ENGINE = INNODB,
    PARTITION a2 VALUES IN (2) ENGINE = INNODB,
    PARTITION a3 VALUES IN (3) ENGINE = INNODB
);

注意上文中的DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC

CHARSET=utf8mb4是因爲 utf8 在 mysql 中是不健全的編碼。

ROW_FORMAT=DYNAMIC是爲了避免所以長度過大後導致如下報錯：
ERROR 1709 (HY000): Index column size too large. The maximum column size is 767 bytes.
也可以在 my.ini 配置文件中設置爲 true 解決這個問題，但是要重啓數據庫，會比較麻煩。
[mysqld]
innodb_large_prefix=true

驗證分區情況：

SELECT 
  partition_name part,
  partition_expression expr,
  partition_description descr,
  table_rows 
FROM
  information_schema.partitions 
WHERE TABLE_SCHEMA = SCHEMA() 
  AND TABLE_NAME = 'sys_log_copy' ;

可以查看到添加的3個分區

將數據複製到 copy 表中

INSERT INTO `sys_log_copy` SELECT * FROM `sys_log`

5、刪除表sys_log，再修改sys_log_copy表中的名字爲sys_log。

編寫自動創建分區的倉儲過程

通過存儲過程實現，在分區表中添加分區

DELIMITER ?
USE `my`?
DROP PROCEDURE IF EXISTS `add_table_partition`?
CREATE DEFINER=`root`@`%` PROCEDURE `add_table_partition`(IN _tenantId INT)
BEGIN
  DECLARE IS_FOUND INT DEFAULT 1 ;
  -- 用於記錄遊標中存在分區的表名
  DECLARE v_tablename VARCHAR (200) ;
  -- 用於緩存添加分區時候的sql
  DECLARE v_sql VARCHAR (5000) ;
  -- 分區名稱定義
  DECLARE V_P_VALUE VARCHAR (100) DEFAULT CONCAT('P', REPLACE(_tenantId, '-', '')) ;
  DECLARE V_COUNT INT ;
  DECLARE V_LOONUM INT DEFAULT 0 ;
  DECLARE V_NUM INT DEFAULT 0 ;
  -- 定義遊標，值是所有分區表的表名
  DECLARE curr CURSOR FOR 
  (SELECT 
    t.TABLE_NAME 
  FROM
    INFORMATION_SCHEMA.partitions t 
  WHERE TABLE_SCHEMA = SCHEMA() 
    AND t.partition_name IS NOT NULL 
  GROUP BY t.TABLE_NAME) ;
  -- 如果沒影響的記錄，程序也繼續執行
  DECLARE CONTINUE HANDLER FOR NOT FOUND SET IS_FOUND=0;
  -- 獲取上一步中的遊標中獲取到的表名的個數
  SELECT 
    COUNT(1) INTO V_LOONUM 
  FROM
    (SELECT 
      t.TABLE_NAME 
    FROM
      INFORMATION_SCHEMA.partitions t 
    WHERE TABLE_SCHEMA = SCHEMA() 
      AND t.partition_name IS NOT NULL 
    GROUP BY t.TABLE_NAME) A ;
  -- 只有在存在分區表的時候纔打開遊標
  IF V_LOONUM > 0 
  THEN -- 打開遊標
  OPEN curr ;
  -- 循環
  read_loop :
  LOOP
    -- 聲明結束的時候
    IF V_NUM >= V_LOONUM 
    THEN LEAVE read_loop ;
    END IF ;
    -- 取遊標的值給變量
    FETCH curr INTO v_tablename ;
    -- 依次判斷分區表是否存在改分區，如果不存在則添加分區
    SET V_NUM = V_NUM + 1 ;
    SELECT 
      COUNT(1) INTO V_COUNT 
    FROM
      INFORMATION_SCHEMA.partitions t 
    WHERE LOWER(T.TABLE_NAME) = LOWER(v_tablename) 
      AND T.PARTITION_NAME = V_P_VALUE 
      AND T.TABLE_SCHEMA = SCHEMA() ;
    IF V_COUNT <= 0 
    THEN SET v_sql = CONCAT(
      '  ALTER TABLE ',
      v_tablename,
      ' ADD PARTITION (PARTITION ',
      V_P_VALUE,
      ' VALUES IN(',
      _tenantId,
      ') ENGINE = INNODB) '
    ) ;
    SET @v_sql = v_sql ;
    -- 預處理需要執行的動態SQL，其中stmt是一個變量
    PREPARE stmt FROM @v_sql ;
    -- 執行SQL語句
    EXECUTE stmt ;
    -- 釋放掉預處理段
    DEALLOCATE PREPARE stmt ;
    END IF ;
    -- 結束循環
  END LOOP read_loop;
  -- 關閉遊標
  CLOSE curr ;
  END IF ;
END?
DELIMITER ;

調用存儲過程測試

CALL add_table_partition (8) ;

如果表還不是分區表，那麼調用存儲過程會有如下報錯：
可能會報錯如下：

DECLARE CONTINUE HANDLER FOR NOT FOUND SET IS_FOUND=0;
SELECT 
  partition_name part,
  partition_expression expr,
  partition_description descr,
  table_rows 
FROM
  information_schema.partitions 
WHERE TABLE_SCHEMA = SCHEMA() 
  AND TABLE_NAME = 'sys_log' ;

通過`mybatis`調用存儲過程

<select statementType="CALLABLE" useCache="false" parameterType="string">
        <![CDATA[
                call add_table_partition (
                        #{_tenantId,mode=IN,jdbcType=VARCHAR});
        ]]>
</select>

實現簡單的數據權限

我們可能需要這種場景需求

集團公司有多個子公司，集團公司和每個子公司各自是一個租戶，但是子公司下還有子公司。
不管是集團公司還是說旗下的子公司都有相應的用戶 (t_user)。
用戶需要有查看自己公司下的數據和下面子公司的數據的權限。

從上面的場景需求中，我們知道，t_tenant表需要設計成樹樁結構。下面我們進行測試。

修改上文的`t_tenant`表爲：

CREATE TABLE `t_tenant` (
  `tenant_id` VARCHAR(40) NOT NULL DEFAULT '0' COMMENT '租戶id',
  `path` VARCHAR(200) DEFAULT NOT NULL COMMENT '從根節點開始的id路徑樹，如：0-2-21-211-2111，通過"-"隔開，最末尾爲自己id',
  `tenant_code` VARCHAR(100) DEFAULT NULL COMMENT '租戶編碼',
  `name` VARCHAR(50) DEFAULT NULL COMMENT '租戶名稱',
  `logo` VARCHAR(255) DEFAULT NULL COMMENT '公司logo地址',
  `status` SMALLINT(6) DEFAULT NULL COMMENT '狀態1有效0無效',
  `create_by` VARCHAR(100) DEFAULT NULL COMMENT '創建者',
  `create_time` DATETIME DEFAULT NULL COMMENT '創建時間',
  `last_update_by` VARCHAR(100) DEFAULT NULL COMMENT '最後修改人',
  `last_update_time` DATETIME DEFAULT NULL COMMENT '最後修改時間',
  `street_address` VARCHAR(200) DEFAULT NULL COMMENT '街道樓號地址',
  PRIMARY KEY (`tenant_id`) USING BTREE
) ENGINE=INNODB DEFAULT CHARSET=utf8 COMMENT='租戶的基本信息表'

修改的地方有：

爲了演示，刪除了些感覺沒是沒用的字段
添加了path字段，實現租戶和子租戶的樹形結構

添加測試數據

新增租戶信息：

通過 path 緩存着t_tenant樹的路徑。

創建用戶表 (t_user)，添加測試用戶：

測試的用戶 id 和 tenant_id 需要對應

創建附件表 (t_file)，添加測試業務數據：

創建人字段 (create_by）關聯用戶表 (t_user)，也要關聯到租戶 (tenant_id)，指明是哪個子公司的數據。

進行測試

查看tenant_id是 "211" 的租戶信息和其下的子租戶信息

SELECT 
  tt.`tenant_id`,
  tt.path 
FROM
  t_tenant tt 
WHERE 
  (SELECT 
    INSTR(tt.path, "211")) ;

查看tenant_id是 "211" 的附件和其下的子租戶的附件信息

SELECT 
  * 
FROM
  t_file tf 
WHERE tf.`tenant_id` IN 
  (SELECT 
    tt.`tenant_id` 
  FROM
    t_tenant tt 
  WHERE 
    (SELECT 
      INSTR(tt.path, "211"))) ;

查看tenant_id是 "2" 的附件和其下的子租戶的附件信息

通過 mybatis 攔截器實現查看子租戶的數據權限

編寫攔截器：

package com.iee.orm.mybatis.common;
import com.baomidou.mybatisplus.core.toolkit.PluginUtils;
import com.iee.orm.mybatis.common.UserHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.executor.statement.StatementHandler;
import org.apache.ibatis.mapping.MappedStatement;
import org.apache.ibatis.mapping.SqlCommandType;
import org.apache.ibatis.mapping.StatementType;
import org.apache.ibatis.plugin.*;
import org.apache.ibatis.reflection.MetaObject;
import org.apache.ibatis.reflection.SystemMetaObject;
import org.springframework.context.annotation.Configuration;
import java.sql.Connection;
import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
 * 實現 攔截select語句，實現尾部拼接sql來查詢本租戶和子租戶信息
 * @author longxiaonan@aliyun.com
 */
@Slf4j
@Configuration
@Intercepts({
        @Signature(type = StatementHandler.class, method = "prepare", args = {Connection.class, Integer.class})
})
public class SqlInterceptor implements Interceptor {
    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        StatementHandler statementHandler = PluginUtils.realTarget(invocation.getTarget());
        MetaObject metaObject = SystemMetaObject.forObject(statementHandler);
        MappedStatement mappedStatement = (MappedStatement) metaObject.getValue("delegate.mappedstatement");
        if (SqlCommandType.SELECT != mappedStatement.getSqlCommandType()
                || StatementType.CALLABLE == mappedStatement.getStatementType()) {
            return invocation.proceed();
        }
        getSqlByInvocation(metaObject, invocation);
        return invocation.proceed();
    }
     * 拼接sql執行
     * @param metaObject
     * @param invocation
     * @return
     */
    private String getSqlByInvocation(MetaObject metaObject, Invocation invocation) throws NoSuchFieldException, IllegalAccessException {
        String originalSql = (String) metaObject.getValue(PluginUtils.DELEGATE_BOUNDSQL_SQL);
        metaObject.setValue(PluginUtils.DELEGATE_BOUNDSQL_SQL, originalSql);
        String targetSql = addDataSql(originalSql);
        return targetSql;
    }
     * 將原始sql進行拼接
     * @param sql
     * @return
     */
    static String addDataSql(String sql) {
        sql = StringUtils.replace(sql, ";", "");
        StringBuilder sb = new StringBuilder(sql);
        String tenantId = UserHelper.getTenantId();
        String suffSql = " `tenant_id` IN " +
                "(SELECT " +
                "tt.`tenant_id` " +
                "FROM " +
                "t_tenant tt " +
                "WHERE " +
                "(SELECT " +
                "INSTR(tt.path," + tenantId + "))) ";
        String regex = "(.*)(where)(.*)";
        Pattern compile = Pattern.compile(regex);
        Matcher matcher = compile.matcher(sql);
        if (matcher.find()) {
            String whereLastSql = matcher.group(matcher.groupCount());
            int left = StringUtils.countMatches(whereLastSql, "(");
            int right = StringUtils.countMatches(whereLastSql, ")");
            if(left == right){
                sb.append(" and ");
                sb.append(suffSql);
                log.info("數據權限替換後sql:--->" + sb.toString());
                return sb.toString();
            }
        }
        sb.append(" where ");
        sb.append(suffSql);
        log.info("數據權限替換後sql:--->" + sb.toString());
        return sb.toString();
    }
    @Override
    public Object plugin(Object target) {
        return Plugin.wrap(target, new SqlInterceptor());
    }
    @Override
    public void setProperties(Properties properties) {
    }
}

用到了 mybatis-plus 的工具類

/*
 * Copyright (c) 2011-2020, baomidou (jobob@qq.com).
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 * <p>
 * https://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.baomidou.mybatisplus.core.toolkit;
import org.apache.ibatis.reflection.MetaObject;
import org.apache.ibatis.reflection.SystemMetaObject;
import java.lang.reflect.Proxy;
import java.util.Properties;
/**
 * 插件工具類
 *
 * @author TaoYu , hubin
 * @since 2017-06-20
 */
public final class PluginUtils {
    public static final String DELEGATE_BOUNDSQL_SQL = "delegate.boundSql.sql";
    private PluginUtils() {
        // to do nothing
    }
    /**
     * 獲得真正的處理對象,可能多層代理.
     */
    @SuppressWarnings("unchecked")
    public static <T> T realTarget(Object target) {
        if (Proxy.isProxyClass(target.getClass())) {
            MetaObject metaObject = SystemMetaObject.forObject(target);
            return realTarget(metaObject.getValue("h.target"));
        }
        return (T) target;
    }
    /**
     * 根據 key 獲取 Properties 的值
     */
    public static String getProperty(Properties properties, String key) {
        String value = properties.getProperty(key);
        return StringUtils.isEmpty(value) ? null : value;
    }
}

測試的時候發現只要是select語句就會去關聯查詢出子租戶的信息。

測試代碼見：

https://github.com/longxiaonan/java-sea/tree/master/javasea-orm/javasea-orm-mybatis-Interceptor

作者：逍嵐子

來源：juejin.cn/post/6844903993085263886

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/KdWLeU_yFq1Cj2_PkUSbgA