久9在线视频,激情五月天快乐激情网,青青草厂元视频

前言

傳統(tǒng)的程序執(zhí)行流程一般是即時|同步|串行的，在某些場景下，會存在并發(fā)低，吞吐量低，響應(yīng)時間長等問題。在大型系統(tǒng)中，一般會引入消息隊列的組件，將流程中部分任務(wù)抽離出來放入消息隊列，并由專門的消費者作針對性的處理，從而降低系統(tǒng)耦合度，提高系統(tǒng)性能和可用性。

thinkphp-queue 是thinkphp 官方提供的一個消息隊列服務(wù)，它支持消息隊列的一些基本特性：

消息的發(fā)布，獲取，執(zhí)行，刪除，重發(fā)，失敗處理，延遲執(zhí)行，超時控制等
隊列的多隊列，內(nèi)存限制，啟動，停止，守護等
消息隊列可降級為同步執(zhí)行

thinkphp-queue 內(nèi)置了 Redis，Database，Topthink ，Sync這四種驅(qū)動。

本文主要介紹如何使用tp5自帶的think-queue消息隊列結(jié)合supervisor進程管理使隊列能夠常駐進程。

think-queue安裝與基本使用

tp5框架及think-queue的安裝方法及隊列驅(qū)動配置

tp5框架及think-queue安裝

推薦使用composer安裝

tp5安裝

composer create-project topthink/think 5.0.*

think-queue安裝
```
composer require topthink/think-queue
```

消息隊列的驅(qū)動配置

配置文件在項目的路徑如下圖：

tp-queue-01.png

內(nèi)容如下：
```
return [
    'connector'  => 'Redis',        // Redis 驅(qū)動
    'expire'     => 60,     // 任務(wù)的過期時間，默認為60秒; 若要禁用，則設(shè)置為 null
    'default'    => 'default',      // 默認的隊列名稱
    'host'       => '127.0.0.1',    // redis 主機ip
    'port'       => 6379,       // redis 端口
    'password'   => '',     // redis 密碼
    'select'     => 0,      // 使用哪一個 db，默認為 db0
    'timeout'    => 0,      // redis連接的超時時間
    'persistent' => false,      // 是否是長連接
];
```
> 具體配置可根據(jù)實際情況自行調(diào)整

創(chuàng)建一張表，用于展示消費隊列寫入數(shù)據(jù)庫的操作

