php - Gearman php并行任务对错误请求的响应

Question

我目前正在测试Gearman，以处理来自Web前端的并行任务/请求，Gearman cleint通过Ajax接收带有post params的请求，然后创建任务并将其发送给Gearman Worker。正在运行21个工作进程实例，以同时处理来自不同客户端的多个请求。一次只能处理一个客户端请求，但一切正常，但是当多个客户端同时请求时，客户端对于他们所请求的信息将得到错误的结果。
例如，如果客户端A请求了customer_id 123的信息，而客户端B重新询问了customer_id 456的信息，并且两个请求都同时触发，则客户端A将获得客户端B的结果，而客户端B将获得客户端A的结果。功能不同，但存在相同的问题。请帮助我在我的代码上查找问题。

我使用的是CURL-multi，没有任何问题(here)，但最近决定尝试测试Gearman，以查看是否可以获得更高的性能。

这是我为客户和工作人员编写的代码:

客户代码:

<?php

header("Expires: Mon, 26 Jul 1997 05:00:00 GMT");    // Date in the past
header("Cache-Control: no-store, no-cache, must-revalidate");  // HTTP/1.1
header("Cache-Control: post-check=0, pre-check=0", false);
header("Pragma: no-cache");


$order_id   = $_POST['order_id'];
$customer_id    = $_POST['customer_id'];

//some validation code here


class DataCollector extends GearmanClient{
    private $data     = array();
    private $tmpArr   = array();

    function addData($content){
        if($content){
            $this->tmpArr   = json_decode($content, true);
            $this->data     = array_merge($this->tmpArr, $this->data);
            //$this->data[] = json_decode($content, true);
        }
    }
    function getData(){
        return $this->data;
    }
    function outputData(){
        echo json_encode($this->getData());
    }

    function taskCompleted($task){

       $this->addData($task->data());

    }

}


$collector = new DataCollector();

$collector->addServer();

# set a function to be called when the work is complete
$collector->setCompleteCallback(array( $collector, "taskCompleted"));


//params to pass to worker
$queryStr = array(
      "order_id" => $order_id,
      "customer_id" => $customer_id
);

$postData = serialize( $queryStr );

# add tasks to be executed in parallel in Gearman server
$collector->addTask("getdata_orderDetails", $postData, null, "1");
$collector->addTask("getdata_customerDetails", $postData, null, "2");

# run the tasks in parallel
$collector->runTasks();


# output the data 
$collector->outputData();

?>

worker 代码:

<?php

class Worker{
    private $worker;
    static $conn;

public function __construct(){

    try{
        self::$conn = oci_connect($user, $pass, $db); // create db connection
    }
    catch (Exception $e) {
        echo "ERROR: " . $e->getMessage();
    }

    $this->worker = new GearmanWorker();
    $this->worker->addServer();

    # Register functions
    $this->worker->addFunction("getdata_orderDetails",    array($this, "getdata_orderDetails_fn"));
    $this->worker->addFunction("getdata_customerDetails", array($this, "getdata_customerDetails_fn"));
}

public function run(){

    while (1) {
      //print "Waiting for job...\n";
      $this->worker->work(); 
      if ($this->worker->returnCode() != GEARMAN_SUCCESS) {
        echo "return_code: " . $this->worker->returnCode() . "\n";
        break;
      }
    }



}


static function getdata_orderDetails_fn($job){


        if(!self::$conn){
        $responseArr = array(
            'response_status'           => -1,         //failed
            'response_message'          => 'Database connection lost',
            'response_id'               => 'DatabaseConnectionErr'
            );

        return  json_encode($responseArr);
        }

    $postData = unserialize($job->workload());

    $order_id   = $postData['order_id'];
    $customer_id    = $postData['customer_id'];

    $sql = "select order_id, order_status, create_date from customer_order where order_id= :order_id";
    $stmt = oci_parse(self::$conn, $sql);
    oci_bind_by_name($stmt, ":order_id", $order_id, -1);
    oci_execute($stmt);
    oci_fetch($stmt);
    $order_id   = oci_result($stmt, 'ORDER_ID');
    $order_status   = oci_result($stmt, 'ORDER_STATUS');
    $create_date    = oci_result($stmt, 'CREATE_DATE');
    oci_free_statement($stmt);


    $responseArr = array(
        'response_status'   => 1,  
        'response_message'  => 'success',
        'response_id'       => 'order_details',
        'order_id'          => $order_id,
        'order_status'      => $order_status,
        'create_date'           => $create_date
    );

    // send result
    return  json_encode($responseArr);
}

static function getdata_customerDetails_fn($job){


        if(!self::$conn){
        $responseArr = array(
            'response_status'   => -1,         //failed
            'response_message'  => 'Database connection lost',
            'response_id'       => 'DatabaseConnectionErr'
            );

        return  json_encode($responseArr);
        }

    $postData = unserialize($job->workload());

    $order_id   = $postData['order_id'];
    $customer_id    = $postData['customer_id'];

    $sql = "select customer_id, customer_fname, customer_lname, customer_address, customer_contact where customer_id= :customer_id";
    $stmt = oci_parse(self::$conn, $sql);
    oci_bind_by_name($stmt, ":customer_id", $customer_id, -1);
    oci_execute($stmt);
    oci_fetch($stmt);
    $customer_id        = oci_result($stmt, 'CUSTOMER_ID');
    $customer_fname     = oci_result($stmt, 'CUSTOMER_FNAME');
    $customer_lname     = oci_result($stmt, 'CUSTOMER_LNAME');
    $customer_address   = oci_result($stmt, 'CUSTOMER_ADDRESS');
    $customer_contact   = oci_result($stmt, 'CUSTOMER_CONTACT');
    oci_free_statement($stmt);

    $responseArr = array(
        'response_id'       => 'customer_details',
        'response_status'   => 1,  
        'response_message'  => 'success',
        'customer_id'       => $customer_id,
        'customer_fname'    => $customer_fname,
        'customer_lname'    => $customer_lname,
        'customer_address'  => $customer_address,
        'customer_contact'  => $customer_contact
    );

    // send result
    return  json_encode($responseArr);

}

}//class


 //start worker
  $worker = new Worker();
  $worker->run();


?>

Answer 1

我认为问题在于您的任务ID并非唯一。尝试为每个任务分配唯一的ID，或者根本不指定任务ID:

$collector->addTask("getdata_orderDetails", $postData);
$collector->addTask("getdata_customerDetails", $postData);

# run the tasks in parallel
$collector->runTasks();