快捷導(dǎo)航

php解決注冊(cè)并發(fā)問(wèn)題并提高QPS

更新時(shí)間：2023年05月25日 10:34:49 作者：北橋蘇

這篇文章主要為大家介紹了php解決注冊(cè)并發(fā)問(wèn)題并提高QPS實(shí)現(xiàn)方法示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

前面在本地的windows通過(guò)apache的ab工具測(cè)試了600并發(fā)下“查詢指定手機(jī)是否存在再提交數(shù)據(jù)”的注冊(cè)功能會(huì)出現(xiàn)重復(fù)提交的情況，并且在注冊(cè)完成時(shí)還需要對(duì)邀請(qǐng)人進(jìn)行獎(jiǎng)勵(lì)，記錄邀請(qǐng)記錄，對(duì)該新用戶自動(dòng)發(fā)布動(dòng)態(tài)信息，發(fā)短信或發(fā)郵件等其他業(yè)務(wù)功能。所以這里當(dāng)并發(fā)時(shí)，注冊(cè)功能就變得低效且容易出現(xiàn)問(wèn)題。

先對(duì)重復(fù)提交的問(wèn)題通過(guò)redis解決，再把注冊(cè)儲(chǔ)存用戶基本信息以后的操作放到隊(duì)列中進(jìn)行異步執(zhí)行，可以很好的優(yōu)化注冊(cè)功能，提高QPS。

一、環(huán)境要求

PHP版本 >= 5.6.0

PHP框架：Thinkphp5.1.*

消息隊(duì)列：Think-queue2.0

PHP擴(kuò)展：Redis

二、下載框架和消息隊(duì)列中間件

下載tp5.1。composer create-project topthink/think=5.1.* tp5 --prefer-dist
安裝think-queue。composer require topthink/think-queue
php安裝redis擴(kuò)展和打開(kāi)redis服務(wù)端和客戶端。

三、解決注冊(cè)重復(fù)提交

配置文件中cache設(shè)置為redis驅(qū)動(dòng)，并新建控制器因?yàn)閏ache相關(guān)命名空間。

use think\Exception;
use think\facade\Cache;
use think\facade\Env;
use think\Queue;

使用無(wú)序集合存手機(jī)號(hào)，通過(guò)判斷當(dāng)前手機(jī)號(hào)是否是在指定鍵里為成員(如果注冊(cè)存入數(shù)據(jù)庫(kù)失敗，通過(guò)sRem刪除該成員)，然后再通過(guò)查詢數(shù)據(jù)庫(kù)判斷是否存在。

private $cache;
private  $handler;
// 實(shí)例化redis
public function __construct() {
    $this->cache = Cache::init();
    $this->handler = $this->cache->handler();
}
// 判斷手機(jī)號(hào)是否在集合中
$is_existe = $this->handler->sIsMember("register:mobile",$mobile);
if(!$is_existe) {
   $this->handler->sAdd("register:mobile",$mobile);
}else {
   //Log::write('---壓力測(cè)試'.date("Y-m-d h:i:s").'---手機(jī)號(hào)已存在');
   var_dump('手機(jī)號(hào)已存在');    // 用戶已存在
   die;
}
// 查詢手機(jī)號(hào)碼是否已注冊(cè)
$user = db('user')->field('mobile')->where('mobile', $mobile)->find();
if ($user) {
    //Log::write('---壓力測(cè)試'.date("Y-m-d h:i:s").'---手機(jī)號(hào)注冊(cè)了');
    var_dump('手機(jī)號(hào)已注冊(cè)');    // 用戶已存在
    die;
}

四、消息隊(duì)列分解注冊(cè)功能

配置消息隊(duì)列，后面以redis驅(qū)動(dòng)為例。

