Ansible源码分析之Python的multiprocessing模块使用

21312

最近在研究Python的multiprocessing模块，顺便去了解下Ansible是如何实现并发执行命令的。
在Ansible目录下搜索multiprocessing
cd /usr/lib/python2.6/site-packages/ansible
find -exec grep -l multiprocessing {} \;

只有./runner/__init__.py这个文件里面引用了multiprocessing模块

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

def _parallel_exec(self, hosts):         ''' handles mulitprocessing when more than 1 fork is required '''         manager = multiprocessing.Manager()         job_queue = manager.Queue()         for host in hosts:             job_queue.put(host)         result_queue = manager.Queue()         try:             fileno = sys.stdin.fileno()         except ValueError:             fileno = None         workers = []         for i in range(self.forks):             new_stdin = None             if fileno is not None:                 try:                     new_stdin = os.fdopen(os.dup(fileno))                 except OSError, e:                     # couldn't dupe stdin, most likely because it's                     # not a valid file descriptor, so we just rely on                     # using the one that was passed in                     pass             prc = multiprocessing.Process(target=_executor_hook,                 args=(job_queue, result_queue, new_stdin))             prc.start()             workers.append(prc)         try:             for worker in workers:                 worker.join()         except KeyboardInterrupt:             for worker in workers:                 worker.terminate()                 worker.join()         results = []         try:             while not result_queue.empty():                 results.append(result_queue.get(block=False))         except socket.error:             raise errors.AnsibleError("<interrupted>")         return results     # *****************************************************

Ansible能够同时对成百上千的服务器同时执行操作，都是通过这个函数实现的，使用Python的多进程模块来提升效率
首先定义一个