6

我有一个连接到许多 serverlet 之一的 Web 服务器。Web 服务器最多可以排队 40 个作业，每个作业可能需要 20 分钟或 30 小时才能运行。

Web 服务器使用套接字连接到 serverlet，并且 serverlet 运行使用线程发送的作业。

我想对可以一次运行的线程（作业）数量设置一个上限，比如 3 个，一旦达到该限制，它就会保留主线程。当其中一个线程结束时，它允许主线程继续并拾取另一项工作。

# Wait for thread count to reduce before continuing
while threading.active_count() >= self.max_threads:
    pass

我目前正在使用循环让我的主线程等到有空闲线程可用。它有效，但感觉像是一个快速而肮脏的解决方案。我想知道是否有更好的方法来做到这一点？

服务器.py

import socket
import sys
import urllib, urllib2
import threading
import cPickle

from supply import supply


class supply_thread(threading.Thread):

    def __init__(self, _sock):
        threading.Thread.__init__(self)
        self.__socket = _sock

    def run(self):
        data = self.readline()
        self.__socket.close()
        new_supply = supply.supply(data)
        new_supply.run()

    def readline(self):
        """ read data sent from webserver and decode it """

        data = self.__socket.recv( 1024 )
        if data:
            data = cPickle.loads(data)
            return data



class server:

    def __init__(self):
        ## Socket Vars
        self.__socket = None
        self.HOST = ''
        self.PORT = 50007
        self.name = socket.gethostname()

        self.max_jobs = 3


    def listen(self):
        """ Listen for a connection from the webserver """

        self.__socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # Allows quick connection from the same address
        self.__socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        self.__socket.bind((self.HOST, self.PORT))
        return self.__socket.listen(1)

    def connect(self):
        webserver = self.__socket.accept()[0]
        print 'Connected by', webserver

        new_thread = supply_thread(webserver)
        print 'Starting thread' , new_thread.getName()

        new_thread.start()

    def close(self):
        return self.__socket.close()


    def run(self):
        import time

        while True:
            print(sys.version)

            # Wait for connection from Webserver
            self.listen()

            time.sleep(3)

            # Let the Webserver know I'm avilable
            self.status(status='Available')

            print 'Waiting for connection...'
            self.connect()

            print 'thread count:', threading.enumerate()
            print 'thread count:', threading.active_count()

            while threading.active_count() >= self.max_jobs:
                pass


    def status(self, status='Available'):
        computer_name = socket.gethostname()
        svcURL = "http://localhost:8000/init/default/server"
        params = {
            'computer_name':computer_name,
            'status':status,
            'max_jobs':self.max_jobs
        }
        svcHandle = urllib2.urlopen(svcURL, urllib.urlencode(params))
4

1 回答 1

5

这听起来像是Celery的一个很好的用例。

基本上，您会在文件中创建一个 Celery 任务，tasks.py然后使用taskname.delay(). 如果工人准备好接受另一个任务，它会将任务分派给芹菜工人并开始处理它。这是docs 中的一个示例。

默认情况下，根据文档，Celery 将创建一个具有等于您机器中的核心数的工作线程。如果需要，您可以更改它。

或者，您可以使用Queues。这是另一个可能看起来如何的示例。

于 2013-05-10T04:07:04.257 回答