<?php
return [
    'connector'  => 'Redis',            // Redis 驅(qū)動(dòng)
    'expire'     => 60,                // 任務(wù)的過(guò)期時(shí)間，默認(rèn)為60秒; 若要禁用，則設(shè)置為 null
    'default'    => 'default',        // 默認(rèn)的隊(duì)列名稱
    'host'       => '127.0.0.1',        // redis 主機(jī)ip
    'port'       => 6379,            // redis 端口
    'password'   => '',                // redis 密碼
    'select'     => 0,                // 使用哪一個(gè) db，默認(rèn)為 db0
    'timeout'    => 0,                // redis連接的超時(shí)時(shí)間
    'persistent' => false,            // 是否是長(zhǎng)連接
    //    'connector' => 'Database',   // 數(shù)據(jù)庫(kù)驅(qū)動(dòng)
    //    'expire'    => 60,           // 任務(wù)的過(guò)期時(shí)間，默認(rèn)為60秒; 若要禁用，則設(shè)置為 null
    //    'default'   => 'default',    // 默認(rèn)的隊(duì)列名稱
    //    'table'     => 'jobs',       // 存儲(chǔ)消息的表名，不帶前綴
    //    'dsn'       => [],
    //    'connector'   => 'Topthink',    // ThinkPHP內(nèi)部的隊(duì)列通知服務(wù)平臺(tái) ，本文不作介紹
    //    'token'       => '',
    //    'project_id'  => '',
    //    'protocol'    => 'https',
    //    'host'        => 'qns.topthink.com',
    //    'port'        => 443,
    //    'api_version' => 1,
    //    'max_retries' => 3,
    //    'default'     => 'default',
//        'connector'   => 'Sync',        // Sync 驅(qū)動(dòng)，該驅(qū)動(dòng)的實(shí)際作用是取消消息隊(duì)列，還原為同步執(zhí)行
];

完成添加新用戶后將指定數(shù)據(jù)加入消息隊(duì)列。

<?php
namespace app\index\controller;
use think\Db;
use think\Validate;
use think\Exception;
use think\facade\Cache;
use think\facade\Env;
use think\Queue;
use think\Log;
class Index
{    
    private $cache;
    private  $handler;
    public function __construct() {
        $this->cache = Cache::init();
        $this->handler = $this->cache->handler();
    }
    public function index()
    {
        $data = input('post.');
        unset($data['balance']);
        unset($data['credit']);
        // $blacklist = [
        //     "18124198164","13401363108","17688552009","15089352898","13602940094","13346643336","13181351655","18301123028","13598020751","13014568187",
        //     "13428733909","17337991130","13275342497"
        // ];
        $rule = [
            'mobile' => 'require|number|length:11',
            'password' => 'require|length:6,32',
        ];
        $msg = [
            'mobile.require' => '手機(jī)號(hào)必須',
            'mobile.length' => '手機(jī)號(hào)為11位數(shù)字',
            'mobile.number' => '手機(jī)號(hào)為11位數(shù)字',
            'password.require' => '密碼必須',
            'password.length' => '密碼為6-12位之間',
        ];
        //驗(yàn)證數(shù)據(jù)是否合法
        $mobile = isset($data['mobile']) ? $data['mobile'] : '';
        $validate = new Validate($rule, $msg);
        $result = $validate->check($data);
        if (!$result) {
            var_dump($validate->getError());
            die;
        }
        // if(in_array($mobile,$blacklist)) {
        //     var_dump('該手機(jī)號(hào)已注冊(cè)了');    // 黑名單
        //     die;
        // }
        // 判斷手機(jī)號(hào)是否在集合中
        $is_existe = $this->handler->sIsMember("register:mobile",$mobile);
        if(!$is_existe) {
            $this->handler->sAdd("register:mobile",$mobile);
        }else {
            //Log::write('---壓力測(cè)試'.date("Y-m-d h:i:s").'---手機(jī)號(hào)已存在');
            var_dump('手機(jī)號(hào)已存在');    // 用戶已存在
            die;
        }
        // 查詢手機(jī)號(hào)碼是否已注冊(cè)
        $user = db('user')->field('mobile')->where('mobile', $mobile)->find();
        if ($user) {
            //Log::write('---壓力測(cè)試'.date("Y-m-d h:i:s").'---手機(jī)號(hào)注冊(cè)了');
            var_dump('手機(jī)號(hào)已注冊(cè)');    // 用戶已存在
            die;
        }       
        // 用戶不存在注冊(cè)
        // $data['id']          = getNewUserid();
        $data['no'] = date("Ymdhis").rand(100, 999);
        $data['avatar'] = 'https://rumcdn-1255484416.cos.ap-chengdu.myqcloud.com/img/d_h.png';
        $data['password'] = md5($data['password']);
        $randomNickname = date("Ymdhis").rand(100, 999);         
        $data['nickname'] = 'rm_' . $randomNickname;
        $data['create_time'] = time();
        $data['type'] = 1;
        /***是否存在邀請(qǐng)人的跑步錢(qián)進(jìn)號(hào)***/
        if(isset($data['pbqj_no']) && !empty($data['pbqj_no'])) {
            $inviter = db('user')->field('id')->where(["no"=>$data['pbqj_no']])->find();
            if($inviter) {
                $data['inviter_id'] = $inviter['id'];
            }
        }
        /***是否存在邀請(qǐng)人的跑步錢(qián)進(jìn)號(hào)***/
        unset($data['pbqj_no']);
        $userid = db('user')->insertGetId($data);
        if ($userid) {
        /******************加入消息隊(duì)列異步處理后續(xù)操作*******************/
            // 1.當(dāng)前任務(wù)將由哪個(gè)類來(lái)負(fù)責(zé)處理。 
            // 當(dāng)輪到該任務(wù)時(shí)，系統(tǒng)將生成一個(gè)該類的實(shí)例，并調(diào)用其 fire 方法
            $jobHandlerClassName  = 'app\index\job\JobUser'; 
            // 2.當(dāng)前任務(wù)歸屬的隊(duì)列名稱，如果為新隊(duì)列，會(huì)自動(dòng)創(chuàng)建
            $jobQueueName         = "userJobQueue"; 
            // 3.當(dāng)前任務(wù)所需的業(yè)務(wù)數(shù)據(jù) . 不能為 resource 類型，其他類型最終將轉(zhuǎn)化為json形式的字符串
            // ( jobData 為對(duì)象時(shí)，需要在先在此處手動(dòng)序列化，否則只存儲(chǔ)其public屬性的鍵值對(duì))
            //$jobData              = ['ts' => time(), 'bizId' => uniqid() , 'a' => 1];
            $jobData              = ['userid'=>$userid,'time'=>time(),'mobile'=>$mobile,'inviterid'=>(isset($data['inviter_id']) ? $data['inviter_id'] : 0)];
            // 4.將該任務(wù)推送到消息隊(duì)列，等待對(duì)應(yīng)的消費(fèi)者去執(zhí)行
            $isPushed = Queue::push($jobHandlerClassName , $jobData , $jobQueueName);    
            // database 驅(qū)動(dòng)時(shí)，返回值為 1|false  ;   redis 驅(qū)動(dòng)時(shí)，返回值為 隨機(jī)字符串|false
            if($isPushed !== false) { 
                var_dump('加入隊(duì)列成功');
                die;
                //Log::write('-----------加入消息隊(duì)列成功-----------');
                //echo date('Y-m-d H:i:s') . " a new Hello Job is Pushed to the MQ"."
";
            }else{
                var_dump('加入消息隊(duì)列');
                die;
                //Log::write('-----------加入消息隊(duì)列失敗-----------');
                //echo 'Oops, something went wrong.';
            }
        /******************加入消息隊(duì)列異步處理后續(xù)操作*******************/
            $res['id'] = $userid;
            $res['no'] = $data['no'];
            // // token處理類
            // $accessToken = new AccessToken();
            // $accessToken = $accessToken->getToken($userid);
            // if (empty($accessToken)) {
            //     //Log::write('---壓力測(cè)試'.date("Y-m-d h:i:s").'---秘鑰生成失敗');
            //     var_dump('秘鑰生成失敗');
            // } else {
            //     $res['user_token'] = $accessToken;
            // }
            // if (method_exists(\chat\User::class, 'getToken')) {
            //     $chat_token = \chat\User::getToken($res['id'], $data['nickname'], $data['avatar']);
            //     if (!$chat_token) {
            //         //Log::write('---壓力測(cè)試'.date("Y-m-d h:i:s").'---聊天秘鑰生成失敗');
            //         var_dump('聊天秘鑰生成失敗');
            //     } else {
            //         $res['chat_token'] = $chat_token;
            //     }
            // } else {
            //     $res['chat_token'] = '';
            // }
            //Log::write('---壓力測(cè)試'.date("Y-m-d h:i:s").'---注冊(cè)成功');
            var_dump($res);
            die;
        } else {
            //Log::write('---壓力測(cè)試'.date("Y-m-d h:i:s").'---數(shù)據(jù)庫(kù)錯(cuò)誤');
            $this->handler->sRem("register:mobile",$mobile);
            var_dump('數(shù)據(jù)庫(kù)錯(cuò)誤');
            die;
        }
    }
    public function hello($name = 'ThinkPHP5')
    {
        return 'hello,' . $name;
    }
}

