Ruby & PHP. Скрещиваем ужа с ежом с помощью Starling и Zend_Queue.

В этой статье я расскажу о животрепещущем для многих вопросе. Как соединить между собой приложения на разных языках. Например, Ruby и PHP. В Twitter проблему интеграции с очередью решили с помощью Starling. Вообще сейчас намечается тенденция, что для каждой задачи подбирают свой язык. Гомогенных систем становится всё меньше. В следствии этого возникает потребность в стандартах на интеграцию разношерстного ПО в единую систему.

Постановка задачи

Допустим у нас есть ситуация, при которой во время какого-либо HTTP запроса необходимо сделать длительную операцию. Например, при добавлении пользователем комментария к статье отправить email уведомление её автору. В классической схеме взаимодействия (рис 7-1) мы вынуждены ждать завершения этой операции, чтобы уведомить пользователя. Получается что эта операция блокирует выполнение веб-приложения, а значит происходит потеря производительности. Да и пользователю приходится ждать лишнее время. Когда количество пользователей невелико – это не проблема, но в Highload проектах этот вопрос приобретает большое значение. Мы определились с задачей – вынести все ресурсоемкие задачи в фоновые процессы. Рассмотрим случай с хостингом и добавлением домена в веб-панели. Задачей будет выполнение скрипта на Ruby, который будет производить манипуляции с DNS зонами и перезагружать DNS сервис.

Теперь определимся с схемой работы. Клиентом очереди будет ZendFramework приложение, которое будет добавлять задачи в очередь (отправлять сообщения в очередь). Сервером (воркером) будет Ruby, который будет принимать сообщения из очереди и обрабатывать их.

Почему выбрана именно такая комбинация? ZF для клиента исторически сложился, а вот сервер на Ruby выбран, т.к. очень не хочется давать PHP права root, думаю многие админы меня поймут. Выбор очереди был сделан в пользу Starling, т.к. она основана на Memcached и использует его протокол (ну почти 🙂 ) и реализация на руби предельно проста.

Что такое очереди сообщений

“А как вы организуете очередь?
– С помощью бабушек!”

с Хабрахабра

Тема очередей сообщений была навеяна докладом “Разделение труда: Организация многозадачной, распределенной системы в Zend Framework с помощью Job Queue” Александра Готгельфа на последнем ZFCONF. Он работал с сервером очередей Gearman, и рассказал о его немногочисленных, но очень серьезных багах (таких, как проблема освобождения памяти). В связи с этим я начал смотреть в торону других и остановился на MemcacheQ и Starling, который очень советовали пользователи хабра. А т.к. имплементация на руби проще со вторым он и был выбран.

Что такое Starling?

Starling – это скворец, певчая птица семейства скворцовых, широко распространённая на значительной территории Евразии, а также успешно интродуцированная в Южную Африку, Северную Америку, Австралию и Новую Зеландию. На юге и западе Европы ведёт оседлый образ жизни, а в северной и восточной её части является перелётной, в зимние месяцы мигрируя на юг. Внешне (размерами, желтым клювом и темным оперением) слегка напоминает чёрных дроздов, но в отличие от них ходит по земле, а не прыгает. 🙂

А ещё так называется очередь сообщений, написанная на Ruby. Её используют в Twitter! Для установки сервера очередей у вас уже должен быть установлен Ruby. Сам же Starling ставится очень просто. Следующая строка выполняется только один раз!

gem sources -a http://gems.github.com/

1	gem sources -a http://gems.github.com/

Ставим гем (можно делать сколько угодно раз).

sudo gem install starling

1	sudo gem install starling

Если вы ставили руби для всех юзеров, то выполняем вместо sudo rvmsudo

Дальше надо запустить сервер и можно коннектить серверов и клиентов.

andrey@comp$ starling
Starting at 127.0.0.1:22122.
I, [2011-05-20T10:53:34.346297 #6017]  INFO -- : Starling STARTUP on 127.0.0.1:22122

andrey@comp$ starling

Starting at 127.0.0.1:22122.

I, [2011-05-20T10:53:34.346297 #6017] INFO -- : Starling STARTUP on 127.0.0.1:22122

Клиентами будем называть тех, кто отправляет задания в очередь, а серверами – тех кто получает и обслуживает их (to serve). Теперь можно подключать клиентов на руби и php.

Общий формат данных. Ruby маршализация в PHP

Первое с чем я столкнулся это формат хранения сообщений в очереди. На выбор было несколько вариантов.

[б] Маршализация (marshalling) объектов. Так делает руби и это дефолтовый способ работы Starling,
[т] Сериализация (serializing). Так хранит объекты Zend Framework, а именно Zend_Queue,
[т] JSON. Стандарт де-факто для взаимодействия в гетерогенных средах,
[т] XML. Стандарт де-юро, однако имеет больший расход памяти для хранения сообщений, т.к. имеет теговую природу,

где [т] – текстовый формат, [б] -бинарный формат.

Скажу так, в начале у меня был соблазн вообще не трогать Starling, а научиться в PHP работать с маршализованными руби объектами. Но ничего похожего я найти не смог. Использовать XML не хотелось, для работы с JSON надо было бы здорово переписывать сервер очередей. Вот тогда то мне на глаза и попался гем php_serialize.

Ruby + Starling

Делает он ровно то, что я от него и ожидал, а именно выполняет serialize() функцию над Ruby объектами и наоборот. Итак схема такова: php -> serialize -> starling -> unserialize -> ruby. Для этого правда всё равно пришлось коё-что переписать в Starling. Приведу здесь код простейшего воркера очереди.

require 'starling'
require 'php_serialize'

class Starling
 def get_raw(key, raw = false)
 with_server(key) do |server, cache_key|
 logger.debug { "get #{key} from #{server.inspect}" } if logger
 value = cache_get server, cache_key
 return nil if value.nil?
 value = PHP.unserialize value unless raw
 return value
 end
 rescue TypeError => err
 handle_error nil, err
 end
end
# --------------------------------------------------
starling = Starling.new('0.0.0.0:22122')

puts starling.sizeof('myqueue1')
puts starling.get_raw('myqueue1')

exit

require 'starling'

require 'php_serialize'

class Starling

def get_raw(key, raw = false)

with_server(key) do |server, cache_key|

logger.debug { "get #{key} from #{server.inspect}" } if logger

value = cache_get server, cache_key

return nil if value.nil?

value = PHP.unserialize value unless raw

return value

end

rescue TypeError => err

handle_error nil, err

end

# --------------------------------------------------

starling = Starling.new('0.0.0.0:22122')

puts starling.sizeof('myqueue1')

puts starling.get_raw('myqueue1')

exit

Как видите, вместо Marshal.load из MemCache::get и использую PHP::unserialize и пропустив один уровень наследования (MemCache) для того, чтобы не поломать обычный MemCache клиент я внедрил его в класс Starling. Отлчино, теперь примемся за вторую часть нашей системы.

PHP + Starling (ZendExtra_Queue_Adapter_Starling)

В Zend Framework уже есть неплохой класс работы с очередями – Zend_Queue, но к сожалению он не поддерживает Starling. Для внедрения поддержки создадим наш адаптер, который отнаследуем от ближайшего родственника (Zend_Queue_Adapter_MemcacheQ).

В Зенд-Кью (а именно так читается Zend_Queue) очень хорошо реализована работа с адаптерами хранилищ. Каждый из них очень разный, и паттерн “Стратегия” тут бы не подошёл, однако авторы нашли выход и добавили функцию getCapabilities(), которая возвращает масси возможностей хранилища.

    /**
     * Supporting abilities
     *
     * @return array
     */
    public function getCapabilities()
    {
        return array(
            'create'        => true,
            'delete'        => true,
            'send'          => true,
            'receive'       => true,
            'deleteMessage' => false,
            'getQueues'     => true,
            'count'         => true,
            'isExists'      => true,
        );
    }

/**

* Supporting abilities

* @return array

public function getCapabilities()

{

return array(

'create' => true,

'delete' => true,

'send' => true,

'receive' => true,

'deleteMessage' => false,

'getQueues' => true,

'count' => true,

'isExists' => true,

);

}

Для нашего хранилища есть возможность получить количество сообщений в очереди на обслуживание, но его надо реализовать. В написал/переписал/дописал часть функций адаптера в результате родился
ZendExtra_Queue_Adapter_Starling

<?php

/**
 * Adapter for storage messags in Starling queue
 * Note: messages are stored serialized
 * After recieving outside this adapter unserialization is needed
 * For Ruby you can use gem "php_serialize" (not "php-serialize")
 */
require_once 'Zend/Queue/Adapter/AdapterAbstract.php';
class ZendExtra_Queue_Adapter_Starling extends Zend_Queue_Adapter_Memcacheq
{
 const DEFAULT_HOST = '127.0.0.1';
 const DEFAULT_PORT = 22122;
 const EOL          = "\r\n";

 /**
 * Supporting abilities
 *
 * @return array
 */
 public function getCapabilities()
 {
 return array(
 'create'        => true,
 'delete'        => true,
 'send'          => true,
 'receive'       => true,
 'deleteMessage' => false,
 'getQueues'     => true,
 'count'         => true,
 'isExists'      => true,
 );
 }

 public function __construct($options, Zend_Queue $queue = null)
 {
 if (!extension_loaded('memcache')) {
 require_once 'Zend/Queue/Exception.php';
 throw new Zend_Queue_Exception('Memcache extension does not appear to be loaded');
 }

 Zend_Queue_Adapter_AdapterAbstract::__construct($options, $queue); // instead of parent::, which is MemcachedQ

 $options = &$this->_options['driverOptions'];

 if (!array_key_exists('host', $options)) {
 $options['host'] = self::DEFAULT_HOST;
 }
 if (!array_key_exists('port', $options)) {
 $options['port'] = self::DEFAULT_PORT;
 }

 $this->_cache = new Memcache();

 $result = $this->_cache->connect($options['host'], $options['port']);

 if ($result === false) {
 require_once 'Zend/Queue/Exception.php';
 throw new Zend_Queue_Exception('Could not connect to MemcacheQ');
 }

 $this->_host = $options['host'];
 $this->_port = (int)$options['port'];
 }

 /**
 * Return queues list
 *
 * @return array
 */
 public function getQueues()
 {
 $this->_queues = array();

 $response = $this->_sendCommand('stats', array('END'));

 $postfixesArray = array('_items', '_total_items', '_logsize', '_expired_items', '_age', '_total', '_expired');
 $arr = array();
 foreach ($response as $i => $line) {
 if(strpos($line, 'STAT queue_') === 0 ) { // Zero position (start), not false
 $tmp = str_replace('STAT queue_', '', $line);
 $tmp = explode(' ', $tmp);
 $tmp = $tmp[0];
 foreach($postfixesArray as $postfix) {
 $tmp = str_replace($postfix, '', $tmp);
 }
 $arr[] = $tmp;
 }
 }
 $this->_queues = array_unique($arr);
 return $this->_queues;
 }

 /**
 * Return the approximate number of messages in the queue
 *
 * @return integer
 * @throws Zend_Queue_Exception (not supported)
 */
 public function count(Zend_Queue $queue = null)
 {
 if($queue === NULL) {
 $keyName = 'queue_'.$this->getQueue()->getName().'_items';
 } else {
 $keyName = 'queue_'.$queue->getName().'_items';
 }
 $response = $this->_sendCommand('stats', array('END'));
 foreach($response as $line) {
 if(strpos($line, $keyName) !== false) {
 $tmp = str_replace('STAT '.$keyName.' ', '', $line);
 return (int) $tmp;
 }
 }
 return 0;
 }

 /**
 * Get messages in the queue
 *
 * @param  integer    $maxMessages  Maximum number of messages to return
 * @param  integer    $timeout      Visibility timeout for these messages
 * @param  Zend_Queue $queue
 * @return Zend_Queue_Message_Iterator
 * @throws Zend_Queue_Exception
 */
 public function receive($maxMessages=null, $timeout=null, Zend_Queue $queue=null)
 {
 if ($maxMessages === null) {
 $maxMessages = 1;
 }

 if ($timeout === null) {
 $timeout = self::RECEIVE_TIMEOUT_DEFAULT;
 }
 if ($queue === null) {
 $queue = $this->_queue;
 }

 $msgs = array();

 // Setting up the limit upon to queue count
 if($maxMessages != NULL) {
 $count = $this->count($queue);
 if($count < $maxMessages)
 $maxMessages = $count;
 }

 if ($maxMessages > 0 ) {
 for ($i = 0; $i < $maxMessages; $i++) {
 $data = unserialize($this->_cache->get($queue->getName()));
 $msgs[] = $data;
 }
 }

 $options = array(
 'queue'        => $queue,
 'data'         => $msgs,
 'messageClass' => $queue->getMessageClass(),
 );
 $classname = $queue->getMessageSetClass();
 if (!class_exists($classname)) {
 require_once 'Zend/Loader.php';
 Zend_Loader::loadClass($classname);
 }
 return new $classname($options);
 }

 /**
 * Delete a queue and all of it's messages
 *
 * Returns false if the queue is not found, true if the queue exists
 *
 * @param  string  $name queue name
 * @return boolean
 * @throws Zend_Queue_Exception
 */
 public function delete($name)
 {
 $queue = $this->getQueue($name);
 $count = $this->count($queue);
 $this->receive($count, NULL, $queue);
 }

 /**
 * Send a message to the queue
 *
 * @param  string     $message Message to send to the active queue
 * @param  Zend_Queue $queue
 * @return Zend_Queue_Message
 * @throws Zend_Queue_Exception
 */
 public function send($message, Zend_Queue $queue=null)
 {
 if ($queue === null) {
 $queue = $this->_queue;
 }

 if (!$this->isExists($queue->getName())) {
 require_once 'Zend/Queue/Exception.php';
 throw new Zend_Queue_Exception('Queue does not exist:' . $queue->getName());
 }

 $message = serialize($message);
 $data    = array(
 'message_id' => md5(uniqid(rand(), true)),
 'handle'     => null,
 'body'       => $message,
 'md5'        => md5($message),
 );

 $result = $this->_cache->set($queue->getName(), $message, 0, 0);
 if ($result === false) {
 require_once 'Zend/Queue/Exception.php';
 throw new Zend_Queue_Exception('failed to insert message into queue:' . $queue->getName());
 }

 $options = array(
 'queue' => $queue,
 'data'  => $data,
 );

 $classname = $queue->getMessageClass();
 if (!class_exists($classname)) {
 require_once 'Zend/Loader.php';
 Zend_Loader::loadClass($classname);
 }
 return new $classname($options);
 }

}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

<?php

/**

* Adapter for storage messags in Starling queue

* Note: messages are stored serialized

* After recieving outside this adapter unserialization is needed

* For Ruby you can use gem "php_serialize" (not "php-serialize")

require_once 'Zend/Queue/Adapter/AdapterAbstract.php';

class ZendExtra_Queue_Adapter_Starling extends Zend_Queue_Adapter_Memcacheq

{

const DEFAULT_HOST = '127.0.0.1';

const DEFAULT_PORT = 22122;

const EOL = "\r\n";

/**

* Supporting abilities

* @return array

public function getCapabilities()

{

return array(

'create' => true,

'delete' => true,

'send' => true,

'receive' => true,

'deleteMessage' => false,

'getQueues' => true,

'count' => true,

'isExists' => true,

);

}

public function __construct($options, Zend_Queue $queue = null)

{

if (!extension_loaded('memcache')) {

require_once 'Zend/Queue/Exception.php';

throw new Zend_Queue_Exception('Memcache extension does not appear to be loaded');

}

Zend_Queue_Adapter_AdapterAbstract::__construct($options, $queue); // instead of parent::, which is MemcachedQ

$options = &$this->_options['driverOptions'];

if (!array_key_exists('host', $options)) {

$options['host'] = self::DEFAULT_HOST;

}

if (!array_key_exists('port', $options)) {

$options['port'] = self::DEFAULT_PORT;

}

$this->_cache = new Memcache();

$result = $this->_cache->connect($options['host'], $options['port']);

if ($result === false) {

require_once 'Zend/Queue/Exception.php';

throw new Zend_Queue_Exception('Could not connect to MemcacheQ');

}

$this->_host = $options['host'];

$this->_port = (int)$options['port'];

}

/**

* Return queues list

* @return array

public function getQueues()

{

$this->_queues = array();

$response = $this->_sendCommand('stats', array('END'));

$postfixesArray = array('_items', '_total_items', '_logsize', '_expired_items', '_age', '_total', '_expired');

$arr = array();

foreach ($response as $i => $line) {

if(strpos($line, 'STAT queue_') === 0 ) { // Zero position (start), not false

$tmp = str_replace('STAT queue_', '', $line);

$tmp = explode(' ', $tmp);

$tmp = $tmp[0];

foreach($postfixesArray as $postfix) {

$tmp = str_replace($postfix, '', $tmp);

}

$arr[] = $tmp;

}

$this->_queues = array_unique($arr);

return $this->_queues;

}

/**

* Return the approximate number of messages in the queue

* @return integer

* @throws Zend_Queue_Exception (not supported)

public function count(Zend_Queue $queue = null)

{

if($queue === NULL) {

$keyName = 'queue_'.$this->getQueue()->getName().'_items';

} else {

$keyName = 'queue_'.$queue->getName().'_items';

}

$response = $this->_sendCommand('stats', array('END'));

foreach($response as $line) {

if(strpos($line, $keyName) !== false) {

$tmp = str_replace('STAT '.$keyName.' ', '', $line);

return (int) $tmp;

}

return 0;

}

/**

* Get messages in the queue

* @param integer $maxMessages Maximum number of messages to return

* @param integer $timeout Visibility timeout for these messages

* @param Zend_Queue $queue

* @return Zend_Queue_Message_Iterator

* @throws Zend_Queue_Exception

public function receive($maxMessages=null, $timeout=null, Zend_Queue $queue=null)

{

if ($maxMessages === null) {

$maxMessages = 1;

}

if ($timeout === null) {

$timeout = self::RECEIVE_TIMEOUT_DEFAULT;

}

if ($queue === null) {

$queue = $this->_queue;

}

$msgs = array();

// Setting up the limit upon to queue count

if($maxMessages != NULL) {

$count = $this->count($queue);

if($count < $maxMessages)

$maxMessages = $count;

}

if ($maxMessages > 0 ) {

for ($i = 0; $i < $maxMessages; $i++) {

$data = unserialize($this->_cache->get($queue->getName()));

$msgs[] = $data;

}

$options = array(

'queue' => $queue,

'data' => $msgs,

'messageClass' => $queue->getMessageClass(),

);

$classname = $queue->getMessageSetClass();

if (!class_exists($classname)) {

require_once 'Zend/Loader.php';

Zend_Loader::loadClass($classname);

}

return new $classname($options);

}

/**

* Delete a queue and all of it's messages

* Returns false if the queue is not found, true if the queue exists

* @param string $name queue name

* @return boolean

* @throws Zend_Queue_Exception

public function delete($name)

{

$queue = $this->getQueue($name);

$count = $this->count($queue);

$this->receive($count, NULL, $queue);

}

/**

* Send a message to the queue

* @param string $message Message to send to the active queue

* @param Zend_Queue $queue

* @return Zend_Queue_Message

* @throws Zend_Queue_Exception

public function send($message, Zend_Queue $queue=null)

{

if ($queue === null) {

$queue = $this->_queue;

}

if (!$this->isExists($queue->getName())) {

require_once 'Zend/Queue/Exception.php';

throw new Zend_Queue_Exception('Queue does not exist:' . $queue->getName());

}

$message = serialize($message);

$data = array(

'message_id' => md5(uniqid(rand(), true)),

'handle' => null,

'body' => $message,

'md5' => md5($message),

);

$result = $this->_cache->set($queue->getName(), $message, 0, 0);

if ($result === false) {

require_once 'Zend/Queue/Exception.php';

throw new Zend_Queue_Exception('failed to insert message into queue:' . $queue->getName());

}

$options = array(

'queue' => $queue,

'data' => $data,

);

$classname = $queue->getMessageClass();

if (!class_exists($classname)) {

require_once 'Zend/Loader.php';

Zend_Loader::loadClass($classname);

}

return new $classname($options);

}

А вот и пример инициализации класса и работы с ним:
StarlingZF.php

$queuePool = new Zend_Queue(new ZendExtra_Queue_Adapter_Starling(array()), array('name' => 'myqueue1'));
$queuePool->send(array('domain' => 'site.ru'));
var_dump($queuePool->receive()->toArray()); // Messages also deleting from queue

$queuePool = new Zend_Queue(new ZendExtra_Queue_Adapter_Starling(array()), array('name' => 'myqueue1'));

$queuePool->send(array('domain' => 'site.ru'));

var_dump($queuePool->receive()->toArray()); // Messages also deleting from queue

Или тоже самое но на уровне PHP:
StarlingPHP.php

connect('localhost', 22122) or die ("Could not connect");
$data = array('domain' => 'test2.ru');
$memcache->set('myqueue1', serialize($data)) or die ("Failed to save data at the server");
print 'ok';

connect('localhost', 22122) or die ("Could not connect");

$data = array('domain' => 'test2.ru');

$memcache->set('myqueue1', serialize($data)) or die ("Failed to save data at the server");

print 'ok';

Как видите всё не так сложно. Теперь можно даже написать утилиту мониторинга (если вам не подходят существующие), которая будет следить за очередью.

Преимущества и недостатки Starling

Давайте посмотрим, что у нас получилось, и что нам это даёт.

Скорость. Данные хранятся в Memcache, который славится своей производительностью.
Масштабируемость. Очередь может быть распределена на несколько серверов.
Стандартизированность и унификация. Используется стандартный Memcached-протокол.

Но есть и недостатки.

Надежность. Т.к. все данные хранятся в памяти, то при перезапуске сервера задачи – теряются.

Но в некоторых случаях это может быть не критично. Например в случае хостера отметка о завершении задачи ставится в конце неё, и при необходимости клиентская часть сможет повторить запрос. В случае отправки email -есть внешняя очередь Postfix, куда ставятся сообщения на отправку, а Starling является буферной очередью для почтового сервера. В случае ресайза картинок, задача на ресайз может быть поставлена при повторном обращении ккартинке по проверки условия выполнения задачи. Вот, что пишут про неё сами разработчики.

Система надёжна, быстра и использует стандартный протокол memcached. По заверениям разработчиков, это вообще самое стабильное звено твиттера. Когда другие элементы системы отключаются, Starling всегда продолжает работать.

Так что не всё так плохо, важно понимать, что идеальных решений не существует и любое из них компромисс в ту или иную сторону и определитсья с приоритетами. Если приоритет – Highload, то, думаю, сквоец – отлчиное решение!

Литература

http://www.aagh.net/projects/ruby-php-serialize
http://rubypond.com/blog/the-complete-guide-to-setting-up-starling
https://github.com/highgroove/scout-plugins/raw/master/starling_monitor/starling_monitor.rb
https://github.com/starling/starling
http://rubyjunction.us/ruby-asynchronous-messaging
http://habrahabr.ru/blogs/hi/44907/
http://habrahabr.ru/blogs/hi/45891/
ZF Proposal

May 21, 2011 admin

DNS gem JSON php php_serialize queuq Ruby RVM starling XML Zend_Queue маршализация очереди сообщений сериализация хостинг

Постановка задачи

Что такое очереди сообщений

Что такое Starling?

Общий формат данных. Ruby маршализация в PHP

Ruby + Starling

PHP + Starling (ZendExtra_Queue_Adapter_Starling)

Преимущества и недостатки Starling

Литература

Поиск по сайту

Свежие посты

Категории