CREATE TABLE `test` (
`id` int(10) NOT NULL AUTO_INCREMENT,
`task_type` varchar(50) DEFAULT '' COMMENT '任務(wù)類型',
`data` text COMMENT '數(shù)據(jù)',
`pdate` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '時間',
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8

創(chuàng)建消息隊列任務(wù)

入隊(生產(chǎn)者)

在index模塊新增 \application\index\controller\JobTest.php 控制器，在該控制器中添加 actionWithHelloJob 方法

生產(chǎn)者推送消息到隊列有2種方法：push()和later(),push是立即執(zhí)行，later是推送到隊列里，延遲執(zhí)行。代碼如下
```
public function actionWithHelloJob(){

      // 1.當(dāng)前任務(wù)將由哪個類來負責(zé)處理。
      //   當(dāng)輪到該任務(wù)時，系統(tǒng)將生成一個該類的實例，并調(diào)用其 fire 方法
      $jobHandlerClassName  = 'app\index\job\Hello';

      // 2.當(dāng)前任務(wù)歸屬的隊列名稱，如果為新隊列，會自動創(chuàng)建
      $jobQueueName     = "helloJobQueue";

      // 3.當(dāng)前任務(wù)所需的業(yè)務(wù)數(shù)據(jù) . 不能為 resource 類型，其他類型最終將轉(zhuǎn)化為json形式的字符串
      $jobData          = [ 'ts' => time(), 'bizId' => uniqid() , 'data' => $_GET ] ;

      // 4.將該任務(wù)推送到消息隊列，等待對應(yīng)的消費者去執(zhí)行

      $isPushed = Queue::push( $jobHandlerClassName , $jobData , $jobQueueName );

      //$isPushed = Queue::later(10,$jobHandlerClassName,$jobData,$jobQueueName); //把任務(wù)分配到隊列中，延遲10s后執(zhí)行

      // database 驅(qū)動時，返回值為 1|false  ;   redis 驅(qū)動時，返回值為 隨機字符串|false
      if( $isPushed !== false ){
          echo date('Y-m-d H:i:s') . " a new Hello Job is Pushed to the MQ"."<br>";
      }else{
          echo 'something went wrong.';
      }
  }

消費者的消費與刪除

創(chuàng)建Hello 消費者類，用于處理 helloJobQueue 隊列中的任務(wù)；新增 \application\index\job\Hello.php 消費者類，并編寫其 fire() 方法
代碼如下：

<?php
namespace app\index\job;
use think\queue\Job;

class Hello {
    public function fire(Job $job,$data) {
        // 有些消息在到達消費者時,可能已經(jīng)不再需要執(zhí)行了
        $isJobStillNeedToBeDone = $this->checkDatabaseToSeeIfJobNeedToBeDone($data);
        if(!$isJobStillNeedToBeDone){
            $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) {
                //通過這個方法可以檢查這個任務(wù)已經(jīng)重試了幾次了
                print("<warn>Hello Job has been retried more than 3 times!"."</warn>\n");

                $job->delete();

                // 也可以重新發(fā)布這個任務(wù)
                //print("<info>Hello Job will be availabe again after 2s."."</info>\n");
                //$job->release(2); //$delay為延遲時間，表示該任務(wù)延遲2秒后再執(zhí)行
            }
        }
    }

    /**
    * 有些消息在到達消費者時,可能已經(jīng)不再需要執(zhí)行了
    * @param array|mixed    $data     發(fā)布任務(wù)時自定義的數(shù)據(jù)
    * @return boolean                 任務(wù)執(zhí)行的結(jié)果
    */
    private function checkDatabaseToSeeIfJobNeedToBeDone($data){
        return true;
    }

    /**
    * 根據(jù)消息中的數(shù)據(jù)進行實際的業(yè)務(wù)處理...
    */
    private function doHelloJob($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;
    }
}

執(zhí)行之前，看下現(xiàn)在的目錄結(jié)構(gòu)

tp-queue-02.png

發(fā)布任務(wù)(生產(chǎn)者)

在瀏覽器中訪問 http://your.project.domain/index/JobTest/actionWithHelloJob ,可以看到消息推送成功

tp-queue-03.png
處理任務(wù)(消費者)
打開終端切換到當(dāng)前項目根目錄下，執(zhí)行下面的命令：

work命令又可分為單次執(zhí)行和循環(huán)執(zhí)行兩種模式。

單次執(zhí)行：不添加 --daemon參數(shù)，該模式下,work進程在處理完下一個消息后直接結(jié)束當(dāng)前進程。當(dāng)隊列為空時，會sleep一段時間然后退出。

循環(huán)執(zhí)行：添加了 --daemon參數(shù)，該模式下,work進程會循環(huán)地處理隊列中的消息，直到內(nèi)存超出參數(shù)配置才結(jié)束進程。當(dāng)隊列為空時，會在每次循環(huán)中sleep一段時間。
```
php think queue:work --daemon --queue helloJobQueue
```
會看到如下信息：
```
[root@localhost tpqueue]# php think queue:work --daemon --queue helloJobQueue
<info>Hello Job Started. job Data is: array (
'ts' => 1565246136,
'bizId' => '5d4bc2b88f03b',
'data' => 
array (
),
)</info> 
<info>Hello Job is Fired at 2019-08-08 14:35:39</info> 
<info>Hello Job is Done!</info> 
<info>Hello Job has been done and deleted</info>
Processed: app\index\job\Hello
```
到這里我們成功的經(jīng)歷了一個消息的創(chuàng)建->推送->消費->刪除的基本流程

消息隊列的開始，停止與重啟

開始一個消息隊列：
```
php think queue:work
```
停止所有的消息隊列：
```
php think queue:restart
```

重啟所有的消息隊列：

php think queue:restart 
php think queue:work

多模塊，多任務(wù)的處理

多模塊

單模塊項目推薦使用 app\job 作為任務(wù)類的命名空間
多模塊項目可用使用 app\module\job 作為任務(wù)類的命名空間也可以放在任意可以自動加載到的地方

多任務(wù)

多任務(wù)例子:

在 \application\index\controller\JobTest.php 控制器中，添加 multiTask()方法：

public function multiTask() {
    $taskType = $_GET['taskType'];
    switch ($taskType) {
        case "taskOne":
            $jobHandleClassName = "app\index\job\multiTask@taskOne";
            $jobQueueName = "taskOneQueue";
            $jobData = ['ts'=>time(), 'bizId'=>uniqid(), 'data'=>$_GET];
            break;
        case "taskTwo":
            $jobHandleClassName = "app\index\job\multiTask@taskTwo";
            $jobQueueName = "taskTwoQueue";
            $jobData = ['ts'=>time(), 'bizId'=>uniqid(), 'data'=>$_GET];
            break;
        default:
            break;
    }
    $isPushed = Queue::push($jobHandleClassName, $jobData, $jobQueueName);
    if ($isPushed!==false) {
        echo date('Y-m-d H:i:s')."the $taskType of multiTask job has been pushed to $jobQueueName <br>";
    }else {
        throw new Exception("push a new $taskType of multiTask job Failed!");
    }
}

新增 \application\index\job\MultiTask.php 消費者類，并編寫其 taskOne() 和 taskTwo()方法

<?php
namespace app\index\job;


use think\queue\Job;

class MultiTask {
    public function taskOne(Job $job, $data) {
        $isDone = $this->doTaskOne($data);
        if ($isDone) {
            $job->delete();
            print ("INFO:the taskOne of multiTask has been done and delete!\n");
            return;
        }else {
            if ($job->attempts()>3) {
                $job->delete();
            }
        }
    }
    public function taskTwo(Job $job, $data) {
        $isDone = $this->doTaskTwo($data);
        if ($isDone) {
            $job->delete();
            print ("INFO:the taskTwo of multiTask has been done and delete! \n");
        }else {
            if ($job->attempts()>3) {
                $job->delete();
            }
        }
    }

    private function doTaskOne($data) {
        $id = db('test')->insertGetId(['task_type'=>'task one','data'=>json_encode($data)]);
        print ("INFO: doing taskOne of multiTask! the db return id is :$id\n");
        return true;
    }

    private function doTaskTwo($data) {
        $id = db('test')->insertGetId(['task_type'=>'task two','data'=>json_encode($data)]);
        print ("INFO: doing taskTwo of multiTask! the db return id is :$id\n");
        return true;
    }
}

最終執(zhí)行結(jié)果如下：

tp-queue-04.png

tp-queue-05.png

supervisor的安裝和配置

supervisor是用Python開發(fā)的一個client/server服務(wù)，是Linux/Unix系統(tǒng)下的一個進程管理工具。可以很方便的監(jiān)聽、啟動、停止、重啟一個或多個進程。用supervisor管理的進程，當(dāng)一個進程意外被殺死，supervisor監(jiān)聽到進程死后，會自動將它重啟，很方便的做到進程自動恢復(fù)的功能，不再需要自己寫shell腳本來控制。

yum安裝supervisor

yum install epel-release
yum install supervisor
//設(shè)置開機自動啟動
systemctl enable supervisord

配置
找到/etc/supervisord.conf配置文件，編輯信息如下：

[unix_http_server]
file=/tmp/supervisor.sock   ; the path to the socket file
;chmod=0700                 ; socket file mode (default 0700)
;chown=nobody:nogroup       ; socket file uid:gid owner
;username=user              ; default is no username (open server)
;password=123               ; default is no password (open server)

supervisord]
logfile=/tmp/supervisord.log ; main log file; default $CWD/supervisord.log
logfile_maxbytes=50MB        ; max main logfile bytes b4 rotation; default 50MB
logfile_backups=10           ; # of main logfile backups; 0 means none, default 10
loglevel=info                ; log level; default info; others: debug,warn,trace
pidfile=/tmp/supervisord.pid ; supervisord pidfile; default supervisord.pid
nodaemon=false               ; start in foreground if true; default false
minfds=1024                  ; min. avail startup file descriptors; default 1024
minprocs=200                 ; min. avail process descriptors;default 200

[supervisorctl]
serverurl=unix:///tmp/supervisor.sock ; use a unix:// URL  for a unix socket

[include]
;files = relative/directory/*.ini
files = /etc/supervisor/*.conf

file,logfile,pidfile,serverurl,files的路徑可根據(jù)自身需要去自定義

在/etc 目錄里創(chuàng)建一個supervisor文件，然后在/etc/supervisor目錄下創(chuàng)建一個.conf文件，這里命名為queue.conf。
對于index這個單模塊而言，不同的業(yè)務(wù)邏輯為了區(qū)分可能會存在多個隊列名，這種情況將多個隊列名用逗號拼接起來，內(nèi)容如下：
```
[program:queue]
user=root
command=php /www/wwwroot/tpqueue/think queue:work --queue helloJobQueue,taskOneQueue,taskTwoQueue --daemon
```
啟動supervisor
```
supervisorctl -c /etc/supervisord.conf
```
上面這個命令會進入 supervisorctl 的 shell 界面，然后可以執(zhí)行不同的命令了

status # 查看程序狀態(tài)
stop thrift-log # 關(guān)閉 usercenter 程序
start thrift-log # 啟動 usercenter 程序
restart thrift-log # 重啟 usercenter 程序
reread ＃讀取有更新（增加）的配置文件，不會啟動新添加的程序
update ＃重啟配置文件修改過的程序

例如啟動queue程序：

tp-queue-06.png

這時再去推送消息，可以看到如下信息：

tp-queue-07.png

>紅圈中的是日志文件，可見隊列消費完成，數(shù)據(jù)插入成功

數(shù)據(jù)庫插入數(shù)據(jù)如下：

tp-queue-08.png

結(jié)束

至此，tp5(think-queue)消息隊列結(jié)合supervisor已實現(xiàn)進程常駐

原文：西瓜很甜喲~的個人博客

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

Thinkphp5(think-queue)消息隊列結(jié)合supervisor進程管理實現(xiàn)隊列常駐進程

Thinkphp5(think-queue)消息隊列結(jié)合supervisor進程管理實現(xiàn)隊列常駐進程

前言

think-queue安裝與基本使用

tp5框架及think-queue的安裝方法及隊列驅(qū)動配置

創(chuàng)建消息隊列任務(wù)

supervisor的安裝和配置

結(jié)束

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九 欧美,1769亚洲,黄色成人av

Thinkphp5(think-queue)消息隊列結(jié)合supervisor進程管理實現(xiàn)隊列常駐進程

前言

think-queue安裝與基本使用

tp5框架及think-queue的安裝方法及隊列驅(qū)動配置

創(chuàng)建消息隊列任務(wù)

supervisor的安裝和配置

結(jié)束

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av