1

我们有一个异步单线程twisted/python 进程。有一个 X-server 监听端口 X。还有另一个服务器 Y-server 监听端口 Y。Y-server 也是 X-server 的客户端(执行 Y 请求涉及向 X 服务器传递请求) .

它们都应在一个线程中异步执行。它应该像下面这样工作:

  • Y-server 收到请求
  • Y-server 创建一个 X-client:打开一个到 X-server 的连接并发送一个内部请求
  • 同时(!!!) X-server 接受内部请求
  • X-server 完成它的工作并返回内部响应
  • X-client(属于 Y-server)接受内部响应并将其作为 Y-response 进一步发送

我试图实现这样的事情,但我失败了,可能是因为没有使用延迟。据我了解deferred,他们的工作是将上述序列分成更小的块,以便这些部分可以同时由X和Y完成。

我需要的是了解这种通信应该如何工作的方案。伪代码应该做...


以下是我失败尝试的简短代码。

有主要的服务类,它由协议/工厂类组成:

class PyCached(protocol.Protocol):
    def __init__(self, factory, verbose):
        self.factory = factory
        self.verbose = verbose
    def dataReceived(self, data):
        log.msg(data)
        if self.verbose:
            print 'received: %s' % (data,)
        request = json.loads(data)
        if self.verbose:
            print 'request: %s' % (request,)
        command = "handle_%s" % (request.pop('command'),)
        if self.verbose:
            print 'command: %s\n' % (command,)
        result = getattr(self.factory, command)(**request)
        self.transport.write(result + "\n")

class PyCachedFactory(protocol.Factory):
    def __init__(self, verbose=False):
        self.clear()
        self.start_time = datetime.now()
        self.verbose = verbose
        log.msg('starts on %s, verbose=%s' % (self.start_time, self.verbose))

    # many many more commands performed by factory

还有http访问服务器:

from twisted.web.resource import Resource
from twisted.python import log
from twisted.web.server import Site
from client import PyCachedClient

class PyCachedCommand(Resource):
    isLeaf = True

    def getServiceClient(self):
        client = PyCachedClient()
        client.connect(*self.service_address)
        return client

    def render_GET(self, request):
        '''
        Renders service status as plain text.
        '''
        log.msg('GET')
        request.setHeader('Content-Type', 'text/plain')
        try:
            client = self.getServiceClient()
            status = client.status()
            client.close()
            return "PyCached is up since %0.2f seconds" % (status['uptime'],)
        except:
            return "PyCached is down."

    def render_POST(self, request):
        '''
        Executes pycached request ad returns the response.
        '''
        log.msg('POST %s' % (str(request.args)))
        client = self.getServiceClient()
        kwargs = {k: v[0] for k,v in request.args.iteritems()}
        command_name = kwargs.pop('command')
        command = getattr(client, command_name)
        result = str(command(**kwargs))
        client.close()
        request.setHeader('Content-Type', 'text/plain')
        return result

class PyCachedSite(Site):
    '''
    Performs all operations for PyCached HTTP access.
    '''
    def __init__(self, service_address, **kwargs):
        resource = PyCachedCommand()
        resource.service_address = service_address
        Site.__init__(self, resource, **kwargs)

http使用主服务客户端,它是用简单的套接字实现的——这可能就是问题所在,因为这些客户端套接字调用是阻塞的:

class PyCachedClient(object):
    def __init__(self):
        self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    def connect(self, host, port):
        try:
            self.s.connect((host, port))
        except socket.error:
            raise RuntimeError('Something went wrong with PyCached.')

    def close(self):
        self.s.close()

    def _receive(self):
        received = self.s.recv(1024)
        decoded = json.loads(received.rstrip('\n'))
        return decoded['value'] if decoded.has_key('value') else None

    def _send(self, command, options={}):
        request = {'command': command}
        request.update(options)
        self.s.sendall(json.dumps(request))

    def version(self):
        self._send('version')
        return self._receive()

    # many many more commands similar to version

最后,一切都由一个 twistd/TAC 文件运行——所以它驻留在一个线程中:

from twisted.application import internet, service
from server.service import PyCachedFactory
from server.http import PyCachedSite

application = service.Application('pycached')
# pycached core service
pycachedService = internet.TCPServer(8001, PyCachedFactory())
pycachedService.setServiceParent(application)
# pycached http access
addr = ('localhost', 8001)
pycachedHttp = internet.TCPServer(8002, PyCachedSite(addr))
pycachedHttp.setServiceParent(application)

当我 telnet 8001(主要服务)时,例如{"command":"version"},一切正常。但是当我问 http 时,一切都阻塞了,因为客户端套接字被阻塞并且主服务永远无法响应。

4

1 回答 1

1

Deferred 只是一个管理回调的工具。X-client 必须提供一些方法来在收到结果时添加回调,您应该从这个回调中发送 Y-response。Deferred 只是一个实现细节。

于 2013-11-02T16:09:52.333 回答