经验 mysql 按数据量分表代码

一、Laravel 框架使用的trait文件，只需要在Model类中引入即可使用其它方法
use SplitTable;

<?php

namespace App;
use Illuminate\Support\Facades\Redis;
trait SplitTable
{
    /*
     * 获取指定表名的全部分表
     *
     */
    public function get_subtable($table_name){
        $cache_key = 'sptblist'.$table_name;
        //先读缓存。缓存中无数据再读数据库
        if ($tables = Redis::get($cache_key)) {
            return json_decode($tables, true);
        }
       
        $i=1;
        $finish = false;
        $tables = [];
        while (!$finish) {
            $sql = 'select TABLE_NAME from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA="'.env('DB_DATABASE').'" and TABLE_NAME="'.$table_name.'_n'.$i.'"';
            $res = \DB::select($sql);
            if ($res) {
                $tables[] = $table_name.'_n'.$i;
            }
            else{
                $finish = true;
            }
            $i++;
         }
         //读取到数据后存缓存
         if ($tables) {
             Redis::set($cache_key, json_encode($tables));
             Redis::expire($cache_key, 60); //缓存时间1分钟
         }
        
         return $tables;
    }

    /*
     * 创建news数据表的分表
     * 因表字段不一样，每个表一个创建方法，方法名以 create_ 开头加上表名
     */
    public function create_news($num){
        $sql = 'CREATE TABLE IF NOT EXISTS `news_n'.$num."` (
        `id`  bigint(20) NOT NULL COMMENT 'id' ,
        `title`  varchar(50) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '标题' ,
        PRIMARY KEY (`id`)
        )
        ENGINE=MyISAM
        DEFAULT CHARACTER SET=utf8 COLLATE=utf8_general_ci";
        if(\DB::statement($sql) === false)
            return false;
        //创建一个新表后清理缓存
        Redis::delete('sptblistvr_news');
        return true;
    }

    /*
     * 获得一个能够增加数据的分表
     * 如果当前使用的表已满，则自动创建一个新表
     */
    public function get_subtable_for_insert($table_name){
        $max_count = 10000000;   //每个表存储的数据量，超过这个数据量则分表
        $tables = $this->get_subtable($table_name);
        if (!$tables) {
            return false;
        }
        $table = array_pop($tables);
        $sql = 'SELECT COUNT(1) AS c FROM '.$table;
        $res = \DB::select($sql);
        if ($res[0]->c>=$max_count) {
            //表名编号加1
            $num = str_replace($table_name.'_n', '', $table);
            $num++;
            $method = 'create_'.$table_name;
            $this->$method($num);
        }
        else{
            return $table;
        }
    }

    /*
     * 根据主键获取数据
     *
     */
    public function get_by_primary_key($id){
        $cache_key = 'sptb'.$this->getTable().'_'.$id;
        //先读缓存。缓存中无数据再读数据库
        if ($data = Redis::get($cache_key)) {
            return json_decode($data, true);
        }

        $tables = $this->get_subtable($this->getTable());
        if (!$tables) {
            return false;
        }
        //反序，从最近的表中开始读
        $tables = array_reverse($tables);
        foreach ($tables as $table_name) {
            $sql = 'select * from '.$table_name.' where '.$this->getKeyName().'='.$id.' limit 1';
            if ($data = \DB::select($sql)) {
                Redis::set($cache_key, json_encode($data[0]));
                Redis::expire($cache_key, 15*24*60*60); //缓存时间15天，如果有数据变动记得清除缓存
                break;
            }
        }
        return $data?$data[0]:null;
    }
}


二、ThinkPHP 扩展Model基类实现

使用方法：
$Model = new \Home\Model\Model();
$table_list = $Model->get_subtable('news');
$data_list = $Model->get_subtable_page_list('news', 0, 2);
var_dump($Model->getModelName(), $table_list, $data_list );die;


<?php
namespace Home\Model;
class Model extends \Think\Model {
    /*
     * 获取指定表名的全部分表
     *
     */
    public function get_subtable($table_name){
        $cache_key = 'sptblist'.$table_name;
        //先读缓存。缓存中无数据再读数据库
        if ($tables = \Org\FPC::redisHyunA()->get($cache_key)) {
            return json_decode($tables, true);
        }
        $i=1;
        $finish = false;
        $tables = [];
        while (!$finish) {
            $sql = 'select TABLE_NAME from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA="qcloud" and TABLE_NAME="'.$table_name.'_n'.$i.'"';
            $res = $this->db->query($sql);
            if ($res) {
                $tables[] = $table_name.'_n'.$i;
            }
            else{
                $finish = true;
            }
            $i++;
         }
         //读取到数据后存缓存
         if ($tables) {
             \Org\FPC::redisHyunA()->set($cache_key, json_encode($tables), 60); //缓存时间1分钟
         }
         return $tables;
    }

