Using the beanstalkd queue server to distribute our mailing list

In our system ( Shopozz.RU shopping service abroad and Shopozz.COM free mail forwarding in the USA), like in many others, there are many mailing lists with different content. Of course, such emails should reach the end user extremely quickly, especially since the content of most of them implies prompt delivery by default. When the total time of sending the next mailing list exceeded 24 hours, it became clear that it was necessary to somehow optimize this process.



So, the task: there are over 200,000 users for whom you need to generate and deliver a message in the shortest possible time. Solution of the problem - under the cut.
')
The very first thought that comes to mind is to increase the number of channels. However, the good old sync comes into play - it takes about 0.5 - 1 second to generate and send one message, and if several senders take up the job at once, you should not let them take on the same user twice.

In order not to load the base or the server once more, it was decided to use the server queue beanstalkd as a resource allocator .



The choice fell on this server largely because of the ease of implementation. Beanstalkd is simple and no frills. More information about the message queue is already written here and here .

The general scheme of sending messages is as follows:



The list of recipients is formed by the usual database request and converted into a set of job messages for beanstalkd.

It should be noted that in beanstalkd the maximum size of the job is limited and by default it is 256 bytes of text (the maximum size can be specified differently). In addition, you should not allocate a separate job for each user, because receiving and parsing data from beanstalkd also takes up server resources.

By the method of selection, we came to the figure of 100 users on the job.

The Worker in this scheme is a php script that constantly listens to the server channel for new jobs, gets a list of users from them and sends messages.

Algorithm of work of a worker:

while(1) { $ci = get_instance(); $ci->load->model('shared/queue_model'); $ci->load->model('shared/sendmail_model'); if ($ci->queue_model->watch_jobs()) { // ,    job' print 'Memory usage = ' . intval(memory_get_peak_usage() / 1024) . ' KB' . PHP_EOL; } unset($ci); } 

Model code queue_model:

 class Queue_model extends CI_Model { private $_pheanstalk; public function __construct() { $this->_pheanstalk = new Pheanstalk_Pheanstalk('XX.XX.XX.XX:XX'); $this->tube = (ENVIRONMENT === 'production') ? 'www.shopozz.ru' : 'default'; } public function watch_jobs() { @$job = $this->_pheanstalk->watch($this->tube)->reserve(); if ( ! isset($job) OR ! $job) { return FALSE; } $job_data = json_decode($job->getData()); if ( ! isset($job_data->type)) { $this->_pheanstalk->release($job); return FALSE; } if ($this->email_messages_model->can_send($job_data->mail)) { print date('H:i:s ') . 'START JOB!' . PHP_EOL; $data = $job->getData(); $this->_pheanstalk->delete($job); $this->_send_email($data); $time = round(microtime(true) - $start, 2); print date('H:i:s') . 'JOB DONE! Time=' . $time . PHP_EOL; return TRUE; } $this->_pheanstalk->release($job); return FALSE; } 

Here it is important to remember that the method

 $this->_pheanstalk->watch($this->tube)->reserve(); 

marks the current active member in the queue as reserved so that other workers cannot access it. After that, if the task is completed, it must be removed, and if it is not possible - be sure to return to the previous status through

 _pheanstalk->release($job); 

Method

 $this->email_messages_model->can_send 

checks if the broadcast is stopped. Thanks to this check, you can stop sending messages at any time (for example, if an error is found in the distribution text). In this case, each job will be read, but will not be executed, and, as a result, it will not be deleted, but will simply return to the active waiting state.

The final touch remains - since workers are in a state of constant looped start, they will not be able to correctly receive changes in the code. To do this, add to the loop a check for the time of the last change in the file:

 class Restarter { /** *     * @var integer */ private $_changed_time; /** *       * @var type */ private $_self_file; public function __construct() { $this->_self_file = __FILE__; $this->_changed_time = $this->_get_changed_time(); } /** *        * @return type */ private function _get_changed_time() { $fp = fopen($this->_self_file, "r"); $fstat = fstat($fp); $res = $fstat['mtime']; return $res; } /** *   ,      */ public function check_changed_time() { $time = $this->_get_changed_time(); if ($time !== $this->_changed_time) { echo "File has been changed!"; exit(1); } } } 

Now, if there are changes in the file, the script will cause the completion with the status “1”, after which it will be automatically restarted.

As a result, the task set at the top of the post is solved: now our users receive the newsletter as quickly as possible. With the increase in the number of users, which is natural, the service does not experience any problems with the final delivery time of the email to the recipient - everything is solved by a trivial increase in channels.

We remind you that you read the blog of the company Shopozz.com - absolutely free mail forwarding service in the USA with guaranteed delivery times. Subscribe to our blog , register on our service and save even more on purchases abroad.