asynchronous functionality Magento

Question

I want to add asynchronous functionality into magento. I am using events to do run some backend processes and these processes can sometime cause long delays. This under some circumstances become unacceptable especially when customer is trying to place an order. Now I want observer to run independently of event. I tried using mage-resque to use queues but it adds one more point of failure and there is no way to know if it is widely used. Are there any job queues that can easily be integrated with magento which are actively updated and don't overuse database? I have tried using magento SNS to send customer SMS but again I don't know if there is a precedent. The truth is I wouldn't want to experiment at this stage without knowing consequence. Any help would be appreciated

Answer 1

There is no 'standard' approach to queuing within Magento per se. As you've probably seen there's a number of modules floating around that try and help with different messaging systems, for instance Lilmuckers_Queue. My guess is that only a small portion Magento projects end up needing some slow process sent to a job queue. Any performance gain for something like offloading an e-mail during checkout isn't worth the effort of the implementation and maintenance.

One thing to keep in mind is that its possible to have a message queue that is "highly available" thus not a single point of failure. If you have the time/resources it's possible to configure Redis (and other MQ servers) for failover. If you don't mind being "locked in" to AWS then using SQS would be a good option, especially if you're already using some of their services.

Which route one ends up going ultimately depends on the project, the technical team and organization. An internal engineering team of a large merchant would likely have the proficiency and staff required to implement such a solution. A freelancer or agency implementing a project for a small-ish client probably shouldn't over complicate the architecture (especially for the next team/person who ends up with the project).

Answer 2

Use Magento's cron functionality. It works for sure and it's async. If you have time and resources, implement Beanstalkd.

Answer 3

Personally I try and keep as much dependencies away from large job processing, I've used this same class in the past for very large data parsing processes that are heavy CPU/IO bound. Chunk the data into subsets and offsets, to utilize the number of CPU cores available since PHP is single threaded.

Note this may not work well on Windows hosting, an alternative if on Windows.

<?php
class Thread {
    var $pref ; // process reference
    var $pipes; // stdio
    var $buffer; // output buffer
    var $output;
    var $error;
    var $timeout;
    var $start_time;


    function Thread() {
        $this->pref = 0;
        $this->buffer = "";
        $this->pipes = (array)NULL;
        $this->output = "";
        $this->error="";

        $this->start_time = time();
        $this->timeout = 0;
    }

    function Create ($command) {
        $t = new Thread;
        $descriptor = array (0 => array ("pipe", "r"), 1 => array ("pipe", "w"), 2 => array ("pipe", "w"));
        //Open the resource to execute $command
        $t->pref = proc_open($command,$descriptor,$t->pipes);
        //Set STDOUT and STDERR to non-blocking 
        stream_set_blocking ($t->pipes[1], 0);
        stream_set_blocking ($t->pipes[2], 0);
        return $t;
    }

    //See if the command is still active
    function isActive () {
        $this->buffer .= $this->listen();
        $f = stream_get_meta_data ($this->pipes[1]);
        return !$f["eof"];
    }

    //Close the process
    function close () {
        $r = proc_close ($this->pref);
        $this->pref = NULL;
        return $r;
    }

    //Send a message to the command running
    function tell ($thought) {
        fwrite ($this->pipes[0], $thought);
    }

    //Get the command output produced so far
    function listen () {
        $buffer = $this->buffer;
        $this->buffer = "";
        while ($r = fgets ($this->pipes[1], 1024)) {
            $buffer .= $r;
            $this->output.=$r;
        }
        return $buffer;
    }

    //Get the status of the current runing process
    function getStatus(){
        return proc_get_status($this->pref);
    }

    //See if the command is taking too long to run (more than $this->timeout seconds)
    function isBusy(){
        return ($this->start_time>0) && ($this->start_time+$this->timeout<time());
    }

    //What command wrote to STDERR
    function getError () {
        $buffer = "";
        while ($r = fgets ($this->pipes[2], 1024)) {
            $buffer .= $r;
        }
        return $buffer;
    }
}


//Wrapper for Thread class
class Multithread{
    var $output;
    var $error;
    var $thread;
    var $commands = array();

    function __construct($commands){
        $this->commands = $commands;

        foreach ($this->commands as $key=>$command){
            $this->thread[$key]=Thread::create($command);
        }
    }