創(chuàng)建消費(fèi)者(job)，對(duì)執(zhí)行隊(duì)列中的任務(wù)。

(1). 在同一模塊下新建job文件夾和一個(gè)執(zhí)行類(JobUser), 需要對(duì)應(yīng)生產(chǎn)者中jobHandlerClassName。

(2). 前面執(zhí)行完隊(duì)列加入成功后，可以本地使用redis客戶端通過(guò)lrange queues:userJobQueue 0 -1 查看隊(duì)列成員

(queues:userJobQueue中,userJobQueue是自己在加入隊(duì)列前自己起的隊(duì)列名稱，與queues: 拼接就是redis的list的鍵名，所以可以直接查看 )。

(3).隊(duì)列中的data就是自己傳遞的數(shù)據(jù)，后面需要在消費(fèi)者中通過(guò)該數(shù)據(jù)進(jìn)行注冊(cè)功能后的業(yè)務(wù)操作: 送獎(jiǎng)勵(lì)，存儲(chǔ)邀請(qǐng)記錄，發(fā)動(dòng)態(tài)，發(fā)短信，發(fā)郵件等等。

<?php
namespace app\index\job;
use think\queue\Job;
use think\Db;
use think\Exception;
use think\facade\Cache;
use think\facade\Env;
class JobUser {
    private  $cache;
    private  $handler;
    public  function  __construct()
    {
        $this->cache = Cache::init();
        $this->handler = $this->cache->handler();
    }
    /**
     * fire方法是消息隊(duì)列默認(rèn)調(diào)用的方法
     * @param Job            $job      當(dāng)前的任務(wù)對(duì)象
     * @param array|mixed    $data     發(fā)布任務(wù)時(shí)自定義的數(shù)據(jù)
     */
    public function fire(Job $job,$data) {
        $job->delete();
        //print("hahah\n");
        // print("<info>The user already exists "."</info>\n");
        //     exit();
        if(empty($data) || empty($data['userid']) || empty($data['mobile'])) {
            $job->delete();
            print("canshu buzu\n");
            return;
        }
        // 如有必要,可以根據(jù)業(yè)務(wù)需求和數(shù)據(jù)庫(kù)中的最新數(shù)據(jù),判斷該任務(wù)是否仍有必要執(zhí)行.
        $isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);
        if(!$isJobStillNeedToBeDone) {
            print("hahah\n");
            $job->delete();
            return;
        }
        $isJobDone = $this->doHelloJob($data);
        if ($isJobDone) {
            //如果任務(wù)執(zhí)行成功， 記得刪除任務(wù)
            $job->delete();
            print("<info>Hello Job has been done and deleted"."</info>\n");
        }else{
            if ($job->attempts() > 3) {
                //通過(guò)這個(gè)方法可以檢查這個(gè)任務(wù)已經(jīng)重試了幾次了
                print("<warn>Hello Job has been retried more than 3 times!"."</warn>\n");
                //$job->delete();
                // 也可以重新發(fā)布這個(gè)任務(wù)
                //print("<info>Hello Job will be availabe again after 2s."."</info>\n");
                //$job->release(2); //$delay為延遲時(shí)間，表示該任務(wù)延遲2秒后再執(zhí)行
            }
        }
    }
    /**
     * 有些消息在到達(dá)消費(fèi)者時(shí),可能已經(jīng)不再需要執(zhí)行了
     * @param array|mixed    $data     發(fā)布任務(wù)時(shí)自定義的數(shù)據(jù)
     * @return boolean                 任務(wù)執(zhí)行的結(jié)果
     */
    private function checkDatabaseToSeeIfJobNeedToBeDone($data) {
        // 判斷手機(jī)緩存集合中是否存在
        // $is_existe = $this->handler->sIsMember("register:mobile",$data['mobile']);
        // if($is_existe) {
        //     return false;  
        // } 
        // // 查詢當(dāng)前用戶是否在數(shù)據(jù)庫(kù)中存在
        // $userinfo = Db::name('user')->field('id')->where('id',$data['userid'])->find();
        // if($userinfo) {
        //     return false;  
        // } 
        return true;
    }
    /**
     * 根據(jù)消息中的數(shù)據(jù)進(jìn)行實(shí)際的業(yè)務(wù)處理
     * @param array|mixed    $data     發(fā)布任務(wù)時(shí)自定義的數(shù)據(jù)
     * @return boolean                 任務(wù)執(zhí)行的結(jié)果
    */
    private function doHelloJob($data) {
        try{
            if(isset($data['inviterid']) && !empty($data['inviterid'])) {
                // 添加邀請(qǐng)記錄
                $res_record = Db::name('user_inviter')
                    ->insert([
                        'inviterid'   => $data['inviterid'],
                        'userid'      => $data['userid'],
                        'code'        => $data['inviterid'] . 'T' . $data['userid'],
                        'create_time' => $data['time'],
                ]);
                // 給邀請(qǐng)人贈(zèng)送300步幣
                Db::name('user_credit')
                    ->insert([
                        'userid'      => $data['inviterid'],
                        'type'        => 1,
                        'credit'      => 300,
                        'source'      => $res_record,
                        'create_time' => $data['time']
                ]);
                // 更新邀請(qǐng)人步幣（用戶表）
                Db::name('user')->where('id', $data['inviterid'])->setInc('credit', 300);              
            }
            {   // 注冊(cè)成功發(fā)表動(dòng)態(tài)
                $dynamic_data['userid'] = $data['userid'];
                $dynamic_data['dynamic'] = base64_encode('號(hào)外！號(hào)外！我加入跑步錢(qián)進(jìn)了，大家一起走路領(lǐng)紅包吧！');
                $dynamic_data['images'][] = 'https://rumcdn-1255484416.cos.ap-chengdu.myqcloud.com/img/d_d.png';
                $dynamic_data['images'] = serialize($dynamic_data['images']);
                $dynamic_data['create_time'] = $data['time'];
                $result = Db::name('dynamic')->insert($dynamic_data);
            }
        }catch(\Exception $e) {
            Log::write('---執(zhí)行消息隊(duì)列出錯(cuò)---'.$e->getMessage());
            return false;
        }
        return true;
        // 根據(jù)消息中的數(shù)據(jù)進(jìn)行實(shí)際的業(yè)務(wù)處理...
        //var_dump($data);
//        print("<info>Hello Job Started. job Data is: ".var_export($data,true)."</info> \n");
//        print("<info>Hello Job is Fired at " . date('Y-m-d H:i:s') ."</info> \n");
//        print("<info>Hello Job is Done!"."</info> \n");
        //return true;
    }
    /**
     * 該方法用于接收任務(wù)執(zhí)行失敗的通知，你可以發(fā)送郵件給相應(yīng)的負(fù)責(zé)人員
     * @param $jobData  string|array|...      //發(fā)布任務(wù)時(shí)傳遞的 jobData 數(shù)據(jù)
    */
    public function failed($jobData) {
        //send_mail_to_somebody() ;
        print("Warning: Job failed after max retries. job data is :".var_export($jobData,true)."\n");
    }
}

