快捷導(dǎo)航

Python實(shí)現(xiàn)的多進(jìn)程和多線程功能示例

更新時(shí)間：2018年05月29日 11:20:47 作者：荷蘭人號(hào)

這篇文章主要介紹了Python實(shí)現(xiàn)的多進(jìn)程和多線程功能,結(jié)合實(shí)例形式分析了Python多線程與多進(jìn)程實(shí)現(xiàn)分布式系統(tǒng)功能相關(guān)操作技巧,需要的朋友可以參考下

本文實(shí)例講述了Python實(shí)現(xiàn)的多進(jìn)程和多線程功能。分享給大家供大家參考，具體如下：

聽了朋友說起，他們目前開發(fā)的測試框架，用python實(shí)現(xiàn)的分布式系統(tǒng)。雖然python的執(zhí)行效率沒有c和c++那么高，但是依靠集群的力量，產(chǎn)生的壓力很是牛逼啊。

了解了下大概的方式就是

1、有臺(tái)主控機(jī)，負(fù)責(zé)調(diào)度，比如執(zhí)行的參數(shù)等

2、有n多臺(tái)執(zhí)行機(jī)，每個(gè)執(zhí)行機(jī)上部署一個(gè)python的xmlRPC server，主控機(jī)調(diào)用rpccall，然后執(zhí)行機(jī)執(zhí)行。rpccall里面會(huì)fork一些進(jìn)程，每個(gè)進(jìn)程再創(chuàng)建一些線程，去調(diào)用測試方法。這樣，擴(kuò)展性就很好了。

對(duì)于python的rpc call，之前也沒有接觸過，不是很了解，google了一下，發(fā)現(xiàn)很簡單，拿了個(gè)網(wǎng)上的例子，如下，先部署一個(gè)rpcServer

from SimpleXMLRPCServer import SimpleXMLRPCServer
def add(a , b): 
  return a+bserver = SimpleXMLRPCServer(("10.249.192.38", 8000))#這里不要用localhost，否則只有本機(jī)才能訪問
server.register_function(add)
server.serve_forever()

客戶端：

from xmlrpclib import Server
Proxyserver = ServerProxy("http://localhost:8000")
try: ret = server.add(30,90) print 'result:', ret print 'result type:', type(ret)
except
 Exception, e: print "exception",e

其實(shí)還是很簡單的。

然后再看了下python的多進(jìn)程和多線程的方法，寫了個(gè)例子，如下：

#encoding=utf-8
import sys
import os
import time
import pdb
import httplib
import thread
import threading
constant_p = 0 #創(chuàng)建全局變量，進(jìn)程數(shù)
constant_s = 0 #創(chuàng)建全局變量，線程數(shù)
mutex_g = threading.RLock() #創(chuàng)建全局鎖
def run(count):#該函數(shù)創(chuàng)建3個(gè)線程，同時(shí)調(diào)用3個(gè)不同的函數(shù)
  a = 1
  b = 0
  thread.start_new_thread(test0,(a,b))#這里傳入的參數(shù)需要以元組的形式，跟void指針其實(shí)也差不多
  thread.start_new_thread(test1,(a,b))
  thread.start_new_thread(test2,(a,b))
def test0(a,b):
  global mutex_g
  global constant_s
  threadid = thread.get_ident()
  mutex_g.acquire()#這里需要把線程數(shù)說鎖起來，否則結(jié)果會(huì)被修改
  constant_s = constant_s+1
  mutex_g.release()
  print "thread 0 called,and the threadid is:%d"%(threadid)
  sys.exit(0)
def test1(a,b):
  global mutex_g
  global constant_s
  threadid = thread.get_ident()
  mutex_g.acquire()
  constant_s = constant_s+1
  mutex_g.release()
  print "thread 1 called,and the threadid is:%d"%(threadid)
  sys.exit(0)
def test2(a,b):
  global mutex_g
  global constant_s
  threadid = thread.get_ident()
  mutex_g.acquire()
  constant_s = constant_s+1
  mutex_g.release()
  print "thread 2 called,and the threadid is:%d"%(threadid)
  sys.exit(0)
def my_fork():
  global constant_p
  pid = os.fork()#fork一個(gè)子進(jìn)程，子進(jìn)程的pid=0同時(shí)2個(gè)進(jìn)程會(huì)執(zhí)行my_fork()函數(shù)
  if (pid == 0):#子進(jìn)程執(zhí)行到這個(gè)if里面
    constant_p = constant_s + 1
    run(3)
    time.sleep(5)
    print "total thread is %d"%constant_s#這個(gè)結(jié)果是3，因?yàn)樽舆M(jìn)程創(chuàng)建按了3個(gè)線程
  elif (pid >0):#父進(jìn)程執(zhí)行到這個(gè)if里面
    constant_p = constant_s + 1 run(4)
    time.sleep(5)
    print "total thread is %d"%constant_s#這個(gè)結(jié)果也是3，因?yàn)樵撨M(jìn)程也創(chuàng)建了3個(gè)線程
  else:
    print "fork error"
    sys.exit(0)
  print "total process is %d"%constant_p#該結(jié)果是1,因?yàn)?個(gè)進(jìn)程只執(zhí)行了一次
  constant_p = constant_s + 1
  sys.exit(0)
if __name__ == "__main__":
  my_fork()
  #my_fork()
  #my_fork()

更多關(guān)于Python相關(guān)內(nèi)容感興趣的讀者可查看本站專題：《Python進(jìn)程與線程操作技巧總結(jié)》、《Python Socket編程技巧總結(jié)》、《Python數(shù)據(jù)結(jié)構(gòu)與算法教程》、《Python函數(shù)使用技巧總結(jié)》、《Python字符串操作技巧匯總》、《Python入門與進(jìn)階經(jīng)典教程》及《Python文件與目錄操作技巧匯總》

希望本文所述對(duì)大家Python程序設(shè)計(jì)有所幫助。

您可能感興趣的文章: