python中的多线程多进程

太过爱你忘了你带给我的痛 2022-05-30 01:16 332阅读 0赞

### 并发编程 ###

*  并发  
    
    
     *  同时有很多事要做，可以并行处理也可以串行处理
 *  并行
    
     *  同时做多件事
    
    #### 并发模块 ####
    
     *  threading模块  
        
        
         *  线程也叫轻量级进程，它是一个基本的CPU单元，也是程序执行过程中的最小单位。
     *  multiprocessing模块  
        
        
         *  进程就是一个程序在一个数据集上的一次动态执行的过程。进程一般有程序、数据集、进程控制块三部分组成。
     *  进程和线程的关系  
        
        
         *  线程是属于进程的
         *  线程是运行在进程空间内
         *  进程运行时会有系统资源的调度，而进程不需要调度资源，因此进程的启动会比线程快。
    
    #### 基于线程的并行-关键函数 ####
    
     *  创建线程的两种方法  
        
        
         *  创建一个函数
         *  继承thread类
         *  Thread是threading模块中最重要的类之一，可以使用它来创建一个线程
     *  函数调用线程
     *  构造方法：  
        Tread(group = None, target = None,name=None,args=(),kwargs=\{\})  
        
        
         *  group:线程组，目前还没有实现，库引用中提示必须是None
         *  target:要执行的方法
         *  name:线程名
         *  args/kwargs:要传入方法的参数
     *  thread的方法说明名  
        
        
         *  isAlive():返回线程是否在运行。正在运行指启动后、终止前。
         *  get/setName(name):获取/设置线程名
         *  t.start() 启动线程
         *  t.join() 逐个执行每个线程，执行完毕后继续往下执行  
            另一种说法 join(\[timeout\]):阻塞当前上下文环境的线程，知道调用此方法的线程终止或达到指定的timeout.（可选参数）
         *  t.run() 线程被cpu调度后，自动执行线程对象的run方法
         *  Daemon后台线程：默认False
         *  (其他论坛版本)is/setDaemon(bool):获取/设置是后台线程（默认是前台线程(False））。（在start前设置）  
            
            
             *  如果后台线程，主线程执行过程中，后台线程也在进行，主线程完毕后，后台线程不论成功与否，主线程和后台线程均停止
             *  如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止
    
    ##### 创建线程-创建函数 #####
    
        import threading
        import time
        
        def counter(n):
            cnt = 0;
            for i in xrange(n):
                cut += 1
                time.sleep(0.1)
                print cnt
        
        th = threading.Thread(target = counter, args=(10,));
        th.start()
        th.join()
        print 'main thread task done'
    
    ##### 创建线程-继承类 #####
    
        import threading, time, random
        
        def counter():
            cnt = 0;
            for i in xrange(10000):
                for j in xrange(i):
                    cnt += j;
        
        class SubThread(threading.Thread):
            def __init__(self,name):
                threading.Thread.__init__(self,name=name);
                    pass
        
            def run(self):
                i = 0;
                while i < 3:
                    print self.name, 'counting...\n';
                    counter()
                    print self.name,'finish\n'
                    i += 1;
        
        th = SubThread('thread-1');
        th.start();
        th.join();
        print 'all done'

--------------------

th.setDaemon(False)   ---- 表示令th这个子进程为不重要进程，
      那么主进程就不必等待这个th子进程执行完才去继续执行，可以在th没有执行完主进程
      就开始执行。

--------------------

##### threading模块提供的线程同步原语包括： #####

*  Lock
 *  RLock
 *  Condition
 *  Event
 *  Semaphore等对象
    
    ![RLock][]
    
    acquire(wait=True) 此方法用来请求资源，如果资源没有被占用就调用资源且将资源设为被占用，如果资源已经被调用，那么使用此方法的线程就只能阻塞等待  
    lock()对象， 用于判断这个资源是否上锁  
    release() 这个方法是用于线程把占用的资源释放掉

--------------------

##### 线程同步(*增加的为同步锁*) #####

from threading import Lock, Thread
    # lock = Lock
    some_var = 0
    class IncrementThread(Thread):
        def run(self):
            global some_var
            read_value = some_var
            print "some_var in %s is %d" %(self.name, read_value)
            # lock.acquire()
            some_var = read_value + 1
    
    
    def use_increment_thread():
        threads = []
        for i in range(50):
            t = IncrementThread()
            threads.append(t)
            t.start()
        for t in threads:
            t.join()
        print "After 50 modifications, some_var should have become 50"
        print "After 50 modification, some_var is %d" % (some_var,)
    use_increment_thread()

--------------------

###### 例：不加锁计数器（加锁保证） ######

import time
    from threading import Thread, Lock
    value = 0
    lcok = Lock()
    def getlock():
        # with lock:
        gloabal value
        new = value + 1
        time.sleep(0.001)
        value = new
    threads = []
    for i in range(1000):
        t = Thread(target=getlock)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    print value

###### 当资源为外部资源时 ######

import threading
    import time
    
    def test_xc():
        mutex.acquire() #取得锁
        f = open("text.txt", "a")
        f.write("text_dxc" + '\n')
        f.close()
        mutex.release()  # 释放锁
    
    mutex = threading.lock() #创建锁
    threads = []
    for i in xrange(5):
        t = threading.Thread(target=text_xc)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()

如果没有这个线程锁会出现什么情况

#### 可重用锁 ####

![可重用锁][70]

![同步机制][70 1]

items\_cv.notify() 的作用是通知等待的进程，将其唤醒,例子如下:

import threading, time
    class Hider(threading.Thread):
        def __init__(self, cond, name):
            super(Hider, self).__init__()
                self.cond = cond
                self.name = name
        def run(self):
            time.sleep(1)  # 确保先运行Seeker中的方法
            self.cond.acquire()
            print self.name + ': 我已经藏好了'
            self.cond.notify()
            self.cond.wait()
    
            print self.name + ':  被你找到了,哎~~~'
            self.cond.notify()
            self.cond.release()
    
    class Seeker(threading.Thread):
        def __init__(self,cond,name):
            super(Seeker, self).__init__()
            self.cond = cond
            self.name = name
        def run(self):
    
            self.cond.acquire()
            self.cond.wait()  # 释放对锁的占用，同时线程挂起在这里，直到被notify并继续执行
    
            print self.name + ':我已经蒙好眼了，你藏吧~ '
            self.cond.notify()
            self.cond.wait()
    
            self.cond.release()
            print self.name + '：我赢了'
    cond = threding.Condition()
    seeker = Seeker(cond, 'seeker')
    hider = Hider(cond, 'hider')
    seeker.start()
    hider.start()

--------------------

##### 同步机制-Event #####

*  基于事件的同步机制  
    
    
     *  一个线程发送/传递时间，另一个线程等待时间的触发。
 *  简单的线程通信机制  
    
    
     *  他提供了设置信号，清除信号，等待等实现线程间的通信
 *  工作原理  
    
    
     *  Event对象和条件标记类似，允许线程等待某个事件发生。
     *  初始状态时事件被设置为0
     *  如果事件没有被设置而线程正在等待该事件，那么线程就会被阻塞直到事件被设置为止
     *  当有线程设置了这个事件时就会唤醒所有正在等待该事件的线程
     *  如果线程等待的该事件已经设置了，那么线程会继续执行

###### 交通灯 ######

improt threading
    import random
    import time
    
    class VehicleThread(threading.Thread):
        """Class representing a motor vehicle at an intersection"""
    
        def __init__(self, threadName, event):
            """Initializes thread"""
    
            threading.Thread.__init__(self, name=threadName)
            self.threadEvent = event
    
        def run(self):
            """Vehicle waits unless/until light is green"""
            time.sleep(random.randrange(1,10))
            # print arrival time of car at intersection
            print "%s arrived at %s\n" % \
                (self.getName(), time.ctime(time.time())
    
            # wait for green light
            self.threadEvent.wait()
    
            # displays time that car departs intersection
            print "%s passes through intersection at %s\n" %\
                (self.getName(), time.ctime(time.time()))
    
    greenLight = threading.Event()
    vehicleThreads = []
    
    # creates and starts five Vehicle threads
    for i in rang(1, 5):
        vehicleThreads.append(VehicleThread("Vehicle" + str(i), greenLight))
    
    for vehicle in vehicleThreads:
        vehicle.starts()
    
    while threading.activeCount() >1:
        # sets the Event's flag to false -- block all incoming vehicles
        greenLight.clear()
        print "RED LiGHT! at", time.ctime(time.time())
        time.sleep(3)
    
        # sets the Event's flag to true -- awaken all waiting vehicles
        print "GREEN LIGHT! at", time.ctime(time.time())
        greenLight.set()
        time.sleep(1)

--------------------

##### threading模块中常用的方法 #####

*  thread.currentThread() :返回当前的线程变量
 *  thread.enumerate():返回一个包含正在运行的线程的list。正在运行指程序启动后、结束前，不包括启动前和终止后的线程。
 *  thread.activeCount():返回正在运行的线程数量，与len(thread.enumearte())由相同结果

##### thread模块提供的常量 #####

*  thread.TIMEOUT\_MAX设置thread全局超时时间

##### 同步机制-信号量 #####

*  工作原理  
    
    
     *  信号量通过一个计数器控制对共享资源的访问
     *  当此线程不再需要访问共享资源时
     *  它释放该通行证，这导致信号量的计数递增
     *  如果另一个线程等待通行证，则那个线程将在那时获得通行证
 *  操作：
    
     *  每当调用acquire()时，内置计数器-1
     *  每当调用release()时，内置计数器+1
        
            import time
            from random import random
            from threading import Thread, Semaphore
            sema = Semaphore(3)
            def foo(tid):
                with sema:   # 每次一个线程运行了with语句信号量就会减一
                    print '{} acquire sema'.format(tid)
                    wt = random()*2
                    time.sleep(wt)
                print '{} release sema'.format(tid)
            threads = []
            for i in range(5):
                t = Thread(target=foo, args = (i,))
                threads.append(t)
                t.start()   
            fort t in threads:
                t.join()

![GIL][]

#### 基于进程的并行-简介 ####

*  multiprocessing模块  
    
    
     *  Python中的多进程管理包
     *  multiprocessing包中也有Lock/Event/Semaphore/Condition类
 *  用途  
    
    
     *  Python设计的限制
     *  最多只能用满1个CPU核心
     *  借助各个包，可以轻松的完成从单进程到并发执行的转换。

#### 基于进程的并行-创建 ####

*  可以轻松完成从单进程到多进程并发执行的转换
 *  multiprocessing支持子进程、通信和共享数据、执行不同形式的同步
 *  提供了Poccess、Queue、Pipe、Lock等组件。
 *  Process(\[group \[,target,\[name \[, args \[, kwargs\]\]\]\]\])  
    
    
     *  target表示调用对象，可以传入方法的名字
     *  args表示被调用对象的位置参数元组
     *  kwargs表示调用对象的字典
     *  name是别名, 相当于给这个进程取一个名字
     *  group分组， 实际上不使用

--------------------

创建进程

import multiprocessing
    
    def foo(i):
        print 'called funtion in process: %s' % i
        return 
    
    Process_jobs = []
    for i in range(5):
        p = multiprocessing.Process(target = foo, args=(i,))
        Process_jobs.append(p)
        p.start()
        p.join()

##### 后台进程 #####

##### 继承创建进程 #####

##### 进程间的数据共享 #####

*  共享内存
    
        from multiprocessing import Process,  Value, Array
        
        def f(n, a):
            n.value = 3.1415927
            for i in range(len(a)):
                a[i] = -a[i]
        
        if __name__ ==  '__main__':
            num = Value('d', 0.0)
            arr = Array('i', range(10))
        
            p = Process(target=f, args=(num, arr))
            p.start()
            p.join()
        
            print num.value
            print arr[:]
 *  进程共享队列
    
     *  消息队列  
        
        
         *  提供了一种从一个进程向另一个进程发送数据块的方法。
         *  通过发送消息来避免命名管道的同步和阻塞问题。
     *  multiprocessing当中的Queue使用方式
        
         *  put操作，将对象放入Queue （放可以放多个）
         *  get操作，将对象从Queue当中读出 （拿却只能拿一次）
         *  Queue对象负责进程之间的对象传输
            
                import  multprocessing
                import random
                import time
                class producer(multiprocessing.Process):
                    def __init__(self. queue):
                        multiprocessing.Process.__init__(self)
                        self.queue = queue
                    def run(self):
                        for i in range(10):
                            item =random.randint(0,256)
                            self.queue.put(item)
                            print ("Process Producer : item %d appended to queue %s" %　(item, self.name)
                            time.sleep(1)
                            print ("The size of queue is %s" % self.queue.qsize())
                class consumer(multiprocessing.Process):
                    def __init__(self, queue):
                        multiprocessing.Process.__init__(self)
                        self.queue = queue
                    def run(self):
                        while True:
                            if (self.queue.empty()):
                                print("the queue is empty")
                                break
                            else:
                                time.sleep(2)
                                item = self.queue.get()
                                print ('Process Consumer : item %d popped from by %s  \n'  %(item,self.name)
                        time.sleep(1)
 *  管道通信
    
     *  管道，顾名思义，一端发一端收。
     *  Pipe可以单向（half-duplex）,也可以是双向（duplex）
     *  我们通过multiprocessing.Pipe(duplex=False)创建单向管道（默认为双向）
     *  一个进程从PIPE一端输入对象，然后被PIPE另一端的进程接收。
     *  单项管道只允许管道一端的进程输入
     *  双向管道则允许从两端输入
        
            from multiprocessing improt Process,Pipe
            
            class Consumer(Process):
                def __init__(self,pipe):
                    Process.__init__(self)
                    self.pipe = pipe
            
                def run(self):
                    self.pipe.send("Consumer Words")
                    print "Consumer Recevived:", self.pipe.recv()
            
            class Producer(Process):
                def __init__(self,pipe):
                    Process.__init__(self)
                    self.pipe = pipe
            
                    def run(self):
                        print "Producer Received:", self.pipe.recv()
                        self.pipe.send("Producer Words")
            
            pipe = pipe()
            p =　Producer(pip[0])
            c = Consumer(pip[1])
            p.daemon = c.daemon = True
            p.start()
            c.start()
            p.join()
            c.join()
            print "Ended!"

![进程池][70 2]

#阻塞方式
    from multiprocessiong import Lock, Pool
    import t
    
    def function(index):
        print "Start Process", index
        time.sleep(3)
        print "End Process", index
    
    pool = Pool(processes=3)
    for i in xrange(4):
        pool.apply(function,(i,))
        # pool.apply_async(function,(i,))  非阻塞方式
    
    print "Started process"
    pool.close()
    pool.join()
    print "Subprocess done"

![协程][70 3]

[RLock]: /images/20220530/554115ec24544e5a88bf590a4fdc7f30.png
[70]: /images/20220530/1997b9910d6a41d196ebb61fca7346a2.png
[70 1]: /images/20220530/2be11ae879c64178ab7ab79859aae83e.png
[GIL]: /images/20220530/8dfb845e60624f2b93cfe9561224e8e3.png
[70 2]: /images/20220530/d4fbbaed713642c7bc7a0148114d8eff.png
[70 3]: /images/20220530/08809047e1e241b8a876a4cb8ec8d74e.png