PHP7生產(chǎn)環(huán)境隊列Beanstalkd用法詳解

更新時間：2020年05月19日 08:16:56 作者：涼官灰

在本篇文章里小編給大家分享的是關(guān)于PHP7生產(chǎn)環(huán)境隊列Beanstalkd用法內(nèi)容，需要的朋友們可以參考下。

應(yīng)用場景

為什么要用呢，有什么好處？這應(yīng)該放在最開頭說，一件東西你只有了解它是干什么的，適合干什么，才能更好的與自己的項(xiàng)目相結(jié)合，用到哪里學(xué)到哪里，學(xué)了不用等于不會，我們平時就應(yīng)該多考慮一些這樣的問題：自己做個什么項(xiàng)目功能能跟 xx 技術(shù)相結(jié)合呢？這個 xx 技術(shù)放在這種業(yè)務(wù)場景下行不行呢？而不是 “學(xué)了這個 xx 技術(shù)能干嘛呢，公司現(xiàn)在也沒有用這個的呀，學(xué)了也沒用啊”，帶著這樣心情去學(xué)習(xí) xx 技術(shù)，肯定很痛苦。

隊列大家都知道是將一些耗時的操作先不去做，先埋點(diǎn)，再異步去處理，這樣對一些發(fā)郵件發(fā)短信之類的耗時操作，用戶是感覺不到的，因?yàn)槁顸c(diǎn)結(jié)束，操作也就結(jié)束了，消費(fèi)隊列都是在服務(wù)器上做的。主要應(yīng)用在短信或郵件通知，訪問第三方接口訂閱消息，商城的一些秒殺活動，都可以結(jié)合隊列來完成。

Beanstalkd 介紹

Beanstalkd 是一個高性能，輕量級的分布式內(nèi)存隊列，C 代碼，典型的類 Memcached 設(shè)計，協(xié)議和使用方式都是同樣的風(fēng)格，所以使用過 memcached 的用戶會覺得 Beanstalkd 似曾相識。

beanstalkd 的最初設(shè)計意圖是在高并發(fā)的網(wǎng)絡(luò)請求下，通過異步執(zhí)行耗時較多的請求，及時返回結(jié)果，減少請求的響應(yīng)延遲。

Ubuntu 安裝

sudo apt-get install beanstalkd

配置文件

vim /etc/default/beanstalkd

查看狀態(tài)

service beanstalkd status
# 命令回顯 #
root@:/www/server/php/72/etc# service beanstalkd status
● beanstalkd.service - Simple, fast work queue
  Loaded: loaded (/lib/systemd/system/beanstalkd.service; enabled; vendor preset: enabled)
  Active: active (running) since Tue 2018-10-16 10:42:28 CST; 6 days ago
   Docs: man:beanstalkd(1)
 Main PID: 7033 (beanstalkd)
  Tasks: 1 (limit: 4634)
  CGroup: /system.slice/beanstalkd.service
      └─7033 /usr/bin/beanstalkd -l 0.0.0.0 -p 11300 -b /var/lib/beanstalkd
Oct 16 10:42:28 ip-10-93-2-137 systemd[1]: Started Simple, fast work queue.

配置連通性 + 持久化

ip 用 0.0.0.0 允許所有連接，靠配置安全組或防火墻去約束連接，放開 -b 參數(shù) （默認(rèn)沒有持久化），內(nèi)存的隊列消息可以落地到硬盤 binlog 實(shí)現(xiàn)持久化，斷電可重新讀取隊列消息。

vim /etc/default/beanstalkd
BEANSTALKD_LISTEN_ADDR=0.0.0.0
BEANSTALKD_LISTEN_PORT=11300
BEANSTALKD_EXTRA="-b /var/lib/beanstalkd"

beanstalkd 任務(wù)狀態(tài)

狀態(tài)	注釋
delayed	延遲狀態(tài)
ready	準(zhǔn)備好狀態(tài)
reserved	消費(fèi)者把任務(wù)讀出來，處理時
buried	預(yù)留狀態(tài)
delete	刪除狀態(tài)

管理工具

親測了很多網(wǎng)上能找到的 beanstalkd 工具，這兩款是我最中意的了，一個命令行，一個 web 的。

命令行：https://github.com/src-d/beanstool

web 界面：https://github.com/ptrofimov/beanstalk_console

編程語言客戶端

PHP 客戶端

https://packagist.org/packages/pda/pheanstalk

composer require pda/pheanstalk

寫入 job

<?php
//創(chuàng)建隊列消息
require_once('./vendor/autoload.php');
use Pheanstalk\Pheanstalk;
$pheanstalk = new Pheanstalk('127.0.0.1',11300);
$tubeName = 'email_list';
$jobData = [
  'email' => '123456@163.com',
  'message' => 'Hello World !!',
  'dtime' => date('Y-m-d H:i:s'),
];
$pheanstalk->useTube( $tubeName)->put( json_encode( $jobData ) );

消費(fèi) job