(4). 設(shè)置任務(wù)執(zhí)行失敗后的處理，比如記錄日志或發(fā)郵件給開(kāi)發(fā)者。

a. 在tags.php中配置失敗后執(zhí)行了類。

<?php
// 應(yīng)用行為擴(kuò)展定義文件
return [
    // 應(yīng)用初始化
    'app_init'     => [],
    // 應(yīng)用開(kāi)始
    'app_begin'    => [],
    // 模塊初始化
    'module_init'  => [],
    // 操作開(kāi)始執(zhí)行
    'action_begin' => [],
    // 視圖內(nèi)容過(guò)濾
    'view_filter'  => [],
    // 日志寫(xiě)入
    'log_write'    => [],
    // 應(yīng)用結(jié)束
    'app_end'      => [],
    'queue_failed' => [
        // 數(shù)組形式，[ 'ClassName' , 'methodName']
        ['application\\behavior\\MyQueueFailedLogger', 'logAllFailedQueues']
        // 字符串(靜態(tài)方法)，'StaicClassName::methodName'
        // 'MyQueueFailedLogger::logAllFailedQueues'
        // 字符串(對(duì)象方法)，'ClassName'，此時(shí)需在對(duì)應(yīng)的ClassName類中添加一個(gè)名為 queueFailed 的方法
        // 'application\\behavior\\MyQueueFailedLogger'
        // 閉包形式
        /*
        function( &$jobObject , $extra){
            // var_dump($jobObject);
            return true;
        }
        */
    ],
];

