Python 多处理.如何将 XMLRPC ServerProxy 对象入队 [英] Python Multiprocessing. How to enqueue XMLRPC ServerProxy objects

问题描述

我正在尝试向 XMLRPC 服务器(mosesserver).要启动 XMLRPC 调用，需要一个 ServerProxy 对象(该对象包含服务器的 URL 等)

I am trying to send several parallel requests to a XMLRPC server (mosesserver). To launch the XMLRPC call, a ServerProxy object is needed (this object contains the URL of the server, among other things)

在串行执行中，我可以在程序开始时创建这个对象(通过调用 xmlrpclib.ServerProxy(server_URL))，将它存储在一个变量中，并在需要时使用它.

In serial execution, I can create this object at the beginning of the program (with a call to xmlrpclib.ServerProxy(server_URL)), store it in a variable and use it whenever I need it.

但是当使用进程池时，每个进程都需要这个对象的不同实例.每个进程的任务是从输入中读取一行并通过 XMLRPC 将其发送到翻译服务器，等待结果，获取另一行，再次发送......

But when using a pool of processes, each process needs a different instance of this object. The task of each process is to read a line from an input and send it to the translation server via XMLRPC, wait for the result, get another line, send it again ...

很容易看出，每个进程每次调用 XMLRPC 时都会创建一个新的 ServerProxy 实例，但问题是第一次调用比使用相同 ServerProxy 对象的后续调用花费的时间更长.所以我想要做的是，每个进程在第一次进行 RPC 调用时创建一个 ServerProxy 对象，然后它为以下调用重用相同的 ServerProxy 对象(它在串行执行中完成的方式).但是我在存储该对象时遇到了问题.

It is easy to get that each process creates a new ServerProxy instance each time it calls XMLRPC, but the problem is that the first call takes longer than subsequent calls using the same ServerProxy object. So what I am trying to do is that each process creates a ServerProxy object the first time it makes a RPC call, and then it reuses that same ServerProxy object for the following calls (the way it is done in serial execution). But I am having problems to store that object.

import xmlrpclib
import sys
import os

import multiprocessing
import time

def func(dummy_argument,que):
  server = xmlrpclib.ServerProxy('http://myserver:6060/RPC2',)
#  server="HELLO"
  que.put(server)
  return

def run_parallel(line,num_procesos):
  pool = multiprocessing.Pool(processes=num_procesos)
  m = multiprocessing.Manager()
  q = m.Queue()
  for i in range (0, num_procesos):
    pool.apply_async(func, ("dummy",q))

  while not q.empty():
    print q.get()

  pool.close()
  pool.join()

  return

if __name__ == '__main__':
  line="test"
  run_parallel(line,4)

在上面的代码中，如果我取消注释server =HELLO""行，一切似乎都正常，但是当我删除该行时，必须将 ServerProxy 对象server"加入队列，我收到以下错误:

In the code above if I uncomment the line 'server = "HELLO"', everything seems to work fine, but when I remove that line, so the ServerProxy object 'server' has to be enqueued, I receive the following error:

Exception in thread Thread-3:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 551, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.7/threading.py", line 504, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 353, in _handle_results
    task = get()
TypeError: ('__init__() takes exactly 3 arguments (1 given)', <class 'xmlrpclib.Fault'>, ())

它所指的init方法，是ApplyResult类(类其实例由 Pool.apply_async())

where the init method it refers to, is the constructor of the ApplyResult class (class whose instances are returned by Pool.apply_async())

事实是，即使我能够正确地存储和检索这个对象，我也不确定我是否能够按照我想要的方式使用它，但至少我想尝试一下.

The truth is that, even if I am able to store and retrieve this object correctly, I am not sure if I wil be able to use it in the way I intend, but at least I would like to try.

有人知道问题出在哪里吗?

Anyone know where is the problem?

或者也许有另一种(可能更简单)的方法来做我需要的事情?

Or perhaps is there another (probably simpler) way to do what I need?

谢谢

推荐答案

如果我明白，你的问题是如何在每个子进程中只初始化一次代理.您可以利用 multiprocessing.Pool() 的 initializer 参数，以及每个进程都有自己的全局变量这一事实:

If I understood, your question is how to initialize the proxy only once in each subprocess. You can take advantage of the initializer argument of multiprocessing.Pool(), and the fact that each process has its own global variables:

import xmlrpclib
import multiprocessing

def init():
    global server
    server = xmlrpclib.ServerProxy('http://myserver:6060/RPC2',)

def func(argument):
    result = server.some_function(argument)
    return result

def run_parallel(line,num_procesos):
    pool = multiprocessing.Pool(processes=num_procesos, initializer=init)
    async_results = [pool.apply_async(func, ("dummy",q)) 
                     for i in range (0, num_procesos)]
    for r in async_results:
        print r.get()

    pool.close()
    pool.join()

if __name__ == '__main__':
  line="test"
  run_parallel(line,4)

我还在那里演示了处理 apply_async 结果的典型方法.

I'm also demonstrating a typical way of handling apply_async results there.

这篇关于Python 多处理.如何将 XMLRPC ServerProxy 对象入队的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！