Detailed explanation of PHP using Beanstalkd examples

The basic concepts of Beanstalkd, compilation and yum installation methods have been introduced in the previous article "Detailed Explanation of Beanstalkd Message/Task Queue". Today, I will practice the process of using Beanstalkd with PHP. The best is to use the Pheanstalk class to connect to Beanstalkd

1. Use Composer to install Pheanstalk

composer require pda/pheanstalk

2. Implement the code

php View the beanstalkd status script Status.php

<?php
/**
 * Created by PhpStorm.
 * User: jmsite.cn
 * Date: 2019/1/21
 * Time: 10:32
 */
require "../vendor/autoload.php";
use Pheanstalk\Pheanstalk;
$pheanstalk = new Pheanstalk(&#39;192.168.75.135&#39;,11300);
print_r($pheanstalk->stats());

Producer code Producter.php

<?php
/**
 * Created by PhpStorm.
 * User: jmsite.cn
 * Date: 2019/1/20
 * Time: 16:30
 */
require "../vendor/autoload.php";
use Pheanstalk\Pheanstalk;
$pheanstalk = new Pheanstalk(&#39;192.168.75.135&#39;,11300);
for ($i=0;$i<50;$i++){
    $data = array(
        &#39;key&#39; => &#39;testkey&#39;.$i,
        &#39;value&#39; => &#39;testvalue&#39;,
        &#39;time&#39; => time(),
    );
    $ret = $pheanstalk->putInTube(&#39;test-tube&#39;, json_encode($data), Pheanstalk::DEFAULT_PRIORITY, Pheanstalk::DEFAULT_DELAY, Pheanstalk::DEFAULT_TTR);
    var_dump($ret);
}

Consumer code Consumer.php

<?php
/**
 * Created by PhpStorm.
 * User: jmsite.cn
 * Date: 2019/1/20
 * Time: 16:31
 */
set_time_limit(0);
ini_set(&#39;default_socket_timeout&#39;, 900);
require "../vendor/autoload.php";
use Pheanstalk\Pheanstalk;
$pheanstalk = new Pheanstalk(&#39;192.168.75.135&#39;,11300);
while (true){
    $job = $pheanstalk
        ->watch(&#39;test-tube&#39;)
        ->ignore(&#39;default&#39;)
        ->reserve();
    if ($job){
        sleep(2);
        echo $job->getData();
        echo "\n";
        $pheanstalk->delete($job);
    }
}

Open the command line/terminal window and execute the producer , 50 tasks will be written to the tube

PS E:\repository\work\beanstalk> php .\Producter.php
int(101)
int(102)
int(103)
int(104)
int(105)
int(106)
int(107)
int(108)
int(109)
int(110)
int(111)
int(112)
int(113)
int(114)
......

It can be seen that after $pheanstalk->putInTube succeeds, the ID of the job is returned

View status

PS E:\repository\work\beanstalk> php Status.php
Pheanstalk\Response\ArrayResponse Object
(
    [_name:Pheanstalk\Response\ArrayResponse:private] => OK
    [storage:ArrayObject:private] => Array
        (
            [current-jobs-urgent] => 0
            [current-jobs-ready] => 50
            [current-jobs-reserved] => 0
            [current-jobs-delayed] => 0
            [current-jobs-buried] => 0
            ......

Result It shows that there are 50 jobs in the ready to be read state

Open two or more command line/terminal windows, execute consumers, and simulate multi-consumer competition

Consumer 1

PS E:\repository\work\beanstalk> php .\Consumer.php
{"key":"testkey0","value":"testvalue","time":1548039103}
{"key":"testkey1","value":"testvalue","time":1548039103}
{"key":"testkey2","value":"testvalue","time":1548039103}
{"key":"testkey4","value":"testvalue","time":1548039103}
{"key":"testkey6","value":"testvalue","time":1548039103}
{"key":"testkey8","value":"testvalue","time":1548039103}
{"key":"testkey10","value":"testvalue","time":1548039103}
{"key":"testkey12","value":"testvalue","time":1548039103}
{"key":"testkey14","value":"testvalue","time":1548039103}
{"key":"testkey16","value":"testvalue","time":1548039103}
{"key":"testkey18","value":"testvalue","time":1548039103}
{"key":"testkey20","value":"testvalue","time":1548039103}
{"key":"testkey22","value":"testvalue","time":1548039103}
{"key":"testkey24","value":"testvalue","time":1548039103}
{"key":"testkey26","value":"testvalue","time":1548039103}
{"key":"testkey28","value":"testvalue","time":1548039103}
{"key":"testkey30","value":"testvalue","time":1548039103}
{"key":"testkey32","value":"testvalue","time":1548039103}
{"key":"testkey34","value":"testvalue","time":1548039103}
{"key":"testkey36","value":"testvalue","time":1548039103}
{"key":"testkey38","value":"testvalue","time":1548039103}
{"key":"testkey40","value":"testvalue","time":1548039103}
{"key":"testkey42","value":"testvalue","time":1548039103}
{"key":"testkey44","value":"testvalue","time":1548039103}
{"key":"testkey46","value":"testvalue","time":1548039103}
{"key":"testkey48","value":"testvalue","time":1548039103}

Consumer 2

PS E:\repository\work\beanstalk> php .\Consumer.php
{"key":"testkey3","value":"testvalue","time":1548039103}
{"key":"testkey5","value":"testvalue","time":1548039103}
{"key":"testkey7","value":"testvalue","time":1548039103}
{"key":"testkey9","value":"testvalue","time":1548039103}
{"key":"testkey11","value":"testvalue","time":1548039103}
{"key":"testkey13","value":"testvalue","time":1548039103}
{"key":"testkey15","value":"testvalue","time":1548039103}
{"key":"testkey17","value":"testvalue","time":1548039103}
{"key":"testkey19","value":"testvalue","time":1548039103}
{"key":"testkey21","value":"testvalue","time":1548039103}
{"key":"testkey23","value":"testvalue","time":1548039103}
{"key":"testkey25","value":"testvalue","time":1548039103}
{"key":"testkey27","value":"testvalue","time":1548039103}
{"key":"testkey29","value":"testvalue","time":1548039103}
{"key":"testkey31","value":"testvalue","time":1548039103}
{"key":"testkey33","value":"testvalue","time":1548039103}
{"key":"testkey35","value":"testvalue","time":1548039103}
{"key":"testkey37","value":"testvalue","time":1548039103}
{"key":"testkey39","value":"testvalue","time":1548039103}
{"key":"testkey41","value":"testvalue","time":1548039103}
{"key":"testkey43","value":"testvalue","time":1548039103}
{"key":"testkey45","value":"testvalue","time":1548039103}
{"key":"testkey47","value":"testvalue","time":1548039103}
{"key":"testkey49","value":"testvalue","time":1548039103}

The two consumers competed to complete all tasks. Since my beanstalkd started with binlog persistence enabled, the tasks will not be lost after beanstalkd restarts

3. Things to note

1. When creating a job, the set timeout Pheanstalk::DEFAULT_TTR must be longer than the time it takes for the consumer to process a job, otherwise the job will After the timeout, the tube will be changed to the ready state and be obtained by other consumers. At this time, the current consumer is still processing the job, which leads to the terrible phenomenon of a job being repeatedly executed by multiple consumers.

2. The maintainer of Pheanstalk has changed. In the new version of Pheanstalk, long connections are not supported. When the time for the client socket to connect to the server exceeds the default_socket_timeout set in php.ini, if the job cannot be obtained from the server tube, the connection will be disconnected, so the consumer process needs to be maintained so that the process can be restarted after exiting. It is recommended to use supervisord to maintain the consumer process.

Code to determine socket timeout

public function getLine($length = null)
    {
        $timeout = ini_get('default_socket_timeout');
        $timer   = microtime(true);
        do {
            $data = isset($length) ?
                $this->_wrapper()->fgets($this->_socket, $length) :
                $this->_wrapper()->fgets($this->_socket);
            if ($this->_wrapper()->feof($this->_socket)) {
                throw new Exception\SocketException('Socket closed by server!');
            }
            if (($data === false) && microtime(true) - $timer > $timeout) {
                $this->disconnect();
                throw new Exception\SocketException('Socket timed out!');
            }
        } while ($data === false);
        return rtrim($data);
    }

The above is the detailed content of Detailed explanation of PHP using Beanstalkd examples. For more information, please follow other related articles on the PHP Chinese website!