<?php

class Queue_model extends CI_Model {

    // 콜랙션 이름
    private $queue_collection = 'queues';

    function __construct() {
        parent::__construct();
        // 라이브러리 불러오기
        $this->load->library('mongolib');
    }

    /**
     * 인덱스 만들기, 만들려면 한번 실행을 해주세요.
     */
    public function initialize_index() {
        $this->mongolib->add_index($this->queue_collection, array('locked' => 'ASC'));
    }

    /**
     * 작업 추가하기
     */
    public function add_queue($data) {
        $insert_data = array(
            'data' => $data,
            'locked' => null,
            'locked_at' => null,
            'date' => new MongoDate()
        );
        return $this->mongolib->insert($this->queue_collection, $insert_data);
    }

    /**
     * Queue 사이즈 불러오기, 작업필요한지 확인
     */
    public function count() {
        return $this->mongolib->where(array('locked' => null))->count($this->queue_collection);
    }

    /**
     * 작업 실행, 가장 최근작업 부터 불러와서 여기서 작업을 실행 한다.
     */
    public function pop_queue() {

        // find and modify
        // 같은 작업이 반복되지 않을려면, 잠기지 않은 아이템들부터 불러온다.
        $query = array('locked' => null);
        
        // 찻았으면 잠근다.
        $update = array('$set' => array('locked' => true, 'locked_at' => new MongoDate()));
        
        // 최근 날짜부터
        $sort = array('date' => 1);

        // 실행
        $data = $this->mongolib->find_and_modify($this->queue_collection, $query, null, $update);

        // 성공 했을경우, 아이템을 삭제하고 작업을 실행한다.
        if ($data['ok']) {
            $queue_item = $data['value'];
            $this->mongolib->where('_id', $queue_item['_id'])->delete($this->queue_collection);
            
            // 작업 실행.
            var_dump($queue_item['data']);
        }
    }
}

<?php if (!defined('BASEPATH')) exit('No direct script access allowed');

class queue extends CI_Controller {

    protected $phone;

    function __construct() {
        parent::__construct();
        $this->load->model('queue_model');
    }
    
    /**
     * 대몬으로 실행시킬 일꾼.
     */
    public function daemon() {
        while (1) {
            $cur_count = $this->queue_model->count();
            if ($cur_count > 0) {
                // 작업이 있으면 일을 시작해라.
                $this->queue_model->pop_queue();
            } else {
                // 없으면 잠시 기달림.
                sleep(rand(3, 7));
            }
        }
    }
}

[program:mysite]
command = php /var/www/index.php queue daemon
process_name=%(program_name)s_mysite%(process_num)02d
numprocs=3
autorestart=true
autostart=true