首頁 後端開發 php教程 转载 kestrel php 讯息队列

转载 kestrel php 讯息队列

Jun 13, 2016 am 11:00 AM
queue return the this

转载 kestrel php 消息队列

We've been using Twitter's kestrel queue server for a while now at work, but only from our service layer, which is written in python.? Now that we have some queueing needs from our application layer, written in PHP, I spent a few days this week adding queue support to our web application.? I thought I'd share what I learned, and how I implemented it.

Goals

The kestrel server itself was pretty straightforward to get up and running.? The only thing I would point out is that I recommend sticking to release branches, as master was fairly unstable when I tried to use it.? Regarding implementing the client, there were a few goals I had in mind when I started:

??? Since kestrel is built on the memcache protocol, try and leverage an existing memcache client rather than build one from scratch
??? Utilize our existing batch job infrastructure, which I covered previously here, and make sure our multi-tenant needs are met
??? Keep the queue interface generic in case we change queue servers later
??? Utilize existing kestrel management tools, only build out the the functionality we need

With these goals in mind, I ended up with 4 components: a kestrel client, a producer, a consumer, and a very small CLI harness for running the consumer.? But before I even coded anything, I set up kestrel web, a web UI for kestrel written by my co-worker Matt Erkkila.? Kestrel web allows you to view statistics on kestrel, manage queues, as well as sort and filter queues based on manual inputs.? Having this tool up and running from the get go made it easy to watch jobs get added and consumed from my test queue, and also easily flush out the queues as needed.

The Kestrel Client

I couldn't find any existing kestrel clients for PHP, so I started looking at the two memcache extensions: the older memcache, and Andrei Zmievski's memcached, the latter of which is based on the libmemcached library.? I started with memcache, and while it worked fine initially, I quickly found that I could not modify timeouts.? This interfered with the way? kestrel recommends you poll it for new jobs, and I would see timeout errors from the memcache extension if you tried to set the poll timeout to 1 second or higher (the memcache default).? The memcached extension does not have these issues, so I went with it.

The first gotcha I ran into was serialization.? You can use memcached's serializer for writing to kestrel, but when it reads the data back, it doesn't recognize that it is serialized.? So I just serialize the data manually in my client, and things work fine. One other thing to note is that you'll want to disable compression, or do it manually, as the memcached extension will automatically compress anything over 100 bytes by default, and will not decompress it when reading from kestrel.

The other issue is that if you want to use any custom kestrel commands, you can't.? Since the application layer doesn't need anything fancy, the memcached extension will work fine for it.? Once we need support for the upcoming monitor (batching) in kestrel 2, we may need to implement a kestrel client from scratch.? Kestrel web supplies everything else we need right now.

Once the decision was made to use memcached, I wrote a light decorator for it, EC_KestrelClient.? This handles instantiation of the memcached client, serialization, and helpers for some kestrel specific options to the GET command.? It also has support for passing memcached specific options through it.? The class ended up looking like this:

?

<?php /** * A thin kestrel client that wraps Memcached (libmemcached extension) * * @author Bill Shupp <[email&#160;protected]> * @copyright 2010-2011 Empower Campaigns */ class EC_KestrelClient {     /** * The Memcached instance * * @var Memcached */     protected $_memcached = null;     /** * The Kestrel server IP * * @var string */     protected $_host = '127.0.0.1';     /** * The Kestrel server port * * @var string */     protected $_port = 22133;     /** * Optional options, not currently used * * @var array */     protected $_options = array();     /** * Sets the host, port, and options to be used * * @param string $host The host to use, defaults to 127.0.0.1 * @param int $port The port to use, defaults to 22133 * @param array $options Memcached options, not currently used * * @return void */     public function __construct(         $host = '127.0.0.1', $port = 22133, array $options = array()     )     {         $this->_host = $host;         $this->_port = $port;         $this->setOptions($options);     }     /** * Sets job data on the queue, json_encoding the value to avoid problematic * serialization. * * @param string $queue The queue name * @param mixed $data The data to store * * @return bool */     public function set($queue, $data)     {         // Local json serialization, as kestrel doesn't send serialization flags         return $this->getMemcached()->set($queue, json_encode($data));     }     /** * Reliably read an item off of the queue. Meant to be run in a loop, and * call closeReliableRead() when done to make sure the final job is not left * on the queue. * * @param mixed $queue The queue name to read from * @param int $timeout The timeout to wait for a job to appear * * @return array|false * @see closeReliableRead() */     public function reliableRead($queue, $timeout = 1000)     {         $queue = $queue . '/close/open/t=' . $timeout;         $result = $this->getMemcached()->get($queue);         if ($result === false) {             return $result;         }         // Local json serialization, as kestrel doesn't send serialization flags         return json_decode($result, true);     }     /** * Closes any existing open read * * @param string $queue The queue name * * @return false */     public function closeReliableRead($queue)     {         $queue = $queue . '/close';         return $this->getMemcached()->get($queue);     }     /** * Aborts an existing reliable read * * @param string $queue The queue name * * @return false */     public function abortReliableRead($queue)     {         $queue = $queue . '/abort';         return $this->getMemcached()->get($queue);     }     /** * Set an option to be used with the Memcached client. Not used. * * @param string $name The option name * @param value $value The option value * * @return void */     public function setOption($name, $value)     {         $this->_options[$name] = $value;     }     /** * Sets multiple options * * @param array $options Array of key/values to set * * @return void */     public function setOptions(array $options)     {         foreach ($options as $name => $value) {             $this->setOption($name, $value);         }     }     /** * Gets a current option's value * * @param string $name The option name * * @return mixed */     public function getOption($name)     {         if (isset($this->_options[$name])) {             return $this->_options[$name];         }         return null;     }     /** * Gets all current options * * @return array */     public function getOptions()     {         return $this->_options;     }     /** * Gets a singleton instance of the Memcached client * * @return Memcached */     public function getMemcached()     {         if ($this->_memcached === null) {             $this->_initMemcached();         }         return $this->_memcached;     }     /** * Initialized the Memcached client instance * * @return void */     protected function _initMemcached()     {         $this->_memcached = $this->_getMemcachedInstance();         foreach ($this->_options as $option => $value) {             $this->_memcached->setOption($option, $value);         }         $this->_memcached->addServer($this->_host, $this->_port);         $this->_memcached->setOption(Memcached::OPT_COMPRESSION, false);     }     // @codeCoverageIgnoreStart     /** * Returns a new instance of Memcached. Abstracted for testing. * * @return Memcached */     protected function _getMemcachedInstance()     {         return new Memcached();     }     // @codeCoverageIgnoreEnd } 
登入後複製

?

?

view raw EC_KestrelClient.php This Gist brought to you by GitHub.

The Producer

The producer is very simple.? It just formats the data into a standard structure, including current tenant information, namespaces the queue so it doesn't collide with other projects, and adds it to the queue.? The producer looks like this:

?

<?php /** * Interface for adding jobs to a queue server * * @author Bill Shupp <[email&#160;protected]> * @copyright 2010-2011 Empower Campaigns */ class EC_Producer {     /** * Adds a job onto a queue * * @param string $queue The queue name to add a job to * @param string $jobName The job name for the consumer to run * @param mixed $data Optional additional data to pass to the job * * @return bool */     public function addJob($queue, $jobName, $data = null)     {         $item = array(             'instance' => EC::getCurrentInstanceName(),             'jobName' => $jobName         );         if ($data !== null) {             $item['data'] = $data;         }         // Namespace queue with project         $queue = 'enterprise_' . $queue;         $client = $this->_getKestrelClient();         return $client->set($queue, $item);     }     // @codeCoverageIgnoreStart     /** * Gets a single instance of EC_KestrelClient. Abstracted for testing. * * @return void */     protected function _getKestrelClient()     {         if (APPLICATION_ENV === 'testing') {             throw new Exception(__METHOD__ . ' was not mocked when testing');         }         static $client = null;         if ($client === null) {             $host = EC::getConfigOption('kestrel.host');             $port = EC::getConfigOption('kestrel.port');             $client = new EC_KestrelClient($host, $port);         }         return $client;     }     // @codeCoverageIgnoreEnd } 
登入後複製

?

?

?

view raw EC_Producer.php This Gist brought to you by GitHub.

The Consumer

The consumer has a bit more to it, though still pretty straightforward.? It's intended to be run from a monitoring tool like daemontools or supervisord, so there is a very small CLI harness that just passes the CLI arguments into EC_Consumer and runs it.? After parsing the CLI arguments, EC_Consumer polls kestrel for new jobs, and runs them through our standard batch job infrastructure.? Until we have more confidence in PHP's long running process ability, I added an optional maxium jobs argument, which will stop the consumer from processing more than X jobs and then terminate.? The monitoring service (supervisord) will then just restart it in a matter of seconds.? I also added an optional debug argument for testing, so you can see every action as it happens.? The CLI harness looks like this:

?

#!/bin/env php <?php // External application bootstrapping require_once __DIR__ . '/cli_init.php'; // Instantiate and run the consumer $consumer = new EC_Consumer($argv); $consumer->run(); 
登入後複製

?

view raw consumer_cli.php This Gist brought to you by GitHub.

And the main consumer class, EC_Consumer, looks something like this:

<?php /** * Enterprise queue consumer interface, called by bin/consumer_cli.php * * @author Bill Shupp <[email&#160;protected]> * @copyright 2010-2011 Empower Campaigns */ class EC_Consumer {     /** * Instance of [email&#160;protected] Zend_Console_Getopt} * * @var Zend_Console_Getopt */     protected $_opt = null;     /** * Which APPLICATION_ENV to run under (see -e) * * @var string */     protected $_environment = null;     /** * The kestrel server IP * * @var string */     protected $_host = null;     /** * The kestrel server port * * @var int */     protected $_port = null;     /** * The kestrel queue name to connect to * * @var string */     protected $_queue = null;     /** * Whether we should show debug output * * @var bool */     protected $_debug = false;     /** * Maximum # of jobs for this process to perform (for memory fail safe) * * @var int */     protected $_maxJobs = null;     /** * Current job count * * @var int */     protected $_jobCount = 0;     /** * Parses arguments from the command line and does error handling * * @param array $argv The $argv from bin/ecli.php * * @throw Zend_Console_Getopt_Exception on failure * @return void */     public function __construct(array $argv)     {         try {             $opt = new Zend_Console_Getopt(                 array(                     'environment|e=s' => 'environment name (e.g. development)'                                          . ', required',                     'server|s=s' => 'kestrel server, format of host:port'                                          . ', required',                     'queue|q=s' => 'queue name (e.g. crawler_campaign)'                                          . ', required',                     'max-jobs|m=s' => 'max jobs to run before exiting'                                          . ', optional',                     'debug|d' => 'show debug output'                                          . ', optional',                 )             );             $opt->setArguments($argv);             $opt->parse();             // Set environment             if ($opt->e === null) {                 throw new Zend_Console_Getopt_Exception(                     'Error: missing environment'                 );             }             $this->_environment = $opt->e;             // @codeCoverageIgnoreStart             if (!defined('APPLICATION_ENV')) {                 define('APPLICATION_ENV', $this->_environment);             }             // @codeCoverageIgnoreEnd             // Set server             if ($opt->s === null) {                 throw new Zend_Console_Getopt_Exception(                     'Error: missing server'                 );             }             $parts = explode(':', $opt->s);             if (count($parts) !== 2) {                 throw new Zend_Console_Getopt_Exception(                     'Error: invalid server: ' . $opt->s                 );             }             $this->_host = $parts[0];             $this->_port = $parts[1];             // Set queue             if ($opt->q === null) {                 throw new Zend_Console_Getopt_Exception(                     'Error: missing queue'                 );             }             $this->_queue = $opt->q;             // Set max-jobs             if ($opt->m !== null) {                 $this->_maxJobs = $opt->m;             }             // Set debug             if ($opt->d !== null) {                 $this->_debug = true;             }         } catch (Zend_Console_Getopt_Exception $e) {             echo "\n" . $e->getMessage() . "\n\n";             echo $opt->getUsageMessage();             // @codeCoverageIgnoreStart             if (!defined('APPLICATION_ENV') || APPLICATION_ENV !== 'testing') {                 exit(1);             }             // @codeCoverageIgnoreEnd         }         $this->_opt = $opt;     }     /** * Polls the queue server for jobs and runs them as they come in * * @return void */     public function run()     {         $client = $this->_getKestrelClient();         $queue = 'enterprise_' . $this->_queue;         while ($this->_keepRunning()) {             // Pull job from queue             $job = $client->reliableRead($queue, 500);             if ($job === false) {                 $this->_debug('Nothing on queue ' . $queue);                 continue;             }             if (!isset($job['instance'])) {                 echo 'Instance not set in queue job: ' . print_r($job, true);                 continue;             }             $instance = $job['instance'];             if (!isset($job['jobName'])) {                 echo 'Job name not set in queue job: ' . print_r($job, true);                 continue;             }             $jobName = $job['jobName'];             $data = null;             if (isset($job['data'])) {                 $data = $job['data'];             }             // Run the job             $returnCode = $this->runJob($instance, $jobName, $data);             if ($returnCode !== 0) {                 $client->abortReliableRead($queue);                 continue;             }         }         $client->closeReliableRead($queue);     }     /** * Runs the job via bin/ecli.php * * @param string $instance The instance name to run the job under * @param string $jobName The job name * @param string $data Optional extra data * * @return int */     public function runJob($instance, $jobName, $data)     {         $cmd = BASE_PATH . '/bin/ecli.php '             . '-e ' . $this->_environment             . ' -i ' . $instance             . ' -j ' . $jobName;         if ($data) {             $cmd .= " '" . base64_encode(json_encode($data)) . "'";         }         $returnCode = $this->_passthru($cmd);         $this->_jobCount++;         $this->_debug('Job count: ' . $this->_jobCount);         return $returnCode;     }     /** * Check to see if the job limit has been reached * * @return bool */     protected function _keepRunning()     {         return ($this->_maxJobs === null) ? true                : ($this->_jobCount < $this->_maxJobs);     }     /** * Show debug messages * * @param mixed $message * * @return void */     protected function _debug($message)     {         if (!$this->_debug) {             return;         }         echo $message . "\n";     }     // @codeCoverageIgnoreStart     /** * Calls the passthru() function and returns the exit code. Abstracted * for testing. * * @param string $cmd The command to execute * * @return int */     protected function _passthru($cmd)     {         passthru($cmd, $returnCode);         return $returnCode;     }     /** * Gets a single instance of EC_KestrelClient. Abstracted for testing. * * @return void */     protected function _getKestrelClient()     {         if (APPLICATION_ENV === 'testing') {             throw new Exception(__METHOD__ . ' was not mocked when testing');         }         return new EC_KestrelClient($this->_host, $this->_port);     }     // @codeCoverageIgnoreEnd } 
登入後複製

?

?

view raw EC_Consumer.php This Gist brought to you by GitHub.

Putting it together

Now that all the pieces are put together, let's take a look at in action. Adding example job "HelloWorld" to the queue "hello_world" from within our application looks something like this:

<?php $producer = new EC_Producer(); $producer->addJob('hello_world', 'HelloWorld', array('foo' => 'bar')); ?> view raw gistfile1.php This Gist brought to you by GitHub. 
登入後複製

?

?

And finally, here's an example of running the consumer from the CLI harness, along with some example debug output of processing the job:

./bin/consumer_cli.php -e development -s 127.0.0.1:22133 -q hello_world -d -m 2
Nothing on queue enterprise_hello_world
Nothing on queue enterprise_hello_world
Nothing on queue enterprise_hello_world
Nothing on queue enterprise_hello_world
Running EC_Job_HelloWorld on instance dev under environment development
Hello, world! Here is my data array:
stdClass Object
(
??? [foo] => bar
)
And here are my args: ./bin/ecli.php eyJmb28iOiJiYXIifQ==
Completed job in 0 seconds.
Job count: 1
Nothing on queue enterprise_hello_world
Nothing on queue enterprise_hello_world
Nothing on queue enterprise_hello_world
Nothing on queue enterprise_hello_world
Running EC_Job_HelloWorld on instance dev under environment development
Hello, world! Here is my data array:
stdClass Object
(
??? [foo] => bar
)
And here are my args: ./bin/ecli.php eyJmb28iOiJiYXIifQ==
Completed job in 0 seconds.
Job count: 2

view raw example.txt This Gist brought to you by GitHub.

That's it! I'd be interested to hear how other folks are interfacing with kestrel from PHP.

本網站聲明
本文內容由網友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容,請聯絡admin@php.cn

熱AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智慧驅動的應用程序,用於創建逼真的裸體照片

AI Clothes Remover

AI Clothes Remover

用於從照片中去除衣服的線上人工智慧工具。

Undress AI Tool

Undress AI Tool

免費脫衣圖片

Clothoff.io

Clothoff.io

AI脫衣器

Video Face Swap

Video Face Swap

使用我們完全免費的人工智慧換臉工具,輕鬆在任何影片中換臉!

熱工具

記事本++7.3.1

記事本++7.3.1

好用且免費的程式碼編輯器

SublimeText3漢化版

SublimeText3漢化版

中文版,非常好用

禪工作室 13.0.1

禪工作室 13.0.1

強大的PHP整合開發環境

Dreamweaver CS6

Dreamweaver CS6

視覺化網頁開發工具

SublimeText3 Mac版

SublimeText3 Mac版

神級程式碼編輯軟體(SublimeText3)

2 個月不見,人形機器人 Walker S 會摺衣服了 2 個月不見,人形機器人 Walker S 會摺衣服了 Apr 03, 2024 am 08:01 AM

機器之能報道編輯:吳昕國內版的人形機器人+大模型組隊,首次完成疊衣服這類複雜柔性材料的操作任務。隨著融合了OpenAI多模態大模型的Figure01揭開神秘面紗,國內同行的相關進展一直備受關注。就在昨天,國內"人形機器人第一股"優必選發布了人形機器人WalkerS深入融合百度文心大模型後的首個Demo,展示了一些有趣的新功能。現在,得到百度文心大模型能力加持的WalkerS是這個樣子的。和Figure01一樣,WalkerS沒有走動,而是站在桌子後面完成一系列任務。它可以聽從人類的命令,折疊衣物

Laravel開發:如何使用Laravel Queue處理非同步任務? Laravel開發:如何使用Laravel Queue處理非同步任務? Jun 13, 2023 pm 08:32 PM

隨著應用程式變得越來越複雜,處理和管理大量資料和流程是一個挑戰。為了處理這種情況,Laravel為使用者提供了一個非常強大的工具,即Laravel隊列(Queue)。它允許開發人員在後台運行諸如發送電子郵件,生成PDF,處理影像剪裁等任務,而不會對使用者介面產生任何影響。在這篇文章中,我們將深入研究如何使用Laravel隊列。什麼是LaravelQueue隊列

C語言return的用法詳解 C語言return的用法詳解 Oct 07, 2023 am 10:58 AM

C語言return的用法有:1、對於傳回值類型為void的函數,可以使用return語句來提前結束函數的執行;2、對於傳回值型別不為void的函數,return語句的作用是將函數的執行結果傳回給呼叫者;3、提前結束函數的執行,在函數內部,我們可以使用return語句來提前結束函數的執行,即使函數並沒有回傳值。

Java中return和finally語句的執行順序是怎樣的? Java中return和finally語句的執行順序是怎樣的? Apr 25, 2023 pm 07:55 PM

原始碼:publicclassReturnFinallyDemo{publicstaticvoidmain(String[]args){System.out.println(case1());}publicstaticintcase1(){intx;try{x=1;returnx;}finally{x=3;}}#輸出上述程式碼的輸出可以簡單地得出結論:return在finally之前執行,我們來看下字節碼層面上發生了什麼事情。下面截取case1方法的部分字節碼,並且對照源碼,將每個指令的含義註釋在

多執行緒環境下Java Queue佇列的安全性問題及解決方案 多執行緒環境下Java Queue佇列的安全性問題及解決方案 Jan 13, 2024 pm 03:04 PM

JavaQueue佇列在多執行緒環境下的安全性問題與解決方案引言:在多執行緒程式設計中,程式中的共享資源可能面臨競爭條件,這可能導致資料的不一致性或錯誤。在Java中,Queue佇列是一種常用的資料結構,在多個執行緒同時操作佇列的情況下,就存在安全性問題。本文將討論JavaQueue佇列在多執行緒環境下的安全性問題,並介紹幾個解決方案,重點以程式碼範例的方式解釋。一

Vue3怎麼使用setup語法糖拒絕寫return Vue3怎麼使用setup語法糖拒絕寫return May 12, 2023 pm 06:34 PM

Vue3.2setup語法糖是在單文件組件(SFC)中使用組合式API的編譯時語法糖解決Vue3.0中setup需要繁瑣將聲明的變量、函數以及import引入的內容通過return向外暴露,才能在使用的問題1.在使用中無需return宣告的變數、函數以及import引入的內容,即可在使用語法糖//import引入的內容import{getToday}from'./utils'//變數constmsg='Hello !'//函數func

聊聊Vue2為什麼能透過this存取各種選項中屬性 聊聊Vue2為什麼能透過this存取各種選項中屬性 Dec 08, 2022 pm 08:22 PM

這篇文章帶大家解讀vue原始碼,來介紹一下Vue2中為什麼可以使用 this 存取各種選項中的屬性,希望對大家有幫助!

Queue在Java中的應用 Queue在Java中的應用 Feb 18, 2024 pm 03:52 PM

Java中Queue的用法在Java中,Queue(佇列)是一種常用的資料結構,它遵循先進先出(FIFO)原則。 Queue可用於實作訊息佇列、任務排程等場景,能夠很好地管理資料的排列和處理順序。本文將介紹Queue的用法,並提供具體的程式碼範例。 Queue的定義和常用方法在Java中,Queue是JavaCollectionsFramework中的一個介面

See all articles