Snippets

Qi Peng 用MongoDB做后端实现的队列MongoQueue测试

Created by Qi Peng
# coding: utf-8
# mongoqueue-test.py

import time
import threading
from pymongo import MongoClient, ASCENDING
# https://github.com/pengqi/mongoqueue/blob/master/mongoqueue/mongoqueue.py
from mongoqueue import MongoQueue


def mongoqueue_test():
    """测试mongoqueue性能
    https://pypi.org/project/mongoqueue
    """
    collection = MongoClient()['test']['tasks']
    queue = MongoQueue(collection, consumer_id=0)
    
    # 删除队列里已有数据
    queue.clear()
    
    # 添加10w条任务
    start_time = time.time()
    for i in range(100000):
        queue.put({'url': 'http://www.test.com/page-{}.html'.format(i)})
    end_time = time.time()
    print 'Add tasks time used: {} seconds.'.format(end_time - start_time)
                       
    def thead_work():
        """线程函数
        """   
        while True:
            # 获取任务
            job = queue.next()
            if job:
                # 处理任务
                # ...
                # 标记完成任务
                job.complete()
            else:
                # 没有任务了
                break

    # 启动10个工作线程
    start_time = time.time()
    threads = []
    for i in range(10):
        threads.append(threading.Thread(target=thead_work))
    # 启动线程     
    for thread in threads: 
        thread.start()
    # 等待线程结束 
    for thread in threads: 
        thread.join()
    end_time = time.time()
    print 'Get tasks time used: {} seconds.'.format(end_time - start_time)    

        
if __name__ == '__main__':
    mongoqueue_test()
    

# 一次测试结果
# > python mongoqueue-test.py
# Add tasks time used: 34.3849999905 seconds.
# Get tasks time used: 79.1459999084 seconds.

Comments (0)