    function run(){
        $commands = $this->commands;
        //Cycle through commands
        while (count($commands)>0){
            foreach ($commands as $key=>$command){
                //Get the output and the errors
                $this->output[$key].=$this->thread[$key]->listen();
                $this->error[$key].=$this->thread[$key]->getError();
                //Check if command is still active
                if ($this->thread[$key]->isActive()){
                    $this->output[$key].=$this->thread[$key]->listen();
                    //Check if command is busy
                    if ($this->thread[$key]->isBusy()){
                        $this->thread[$key]->close();
                        unset($commands[$key]);
                    }
                } else {
                    //Close the command and free resources
                    $this->thread[$key]->close();
                    unset($commands[$key]);
                }
            }
        }
        return $this->output;
    }
}
?>

Example PHP file using above class:

<?php
set_time_limit(0);
include "threads.php";
$commands = array('ffmpeg -i '.$inputFile[0].' '.$outputFile[0].' 2>&1','ffmpeg -i '.$inputFile[0].' '.$outputFile[0].' 2>&1');
$threads = new Multithread($commands);
$threads->run();
foreach ($threads->commands as $key=>$command){
    echo "Command ".$command.":<br>";
    echo "Output ".$threads->output[$key]."<br>";
    echo "Error ".$threads->error[$key]."<br><br>";
}
?>

Source:

• http://www.php-code.net/2010/05/running-multiple-processes-in-php/

Also worth mentioning:

• http://mirasvit.com/magento-extensions/magento-asynchronous-reindex.html

Answer 4

There is likely a better way to do this then handling this processing at time of order placement. I would suggest saving the order with some kind of flag, maybe something as simple as setting the order status to "Processing required", "Stage 1", etc. Some kind of status that makes it obvious that the order needs additional logic before it's "complete". Then you have a cron job that looks for orders with this status/state and run your processing logic on the order. After the processing logic is complete, you can change the order status to something signifying that step has been completed, etc. Essentially the order itself would become the "queue" based on the status of the order. No need for 3rd party queues, etc. You can then basically move the order through your different processes until it reaches a "complete" status, etc.

Answer 5

This is quite an old post but still to help people with Magento 1, Magento 2 & PHP, here are few answers.

If you are using Magento 1 then running code asynchronously can be easily done by setting up a cron job or managing your code via queues just like message queues which in turn would require cron setup.

If you are using Magento 2 then it depends which version of Magento 2 you are using,

1. For versions older than 2.3.3 you can refer https://devdocs.magento.com/guides/v2.4/extension-dev-guide/message-queues/async-message-queue-config-files.html
2. For versions after 2.3.3 you can refer https://devdocs.magento.com/guides/v2.4/rest/asynchronous-web-endpoints.html and https://devdocs.magento.com/guides/v2.4/extension-dev-guide/async-operations.html

If you want to go with PHP or none of the above references work for you or you feel like you need a easier way out then you can refer https://www.php.net/manual/en/function.shell-exec.php#118495

Example:

<?php
    
namespace Vendor\Module\Controller\ControllerNameFolder;

class YourCustomAsyncAction extends \Magento\Framework\App\Action\Action
{
    /**
     * resultJsonFactory
     *
     * @var \Magento\Framework\Controller\Result\JsonFactory
     */
    protected $resultJsonFactory;    
    /**
     * _urlInterface
     *
     * @var \Magento\Framework\UrlInterface
     */
    protected $_urlInterface;
        
    /**
     * __construct
     *
     * @param \Magento\Framework\App\Action\Context $context
     * @param \Magento\Framework\Controller\Result\JsonFactory $resultJsonFactory
     * @return void
     */
    public function __construct(
        \Magento\Framework\App\Action\Context $context,
        \Magento\Framework\Controller\Result\JsonFactory $resultJsonFactory,
        \Magento\Framework\UrlInterface $urlInterface
    )
    {
        parent::__construct($context);
        $this->resultJsonFactory = $resultJsonFactory;    
        $this->_urlInterface = $urlInterface;
        $this->location = $location;
    }
    
    public function execute()
    {
        $result = $this->resultJsonFactory->create();
        $params = $this->getRequest()->getParams();
        /* prepare your URL */    
        $url = $this->_urlInterface->getUrl('module_route/controllername/actionname',$params);
        /* prepare your URL */

        /* async code */
        shell_exec("wget $url>/dev/null >/dev/null &");
        /* async code */
        
        return $result->setData(['success'=>true]);
    }
}