b. 在application目錄下創(chuàng)建任務(wù)錯(cuò)誤執(zhí)行后的處理腳本，根據(jù)業(yè)務(wù)需求自定。

<?php
namespace app\behavior;
use think\Db;
class MyQueueFailedLogger
{
    const should_run_hook_callback = true;
    /**
     * @param $jobObject   \think\queue\Job   //任務(wù)對(duì)象，保存了該任務(wù)的執(zhí)行情況和業(yè)務(wù)數(shù)據(jù)
     * @return bool     true                  //是否需要?jiǎng)h除任務(wù)并觸發(fā)其failed() 方法
    */
    public function logAllFailedQueues(&$jobObject) {
        $failedJobLog = [
            'jobHandlerClassName'   => $jobObject->getName(), // 'application\index\job\Hello'
            'queueName' => $jobObject->getQueue(),               // 'helloJobQueue'
            'jobData'   => $jobObject->getRawBody()['data'],  // '{'a': 1 }'
            'attempts'  => $jobObject->attempts(),            // 3
        ];
        var_export(json_encode($failedJobLog,true));
        $data = [
            "content" => json_encode($failedJobLog,true),
            "create_time" => time(),
        ];
        Db::name('ztest')->insertGetId($data);
        // $jobObject->release();     //重發(fā)任務(wù)
        //$jobObject->delete();         //刪除任務(wù)
        //$jobObject->failed();      //通知消費(fèi)者類任務(wù)執(zhí)行失敗
        return self::should_run_hook_callback;
    }
}

五、通過(guò)命令運(yùn)行消息隊(duì)列，以下以windows舉栗

cmd進(jìn)入當(dāng)前項(xiàng)目, 然后輸入 "php think queue:listen --queue userJobQueue" (userJobQueue是自己的隊(duì)列名)。
也可以在項(xiàng)目的根目錄創(chuàng)建bat文件，文件寫(xiě)入"php think queue:listen --queue userJobQueue"，保存只需雙擊就可以執(zhí)行。