public function testDeleteMessage()
 {
     $queue = $this->queue->createQueue('test');
     $queue->send(array(1, 2, 3));
     foreach ($queue->receive() as $message) {
         $queue->deleteMessage($message);
     }
     $values = $this->rediska->getList('Zend_Queue_queue_test');
     $this->assertEquals(array(), $values);
 }

 public function send($template)
 {
     $config = new Zend_Config_Ini(APPLICATION_PATH . '/configs/application.ini', APPLICATION_ENV);
     $queue = new Zend_Queue('Activemq', $config->queue->activemq->toArray());
     $data = array();
     $data['RecipientEmail'] = $template->getRecipientEmail();
     $data['RecipientName'] = $template->getRecipientName();
     $data['From'] = $template->getFrom();
     $data['Subject'] = $template->getSubject();
     $data['BodyText'] = $template->getBodyText();
     $data['BodyHtml'] = $template->getBodyHtml();
     $data['SenderEmail'] = $template->getSenderEmail();
     $data['SenderName'] = $template->getSenderName();
     $message = $queue->send(serialize($data));
 }

 /**
  * List all queues.
  *
  * @param  array $options
  * @return array Queue IDs
  */
 public function listQueues($options = null)
 {
     try {
         return $this->_queue->getQueues();
     } catch (Zend_Queue_Exception $e) {
         throw new Zend_Cloud_QueueService_Exception('Error on listing queues: ' . $e->getMessage(), $e->getCode(), $e);
     }
 }

 public function start(Zend_Queue $q)
 {
     Loader::library('database_indexed_search');
     $this->is = new IndexedSearch();
     Loader::model('attribute/categories/collection');
     Loader::model('attribute/categories/file');
     Loader::model('attribute/categories/user');
     $attributes = CollectionAttributeKey::getList();
     $attributes = array_merge($attributes, FileAttributeKey::getList());
     $attributes = array_merge($attributes, UserAttributeKey::getList());
     foreach ($attributes as $ak) {
         $ak->updateSearchIndex();
     }
     $db = Loader::db();
     $db->Execute('truncate table PageSearchIndex');
     $r = $db->Execute('select Pages.cID from Pages left join CollectionSearchIndexAttributes csia on Pages.cID = csia.cID where (ak_exclude_search_index is null or ak_exclude_search_index = 0) and cIsActive = 1');
     while ($row = $r->FetchRow()) {
         $q->send($row['cID']);
     }
 }

 /**
  * Send a message to the queue
  *
  * @param  Custom_Message|Custom_Messages $message message
  * @return $this
  * @throws Zend_Queue_Exception
  */
 public function send($message)
 {
     if (!($message instanceof Custom_Message || $message instanceof Custom_Messages)) {
         /**
          * @see Zend_Queue_Exception
          */
         require_once 'Zend/Queue/Exception.php';
         throw new Zend_Queue_Exception('$message must be an instance of Custom_Message or Custom_Messages');
     }
     if ($message instanceof Custom_Message) {
         $response = parent::send($message->__toString());
     } else {
         foreach ($message as $i => $one) {
             $response = parent::send($one->__toString());
         }
     }
     return $this;
 }

 /**
  * Return the approximate number of messages in the queue
  *
  * @param  Zend_Queue $queue
  * @return integer
  * @throws Zend_Queue_Exception (not supported)
  */
 public function count(Zend_Queue $queue = null)
 {
     if ($queue !== null) {
         return $queue->count();
     }
     return false;
 }

 /**
  * @group ZF-7650
  */
 public function testReceiveWillRetrieveZeroItems()
 {
     $options = $this->getTestConfig();
     $options['name'] = '/temp-queue/ZF7650';
     $queue = new Zend_Queue('Db', $options);
     $queue2 = $queue->createQueue('queue');
     $queue->send('My Test Message 1');
     $queue->send('My Test Message 2');
     $messages = $queue->receive(0);
     $this->assertEquals(0, count($messages));
 }

 public function test_ZF_7650()
 {
     // Zend_Queue_Adapter_Array
     $queue = new Zend_Queue('Array');
     $queue2 = $queue->createQueue('queue');
     $queue->send('My Test Message 1');
     $queue->send('My Test Message 2');
     $messages = $queue->receive(0);
     $this->assertEquals(0, count($messages));
     // Zend_Queue_Adapter_Memcacheq
     $driverOptions = array();
     if (defined('TESTS_ZEND_QUEUE_MEMCACHEQ_HOST')) {
         $driverOptions['host'] = TESTS_ZEND_QUEUE_MEMCACHEQ_HOST;
     }
     if (defined('TESTS_ZEND_QUEUE_MEMCACHEQ_PORT')) {
         $driverOptions['port'] = TESTS_ZEND_QUEUE_MEMCACHEQ_PORT;
     }
     $options = array('name' => 'ZF7650', 'driverOptions' => $driverOptions);
     $queue = new Zend_Queue('Memcacheq', $options);
     $queue2 = $queue->createQueue('queue');
     $queue->send('My Test Message 1');
     $queue->send('My Test Message 2');
     $messages = $queue->receive(0);
     $this->assertEquals(0, count($messages));
     // Zend_Queue_Adapter_Db
     $driverOptions = array();
     if (defined('TESTS_ZEND_QUEUE_DB')) {
         require_once 'Zend/Json.php';
         $driverOptions = Zend_Json::decode(TESTS_ZEND_QUEUE_DB);
     }
     $options = array('name' => '/temp-queue/ZF7650', 'options' => array(Zend_Db_Select::FOR_UPDATE => true), 'driverOptions' => $driverOptions);
     $queue = new Zend_Queue('Db', $options);
     $queue2 = $queue->createQueue('queue');
     $queue->send('My Test Message 1');
     $queue->send('My Test Message 2');
     $messages = $queue->receive(0);
     $this->assertEquals(0, count($messages));
     // Zend_Queue_Adapter_Activemq
     $driverOptions = array();
     if (defined('TESTS_ZEND_QUEUE_ACTIVEMQ_HOST')) {
         $driverOptions['host'] = TESTS_ZEND_QUEUE_ACTIVEMQ_HOST;
     }
     if (defined('TESTS_ZEND_QUEUE_ACTIVEMQ_PORT')) {
         $driverOptions['port'] = TESTS_ZEND_QUEUE_ACTIVEMQ_PORT;
     }
     if (defined('TESTS_ZEND_QUEUE_ACTIVEMQ_SCHEME')) {
         $driverOptions['scheme'] = TESTS_ZEND_QUEUE_ACTIVEMQ_SCHEME;
     }
     $options = array('driverOptions' => $driverOptions);
     $queue = new Zend_Queue('Activemq', $options);
     $queue2 = $queue->createQueue('queue');
     $queue->send('My Test Message 1');
     $queue->send('My Test Message 2');
     $messages = $queue->receive(0);
     $this->assertEquals(0, count($messages));
 }

<?php

require_once 'config.inc.php';
require_once 'include/utils/utils.php';
require_once 'include/Zend/Queue.php';
global $adb, $current_user;
global $currentModule;
global $dbconfig;
$options = array('name' => 'queue1', 'driverOptions' => array('host' => $dbconfig['db_server'], 'port' => substr($dbconfig['db_port'], 1), 'username' => $dbconfig['db_username'], 'password' => $dbconfig['db_password'], 'dbname' => $dbconfig['db_name'], 'type' => 'pdo_mysql'));
$queue = new Zend_Queue('Db', $options);
//$queue->send('postdata=1&name=1');
$nowdate = date("Y-m-d");
$nowdatetime = date("Y-m-d H:i:s");
$sjid = $_REQUEST["sjid"];
$subject = $_REQUEST["subject"];
$mailcontent = $_REQUEST["mailcontent"];
$mailcontent = stripslashes($mailcontent);
//$mailfrom = $_REQUEST["mailfrom"];
//$mailfrom_arr = explode("(",$mailfrom);
//$from_email = $mailfrom_arr[0];
//$from_name  = substr($mailfrom_arr[1],0,-1);
//$interval  = substr($mailfrom_arr[2],0,-2);
$from_name = $_REQUEST["from_name"];
$from_email = $_REQUEST["from_email"];
$interval = $_REQUEST["interval"];
$receiveaccountinfo = $_REQUEST["receiveaccountinfo"];
//$receiveaccountinfo =  str_replace("##",'&',$receiveaccountinfo);
//$receiveaccountarr = explode("\n",$receiveaccountinfo);
$receiveaccountarr = explode("**", $receiveaccountinfo);
//var_dump($receiveaccountarr);
//exit;

 /**
  * Subscribes the client to the queue.
  *
  * @param  Zend_Queue $queue
  * @return void
  */
 protected function _subscribe(Zend_Queue $queue)
 {
     $frame = $this->_client->createFrame();
     $frame->setCommand('SUBSCRIBE');
     $frame->setHeader('destination', $queue->getName());
     $frame->setHeader('ack', 'client');
     $this->_client->send($frame);
     $this->_subscribed[$queue->getName()] = true;
 }

 /**
  * @group ZF-7650
  */
 public function testReceiveWillRetrieveZeroItems()
 {
     $options = array('driverOptions' => $this->getTestConfig());
     $queue = new Zend_Queue('Activemq', $options);
     $queue2 = $queue->createQueue('queue');
     $queue->send('My Test Message 1');
     $queue->send('My Test Message 2');
     $messages = $queue->receive(0);
     $this->assertEquals(0, count($messages));
 }

 /**
  * @group ZF-7650
  */
 public function testReceiveWillRetrieveZeroItems()
 {
     // Zend_Queue_Adapter_Array
     $queue = new Zend_Queue('Array');
     $queue2 = $queue->createQueue('queue');
     $queue->send('My Test Message 1');
     $queue->send('My Test Message 2');
     $messages = $queue->receive(0);
     $this->assertEquals(0, count($messages));
 }

define('IN_CRMONE', true);
$root_directory = dirname(__FILE__) . "/";
require $root_directory . 'config.php';
require_once $root_directory . 'include/utils/utils.php';
require_once $root_directory . 'include/utils/CommonUtils.php';
require_once $root_directory . 'include/database/PearDatabase.php';
require_once $root_directory . 'include/logging.php';
require_once $root_directory . 'modules/Users/Users.php';
require_once $root_directory . 'include/utils/clean_incoming_data.php';
require_once $root_directory . 'user_privileges/seqprefix_config.php';
require_once $root_directory . 'include/Zend/Queue.php';
global $adb;
global $site_URL;
global $dbconfig;
$options = array('name' => 'queue1', 'driverOptions' => array('host' => $dbconfig['db_server'], 'port' => substr($dbconfig['db_port'], 1), 'username' => $dbconfig['db_username'], 'password' => $dbconfig['db_password'], 'dbname' => $dbconfig['db_name'], 'type' => 'pdo_mysql'));
$queue = new Zend_Queue('Db', $options);
//一分钟发送25条
$messages = $queue->receive(100);
foreach ($messages as $i => $message) {
    //参数字符串
    $postdata = $message->body;
    $posts = explode("&", $postdata);
    //sjid
    $sjid_arr = explode("=", $posts[0]);
    $sjid = $sjid_arr['1'];
    //maillogsid
    $maillogsid_arr = explode("=", $posts[1]);
    $maillogsid = $maillogsid_arr['1'];
    //判断是否已经发送过了
    $mailflag = checkMaillog($maillogsid);
    if (!$mailflag) {

 /**
  * Constructor
  *
  * @param string|Zend_Queue_Adapter_Abstract $adapter
  * @param array  $config
  */
 public function testConstruct()
 {
     // Test Zend_Config
     $config = array('name' => 'queue1', 'params' => array(), 'adapter' => 'array');
     $zend_config = new Zend_Config($config);
     $obj = new Zend_Queue($config);
     $this->assertTrue($obj instanceof Zend_Queue);
     // test logger
     $this->assertTrue($obj->getLogger() instanceof Zend_Log);
     $obj = new Zend_Queue($zend_config);
     $this->assertTrue($obj instanceof Zend_Queue);
     try {
         $obj = new Zend_Queue('ops');
         $this->fail('Zend_Queue cannot accept a string');
     } catch (Exception $e) {
         $this->assertTrue(true);
     }
 }

 /**
  * Process all pending tasks
  */
 public function work()
 {
     while (count($task = $this->_queue->receive()) > 0) {
         try {
             $taskData = $task->toArray();
             $this->getWorker()->execute($taskData[0]);
         } catch (Exception $e) {
             $this->_logger->logException($e);
         }
     }
 }

 /**
  * queues an action
  * 
  * @param string $_action
  * @param mixed  $_arg1
  * @param mixed  $_arg2
  * ...
  * 
  * @return string the job id
  */
 public function queueAction()
 {
     $params = func_get_args();
     $action = array_shift($params);
     $message = array('action' => $action, 'account_id' => Tinebase_Core::getUser()->getId(), 'params' => $params);
     if (Tinebase_Core::isLogLevel(Zend_Log::INFO)) {
         Tinebase_Core::getLogger()->info(__METHOD__ . '::' . __LINE__ . " queueing action: '{$action}'");
     }
     return $this->_queue->send($message);
 }

 /**
  * Constructor
  * 
  * @param  string $queueName
  * 
  * @throws Mage_Core_Exception If $queueName is not a string
  */
 public function __construct($queueName = 'default')
 {
     if (empty($queueName)) {
         $queueName = 'default';
     }
     if (!is_string($queueName)) {
         Mage::throwException('Invalid queueName');
     }
     /** @var Mage_Core_Model_Resource $resource */
     $resource = Mage::getModel('core/resource');
     /** @var Varien_Db_Adapter_Interface $connection */
     $connection = $resource->getConnection(Mage_Core_Model_Resource::DEFAULT_WRITE_RESOURCE);
     $config = array('name' => $queueName, 'messageClass' => 'Aoe_Queue_Model_Message', 'dbAdapter' => $connection, 'dbQueueTable' => $resource->getTableName('aoe_queue/queue'), 'dbMessageTable' => $resource->getTableName('aoe_queue/message'));
     parent::__construct($config);
     /** @var Aoe_Queue_Model_Adapter_Db $adapter */
     $adapter = Mage::getModel('aoe_queue/adapter_db', $this->getOptions());
     $this->setAdapter($adapter);
 }

 public function send($message)
 {
     parent::send((string) $message);
 }

 /**
  * Retrieve reporter instance
  *
  * @return Enterprise_Queue_Model_Worker_Reporter_Interface
  */
 public function getReporter()
 {
     return $this->_queue->getAdapter() instanceof Enterprise_Queue_Model_Adapter_Db ? Mage::getModel('enterprise_queue/worker_reporter_native') : Mage::getModel('enterprise_queue/worker_reporter_null');
 }

 /**
  * Add task to queue
  *
  * array (
  *     'task_name' => '...',
  *     'params' => array(),
  * )
  *
  * @param array $task
  * @return Enterprise_Queue_Model_Consumer
  */
 public function addTask(array $task)
 {
     $this->_queue->send(Zend_Json::encode($task));
     return $this;
 }