背景:
公司某業務模組的使用者通訊記錄和通話記錄的表，單表記錄過億了,mysql很堅挺好不好！！！！
其中某表是205773235 約2.0億
另外某表是575213155 約5.7億
使用者數量 約100W
其中通訊錄有的使用者是有很多重複記錄的,之前設計資料庫的同事按條來存的,也就是說一個使用者會有多條記錄，還有重複的記錄(沒有去重),歷史遺留問題那就不廢話了,我們存在的意義就是把不合理的設計去規範化來
如: id:1 uid:10086 name: 馬雲爸爸 num:88888
id:1 uid:10086 name: 馬雲爸爸 num:88888
這樣的資料表設計在資料量少的時候是沒有問題的,不去重這點會佔用額外的空間，其實這些不太常用而且有很多一對多關係的資料可以存json或者直接用OSS來存的（ps我們技術顧問推薦的儲存方式）
大概思路：
１.是把100Ｗ的使用者拆分成10份，每次迴圈10Ｗ個使用者的資料上傳
２.打通oss介面，檔案儲存按照日期＋小時這樣子儲存檔案
３.儲存方式json檔案方式儲存
4.資料安全方面,用動態密匙加密json，動態的密匙需要按規律儲存好（因為檔案要開放公共讀的,處於對使用者資料的安全考慮得加密）
5.上傳成功返回url儲存好
異常處理:
1.檔案上傳必然會有上傳失敗的,每個檔案迴圈上傳,如果上傳成功,則跳出迴圈,否則程式休息0.5秒繼續上傳,最大限度迴圈10次,如果還是上傳失敗,把這個使用者的uid push到redis的連結串列裡面
2.檔案上傳完成後儲存上傳路徑,同理儲存也有可能儲存失敗,也是上面的套路,做一個迴圈,如果儲存一次成功了就跳出迴圈,否則資訊0.5秒繼續發起儲存sql，成功推出迴圈,失敗10次儲存uid到redis上面
3.每10分鐘檢查一次redis裡面的上傳失敗或者儲存使用者資料失敗的使用者,如果超過500條則繼續上傳,這裡的套路和1.2點同理，如果還是失敗,到這裡還失敗的,其實使用者已經失敗了20次了,那麼應該屬於重災區,把這些使用者的uid儲存到資料庫裡面
4.1.2.3步驟都有和檔案日誌結合的，我的目標是不要有問題，少有問題！
最終實踐：
1.其實我沒有直接拿使用者表的uid,因為有小部分的使用者是沒有資料的，我建立了一張表,專門來儲存符合調教的使用者的uid，匯入適合的資料的命令如花：

INSERT INTO 某臨時表 (uid) 
select * from (SELECT ct.uid FROM (select max(id) as id from  某符合條件的表啦 group by uid ) pct join  某符合條件的表啦 ct on ct.id=pct.id ) as xxx

2.一步步找到之前的儲存資料的方式改成現在的模式
3.各種測試方法的可行性

具體遷移方法：

    /**
     *  通話記錄遷移 written : yangxingyi date: 2017-09-11
     */
    public function send_phone
 
()
    {
        ini_set('memory_limit','10240M');//限制記憶體10GB
        $start = trim(I('start'));
        $end   = trim(I('end'));//2017-09-18 15:47 區間1000-100000
        ($start>$end)&&exit("輸入範圍有誤!");
        (!$start||!$end)&&exit("沒有輸入條件!");

        $uids = 某使用者表模型->where("id between $start and $end"
 
)->field('uid')->select();
        $uids||exit('沒有符合條件的資料!');
        file_log('send_phone','正常資料的開始時間','phone');
        $oss = new OssCore();
        $que = 例項化redis;
        $t1 = $t2 = time();//記錄耗時
        $bucket = self::$contact_bucket; // 儲存空間

        $result = $save_res ='';
        $success_count = $false_count = 0;//記錄成功失敗條數
        $total_count   = count($uids);//記錄總條數
        while(list($index,$item)=each($uids))
        {
            $uid = $item['uid'];
            $exist_oss_path = 某url表->where(array('uid' => $uid))
            ->order("id desc")->field('oss_path')->find();
            if($exist_oss_path['oss_path']){
                file_log('send_phone','已上傳通訊錄的uid: '.$uid,'phone');
                continue;//跳過已經上傳過內容的使用者
            }

            $info = 某通話記錄表->where(array('uid' => $uid))->field('name,number')->select();
            if(!$info||!is_array($info)){
                file_log('send_phone','沒有通訊錄的uid: '.$uid,'phone');
                continue;//跳過沒有內容的使用者
            }

            $path = 'phone/' . date("Ymd") . '/' . date('H') . '/' . $uid;
            $content = 某加密函式(json_encode(make_array_uinque($info,'name','number')),self::$contact_key_prefix.某隨機字串);
            //組裝資料,並且根據name,number欄位去重
            for($i=0;$i<10;$i++){
                $result = $oss->uploadToOss($path, $content,$bucket );
                //uploadToOss(路徑,內容,$bucket="儲存空間")
                if($result){
                    break;
                }else{
                    usleep(500);file_log('send_phone','上傳oss失敗uid: '.$uid." 次數".$i,'phone');
                }
            }
            //echo $result['info']['url']."<hr>";oss返回的url
            for($i=0;$i<10;$i++){
                $save_res = $this->save_oss_path('',$result['info']['url'],$uid);//id是pk(這裡沒pk),傳第二,第三個引數
                if($save_res){
                    break;
                }else{
                    usleep(500);file_log('send_phone','修改資料庫失敗uid: '.$uid." 次數".$i,'phone');
                }
            }
            if(($index>0&&$index%1000)==0){
                $t1000 = time()- $t1;
                file_log('send_phone','千條耗時(秒): '.$t1000,'phone');
                $t1 = time();
            }
            if(($index>0&&$index%10000)==0){
                $t10000 = time()- $t2;
                file_log('send_phone','萬條耗時(秒): '.$t10000,'phone');
                $t2 = time();
            }
            //異常處理
            if (!$result||!$save_res)
            {
                file_log('send_phone','上傳失敗的uid: '.$uid,'phone');
                $que->lpush('send_phone_ids_error', $uid);
                $false_count++;
            }else{
                $que->incr('send_phone');//ok的數目
                $success_count++;//成功條數
            }//異常處理完畢
            unset($uids[$index]);//悉放記憶體
        }
        $tt = time()-$t1;
        file_log('send_phone','結束時間,已上傳總條數:'.$que->get('send_phone').
            ",本次總條數:$total_count,成功:$success_count ,失敗:$false_count 消耗時間(秒):$tt,id區間$start-$end",'phone');
    }

異常處理函式,每10分鐘執行一次：

    /**
     *  儲存資料和異常處理 written : yangxingyi date: 2017-09-12 11:45 Email: [email protected]
     */
    public function  handle_exception(){
        $oss = new OssCore();
        $que = 例項化redis;
        $bucket = self::$contact_bucket; // 儲存空間
        $list_count = $que->llen('send_phone_ids_error');
        $time = time();
        if ($list_count > 500){
            $error_ids = $que->lrange('send_phone_ids_error', 0, -1);//返回來的一維陣列
            $que->del('send_phone_ids_error');//返回來資料就幹掉它
            file_log('handle_exception','大於500條異常資料的開始時間','phone');

            $result = $save_res = '';
            foreach($error_ids as $item)
            {
                $uid = $item;
                $info = 某源資料表->where(array('uid' => $uid))->field('name,number')->select();
                if(!$info||!is_array($info)){
                    continue;//跳過沒有內容的使用者
                }
                $path = 'phone/' . date("Ymd") . '/' . date('H') . '/' . $uid;
                $content = 加密函式(json_encode(make_array_uinque($info,'name','number')),self::$contact_key_prefix.什麼鬼字串);//組裝資料
                for($i=0;$i<10;$i++){
                    $result = $oss->uploadToOss($path, $content,$bucket );//uploadToOss(路徑,內容,$bucket="儲存空間")
                    if($result){
                        break;
                    }else{
                        sleep(1);
                    }
                }

                for($i=0;$i<10;$i++){
                    $save_res = $this->save_oss_path('',$result['info']['url'],$uid);
                    //id是主鍵索引(這沒有),傳遞兩個引數,需要傳第三個引數uid
                    if($save_res){
                        break;
                    }else{
                        sleep(1);
                    }
                }

                //第二次的異常處理
                if ($result&&$save_res)
                {
                    $que->incr('send_phone');
                }else{
                    file_log('handle_exception','第二次的異常處理uid: '.$uid,'phone');
                    儲存錯誤的uid表->add(array('uid'=>$uid,'time'=>$time));
                }

            }
        }
    }

二維陣列去重的方法：

function make_array_uinque($arr,$type1='username',$type2='uid')
{
    $arr_out = $arr_wish = [];
    foreach($arr as $k => $v)
    {
        $key_out = $v["$type1"]."-".$v["$type2"];//提取內部一維陣列的key(任意值)作為外部陣列的鍵

        if(array_key_exists($key_out,$arr_out)){
            continue;
        }else{
            $arr_out[$key_out] = $v; //以key_out作為外部陣列的鍵
            $arr_wish[$k] = $v;  //實現二維陣列唯一性
        }
    }
    return $arr_wish;
}

如果您讀到了這裡，謝謝各位看官，看完這麼枯燥的文章！！！方法有不足的地方，歡迎各位大神指教！！！如果覺得對您有用的話歡迎轉載，如果您有什麼好文章也歡迎跟我分享！！！小弟不勝感激！！！！