    /*
     * 创建数据表的分表
     * 因表字段不一样，每个表一个创建方法
     */
    public function create_news($num){
        $sql = 'CREATE TABLE IF NOT EXISTS `news_n'.$num."` (
            `id`  int(10) UNSIGNED NOT NULL AUTO_INCREMENT ,
            `title`  varchar(50) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '分享标题' ,
            PRIMARY KEY (`id`)
            )
            ENGINE=MyISAM
            DEFAULT CHARACTER SET=utf8 COLLATE=utf8_general_ci
            AUTO_INCREMENT=1";
        if($this->db->execute($sql) === false)
            return false;
        //创建一个新表后清理缓存
        \Org\FPC::redisHyunA()->del('sptblistnews');
        return true;
    }

    /*
     * 获得一个能够增加数据的分表
     * 如果当前使用的表已满，则自动创建一个新表
     */
    public function get_subtable_for_insert($table_name){
        $max_count = 10000000;   //每个表存储的数据量，超过这个数据量则分表
        $tables = $this->get_subtable($table_name);
        if (!$tables) {
            return false;
        }
        $table = array_pop($tables);
        $sql = 'SELECT COUNT(1) AS c FROM '.$table;
        $res = $this->db->query($sql);
        if ($res['c']>=$max_count) {
            //表名编号加1
            $num = str_replace($table_name.'_n', '', $table);
            $num++;
            $method = 'create_'.$table_name;
            $this->$method($num);
        }
        else{
            return $table;
        }
    }

    /*
     * 根据主键获取数据
     *
     */
    public function get_by_condition($table_name, $condition){
        $condition = ksort($condition);
        $condition_sql = $this->condition_to_sql($condition);
        $cache_key = 'sptb'.$table_name.'_'.$condition_sql;
        //先读缓存。缓存中无数据再读数据库
        if ($data = \Org\FPC::redisHyunA()->get($cache_key)) {
            return json_decode($data, true);
        }

        $tables = $this->get_subtable($table_name);
        if (!$tables) {
            return false;
        }
        //反序，从最近的表中开始读
        $tables = array_reverse($tables);
        foreach ($tables as $tb_name) {
            $sql = 'select * from '.$tb_name.$condition_sql.' limit 1';
            if ($data = $this->db->query($sql)) {
                \Org\FPC::redisHyunA()->set($cache_key, json_encode($data[0]), 60); //缓存时间1分钟，如果有数据变动记得清除缓存
                break;
            }
        }
        return $data?$data[0]:null;
    }

    /*
     * 根据主键获取数据
     *
     */
    public function get_by_primary_key($table_name, $id, $key_name='id'){
        $cache_key = 'sptb'.$table_name.'_'.$id;
        //先读缓存。缓存中无数据再读数据库
        if ($data = \Org\FPC::redisHyunA()->get($cache_key)) {
            return json_decode($data, true);
        }

        $tables = $this->get_subtable($table_name);
        if (!$tables) {
            return false;
        }
        //反序，从最近的表中开始读
        $tables = array_reverse($tables);
        foreach ($tables as $tb_name) {
            $sql = 'select * from '.$tb_name.' where '.$key_name.'='.$id.' limit 1';
            if ($data = $this->db->query($sql)) {
                \Org\FPC::redisHyunA()->set($cache_key, json_encode($data[0]), 60); //缓存时间1分钟，如果有数据变动记得清除缓存
                break;
            }
        }
        return $data?$data[0]:null;
    }

    /*
     * 删除"按数据量分表"的数据列表的缓存
     */
    public function delete_subtable_page_list_cache($table_name, $start=0, $page_size=20, $condition=[], $order_by=''){
        $condition = ksort($condition);
        $cache_key = md5( 'sptb_page_list_'.$table_name.'_'.$start.'_'.$page_size.'_'. json_encode($condition).'_'.$order_by );
        \Org\FPC::redisHyunA()->del($cache_key);
    }

    /*
     * 将搜索条件的数组转为sql的where语句
     */
    public function condition_to_sql($condition){
        $condition_sql = [];
        foreach ($condition as $key => $value) {
            if (is_numeric($value)) {
                $condition_sql[] = $key.'='.$value;
            }
            else if(is_array($value)) {
                //symbol为表达式符号，如 = in > < >= <=
                if (isset($value['symbol']) && isset($value['val'])) {
                    if (strtolower($value['symbol'])=='in') {
                        if(is_array($value['val'])) {
                            $condition_sql[] = $key.' in ("'. implode('","', $value['val']).'")';
                        }
                        else {
                            $condition_sql[] = $key.' in ('.$value['val'].')';
                        }
                    }
                    else {
                        $condition_sql[] = $key.$value['symbol'].'"'.$value['val'].'"';
                    }
                }
            }
            else {
                $condition_sql[] = $key.'="'.$value.'"';
            }
        }
        $condition_sql = implode(' and ', $condition_sql);
        return $condition_sql ? ' where '.$condition_sql : '';
    }

    /*
     * 缓存查询结果
     */
    public function cache_by_sql($sql, $ttl){
        $cache_key = md5( 'cache_by_query_sql_'.$sql);
        //先读缓存。缓存中无数据再读数据库
        if ($data = \Org\FPC::redisHyunA()->get($cache_key)) {
            return json_decode($data, true);
        }
        $data = $this->db->query($sql);
        if ($data) {
            \Org\FPC::redisHyunA()->set($cache_key, json_encode($data), $ttl);
        }
        return $data;
    }

    /*
     * 获得按数据量分表的数据列表
     * 分页查询
     * $condition 为搜索条件，多个搜索条件之间为并的关系
     * 搜索条件仅支持等于，如果业务需要更多查询条件方式，可扩展编码
     */
    public function get_subtable_page_list($table_name, $start=0, $page_size=20, $condition=[], $order_by=''){
        $condition = ksort($condition);
        $cache_key = md5( 'sptb_page_list_'.$table_name.'_'.$start.'_'.$page_size.'_'. json_encode($condition).'_'.$order_by );
        //先读缓存。缓存中无数据再读数据库
        if ($data_list = \Org\FPC::redisHyunA()->get($cache_key)) {
            return json_decode($data_list, true);
        }
        $condition_sql = $this->condition_to_sql($condition);

        $tables = $this->get_subtable($table_name);
        if (!$tables) {
            return false;
        }
        //反序，从最近的表中开始读
        $tables = array_reverse($tables);
        $total_count = 0;   //总记录数
        $in_those_tables = [];  //在这些表中查询所需数据
        $left_page_size = $page_size;
        foreach ($tables as $key => $tb_name) {
            $sql = 'select count(1) as c from `'.$tb_name.$condition_sql.'`';
            if ($count = $this->cache_by_sql($sql, ($key==0?3:60) )) {
                $count = intval($count[0]['c']);
                //如果数量在需要读的数据量范围内
                if ( $total_count+$count>=$start && $total_count<=$start+$page_size ) {
                    if ($left_page_size>0) {
                        $st = $start-$total_count>0 ? $start-$total_count : 0;
                        $size = $count-$st>=$left_page_size ? $left_page_size : $count-$st;
                        $in_those_tables[] = [
                            'table' => $tb_name,
                            'start' => $st,
                            'size' => $size,
                        ];
                        $left_page_size -= $size;
                    }
                }
                $total_count += $count;
            }
            else{
                return false;
            }
        }
        $data = [];
        $order_by = $order_by ? ' order by '.$order_by : '';
        foreach ($in_those_tables as $value) {
            $sql = 'select * from `'.$table_name.'` '.$condition_sql.$order_by.' limit '.$value['start'].','.$value['size'];
            if ($tmp = $this->db->query($sql)) {
                $data = array_merge($data, $tmp);
            }
            else {
                return false;
            }
        }
        $data_list = [
            'data' => $data,
            'start' => $start,
            'page_size' => $page_size,
            'total_count' => $total_count,
        ];
        if ($data) {
            //缓存时间5秒钟，如果有数据变动，并且能预知查询方式，可以及时清除缓存
            \Org\FPC::redisHyunA()->set($cache_key, json_encode($data_list), 100);
        }
        // var_dump($total_count, $in_those_tables, $data);
        // die;
        return $data_list;
    }
}