<?php
ini_set('default_socket_timeout', 86400*7);
ini_set( 'memory_limit', '256M' );
// 消費(fèi)隊列消息
require_once('./vendor/autoload.php');
use Pheanstalk\Pheanstalk;
$pheanstalk = new Pheanstalk('127.0.0.1',11300);
$tubeName = 'email_list';
while ( true )
{
  // 獲取隊列信息, reserve 阻塞獲取
  $job = $pheanstalk->watch( $tubeName )->ignore( 'default' )->reserve();
  if ( $job !== false )
  {
    $data = $job->getData();
    /* TODO 邏輯操作 */
    /* 處理完成，刪除 job */
    $pheanstalk->delete( $job );
  }
}

default_socket_timeout 這個參數(shù)是一定要加的，php 默認(rèn)一般是 60s，假如您沒有在代碼里面設(shè)置，采用默認(rèn)的話（60s），60s 之內(nèi)如果沒有 job 產(chǎn)生，腳本就會報 socket 錯誤，我寫的是 7 天超時，您可以根據(jù)業(yè)務(wù)去調(diào)整，記住一定要配置，網(wǎng)上很多搜的 consumer 腳本都沒有配置這個，根本不能投入生產(chǎn)環(huán)境使用，這是我親自實(shí)踐的結(jié)果。

關(guān)于 while true 是否死循環(huán)，很明確告訴你是死循環(huán)，但是不會一直耗性能的那樣執(zhí)行下去，它會在 reserve 這里阻塞不動，直到有消息產(chǎn)生才會往下走，所以大可放心使用，我的項(xiàng)目代碼里面是使用了方法調(diào)用方法自身去實(shí)現(xiàn)循環(huán)的。

就是這樣的代碼，供參考：

public function watchJob()
{
  $job = $this->pheanstalk->watch( config( 'tube' ) )->ignore( 'default' )->reserve();
  if ( $job !== false )
  {
    $job_data = $job->getData();
    $this->subscribe( $job_data );
    $this->pheanstalk->delete( $job );
    /* 繼續(xù) Watch 下一個 job */
    $this->watchJob();
  }
  else
  {
    $this->log->error( 'reserve false', 'reserve false' );
  }
}

監(jiān)控 beanstalkd 狀態(tài)

<?php
//監(jiān)控服務(wù)狀態(tài)
require_once('./vendor/autoload.php');
use Pheanstalk\Pheanstalk;
$pheanstalk = new Pheanstalk('127.0.0.1',11300);
$isAlive = $pheanstalk->getConnection()->isServiceListening();
var_dump( $isAlive );

可以配合 email 做一個報警郵件，腳本每分鐘去執(zhí)行，判斷狀態(tài)是 false，就給管理員發(fā)送郵件報警。

一些相關(guān)命令

查看 beanstalkd 服務(wù)內(nèi)存占用

top -u beanstalkd

后臺運(yùn)行 consumer 腳本

nohup php googlehome_subscribe.php &

查看 consumer 腳本運(yùn)行時間

ps -A -opid,stime,etime,args | grep consumer.php

手工重啟 consumer 腳本

ps auxf|grep 'googlehome_subscribe.php'|grep -v grep|awk '{print $2}'|xargs kill -9 
nohup php googlehome_subscribe.php &

一些總結(jié)

php 要把錯誤日志打開，方便收集 consumer 腳本 crash 的 log，腳本跑出一些致命的 error 一定要及時修復(fù)，因?yàn)橐坏┯绣e就會掛掉，這會影響你腳本的可用性，后期穩(wěn)定之后可以上 supervisor 這種進(jìn)程管理程序來管控腳本生命周期。

一些網(wǎng)絡(luò)請求操作，一定要 try catch 到所有錯誤，一旦沒有 catch 到，腳本就崩。我用的是 Guzzle 去做的網(wǎng)絡(luò)請求，下面是我 catch 的一些錯誤，代碼片段供參考。

try
{
  /* TODO: 邏輯操作 */
}
catch ( ClientException $e )
{
  $results['mid']  = $this->mid;
  $results['code']  = $e->getResponse()->getStatusCode();
  $results['reason'] = $e->getResponse()->getReasonPhrase();
  $this->log->error( 'properties-changed ClientException', $results );
}
catch ( ServerException $e )
{
  $results['mid']  = $this->mid;
  $results['code']  = $e->getResponse()->getStatusCode();
  $results['reason'] = $e->getResponse()->getReasonPhrase();
  $this->log->error( 'properties-changed ServerException', $results );
}
catch ( ConnectException $e )
{
  $results['mid'] = $this->mid;
  $this->log->error( 'properties-changed ConnectException', $results );
}

job 消費(fèi)之后一定要刪除掉，如果長時間不刪除，php 客戶端會有 false 返回，是因?yàn)橛?DEADLINE_SOON 這個超時錯誤產(chǎn)生，所以處理完任務(wù)，一定要記得刪除，這一點(diǎn)跟 kafka 不一樣，beanstalkd 需要開發(fā)者自己去刪除 job。

以上就是PHP7生產(chǎn)環(huán)境隊列Beanstalkd用法詳解的詳細(xì)內(nèi)容，更多關(guān)于PHP7中Beanstalkd正確用法的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: