我正在编写一个小型的多线程http文件下载器,希望能够在代码遇到错误时缩小可用线程。
这些错误将特定于返回的http错误,其中Web服务器不允许更多连接。
例如,如果我设置了一个5个线程的池,每个线程都试图打开自己的连接并下载文件的一部分。服务器可能只允许2个连接,并且会返回503错误。我想检测到这一点并关闭一个线程,最终将池的大小限制为服务器允许的仅2个线程。
我可以让一个线程停止自己吗?
self.Thread_stop()足够吗?
我还需要join()吗?
这是我的工作类,它执行下载操作,从队列中获取要处理的数据,一旦下载完成,将结果倒入resultQ以便主线程保存到文件中。
在这里,我想检测http 503并从可用池中停止/删除线程 - 当然,还要将失败的块重新添加到队列中,以便剩余的线程将其处理。
这些错误将特定于返回的http错误,其中Web服务器不允许更多连接。
例如,如果我设置了一个5个线程的池,每个线程都试图打开自己的连接并下载文件的一部分。服务器可能只允许2个连接,并且会返回503错误。我想检测到这一点并关闭一个线程,最终将池的大小限制为服务器允许的仅2个线程。
我可以让一个线程停止自己吗?
self.Thread_stop()足够吗?
我还需要join()吗?
这是我的工作类,它执行下载操作,从队列中获取要处理的数据,一旦下载完成,将结果倒入resultQ以便主线程保存到文件中。
在这里,我想检测http 503并从可用池中停止/删除线程 - 当然,还要将失败的块重新添加到队列中,以便剩余的线程将其处理。
class Downloader(threading.Thread):
def __init__(self, queue, resultQ, file_name):
threading.Thread.__init__(self)
self.workQ = queue
self.resultQ = resultQ
self.file_name = file_name
def run(self):
while True:
block_num, url, start, length = self.workQ.get()
print 'Starting Queue #: %s' % block_num
print start
print length
#Download the file
self.download_file(url, start, length)
#Tell queue that this task is done
print 'Queue #: %s finished' % block_num
self.workQ.task_done()
def download_file(self, url, start, length):
request = urllib2.Request(url, None, headers)
if length == 0:
return None
request.add_header('Range', 'bytes=%d-%d' % (start, start + length))
while 1:
try:
data = urllib2.urlopen(request)
except urllib2.URLError, u:
print "Connection did not start with", u
else:
break
chunk = ''
block_size = 1024
remaining_blocks = length
while remaining_blocks > 0:
if remaining_blocks >= block_size:
fetch_size = block_size
else:
fetch_size = int(remaining_blocks)
try:
data_block = data.read(fetch_size)
if len(data_block) == 0:
print "Connection: [TESTING]: 0 sized block" + \
" fetched."
if len(data_block) != fetch_size:
print "Connection: len(data_block) != length" + \
", but continuing anyway."
self.run()
return
except socket.timeout, s:
print "Connection timed out with", s
self.run()
return
remaining_blocks -= fetch_size
chunk += data_block
resultQ.put([start, chunk])
以下是我初始化线程池的地方,稍后我将项目放入队列中。
# create a thread pool and give them a queue
for i in range(num_threads):
t = Downloader(workQ, resultQ, file_name)
t.setDaemon(True)
t.start()