use threadpool instead of threading

This commit is contained in:
ricterz 2015-05-10 18:04:07 +08:00
parent e8d38fe820
commit aa354f23c9
3 changed files with 20 additions and 33 deletions

View File

@ -1,4 +1,3 @@
import Queue
from constant import DETAIL_URL, IMAGE_URL from constant import DETAIL_URL, IMAGE_URL
from logger import logger from logger import logger
@ -27,9 +26,9 @@ class Dojinshi(object):
def download(self): def download(self):
logger.info('Start download dojinshi: %s' % self.name) logger.info('Start download dojinshi: %s' % self.name)
if self.downloader: if self.downloader:
download_queue = Queue.Queue() download_queue = []
for i in xrange(1, self.pages + 1): for i in xrange(1, self.pages + 1):
download_queue.put('%s/%d/%d.%s' % (IMAGE_URL, int(self.img_id), i, self.ext)) download_queue.append('%s/%d/%d.%s' % (IMAGE_URL, int(self.img_id), i, self.ext))
self.downloader.download(download_queue, self.id) self.downloader.download(download_queue, self.id)
else: else:
logger.critical('Downloader has not be loaded') logger.critical('Downloader has not be loaded')

View File

@ -5,6 +5,7 @@ import socket
import threading import threading
import Queue import Queue
import requests import requests
import threadpool
from urlparse import urlparse from urlparse import urlparse
from logger import logger from logger import logger
@ -36,9 +37,10 @@ class Downloader(object):
try: try:
os.mkdir(folder) os.mkdir(folder)
except os.error, e: except os.error, e:
logger.error('%s error %s' % (threading.currentThread().getName(), str(e))) logger.critical('Error: %s' % str(e))
sys.exit() sys.exit()
logger.info('Start downloading: %s ...' % url)
filename = filename if filename else os.path.basename(urlparse(url).path) filename = filename if filename else os.path.basename(urlparse(url).path)
try: try:
with open(os.path.join(folder, filename), "wb") as f: with open(os.path.join(folder, filename), "wb") as f:
@ -50,23 +52,14 @@ class Downloader(object):
for chunk in response.iter_content(2048): for chunk in response.iter_content(2048):
f.write(chunk) f.write(chunk)
except (os.error, IOError), e: except (os.error, IOError), e:
logger.error('%s error %s' % (threading.currentThread().getName(), str(e))) logger.critical('Error: %s' % str(e))
sys.exit() sys.exit()
except Exception, e: except Exception, e:
raise e raise e
logger.info('%s %s downloaded.' % (threading.currentThread().getName(), url)) return url
def _download_thread(self, queue, folder=''): def _download_callback(self, request, result):
while not self.kill_received: logger.log(15, '%s download successfully' % result)
if queue.empty():
queue.task_done()
break
try:
url = queue.get(False)
logger.info('%s downloading: %s ...' % (threading.currentThread().getName(), url))
self._download(url, folder)
except Queue.Empty:
break
def download(self, queue, folder=''): def download(self, queue, folder=''):
if not isinstance(folder, (str, unicode)): if not isinstance(folder, (str, unicode)):
@ -80,20 +73,14 @@ class Downloader(object):
else: else:
logger.warn('Path \'%s\' not exist' % folder) logger.warn('Path \'%s\' not exist' % folder)
for i in range(self.thread_count): queue = [([url], {'folder': folder}) for url in queue]
_ = threading.Thread(target=self._download_thread, args=(queue, folder, ))
_.setDaemon(True)
self.threads.append(_)
for thread in self.threads: self.thread_pool = threadpool.ThreadPool(self.thread_count)
thread.start() requests_ = threadpool.makeRequests(self._download, queue, self._download_callback)
[self.thread_pool.putRequest(req) for req in requests_]
while len(self.threads) > 0:
try: try:
self.threads = [t.join(THREAD_TIMEOUT) for t in self.threads if t and t.isAlive()] self.thread_pool.wait()
except KeyboardInterrupt: except KeyboardInterrupt:
logger.warning('Ctrl-C received, sending kill signal.') print
self.kill_received = True logger.error('Ctrl-C pressed, exiting threads ...')
# clean threads list
self.threads = []

View File

@ -1,2 +1,3 @@
requests>=2.5.0 requests>=2.5.0
BeautifulSoup4>=4.0.0 BeautifulSoup4>=4.0.0
threadpool>=1.2.7