多线程与多进程---方法对比与使用

多线程与多进程

 

1. 创建多线程和多进程
   1. 创建多线程

      方法一: 直接调用

   import threading, time  
       
       
   def foo(name):  
       time.sleep(1)  
       print("foo", name)  
       
       
   if __name__ == '__main__':  
       print("开始")  
       th = threading.Thread(target=foo, args=("test",))  
       th.start()  
       th.join()  
       print("end")  

    

    

   方法二:继承式调用

import threading, time  
1.     
2.     
3. class Foo(threading.Thread):  
4.     def __init__(self, name):  
5.         super().__init__()  #注意,这一行必须有,没有则会报错
6.         self.name = name  
7.     
8.     def run(self):  
9.         time.sleep(1)  
10.         print("foo", self.name)  
11.     
12.     
13. if __name__ == '__main__':  
14.     print("开始")  
15.     f = Foo("test")  
16.     f.start()  
17.     f.join()  
18.     print("end")  
1. 创建多进程

   方法一:直接调用

# coding=utf-8  

import multiprocessing, time, os  

    

    

def foo(name):  

    time.sleep(1)  

    print("foo", name, os.getpid())  

    

    

if __name__ == '__main__':  

    print("开始", os.getpid())  

    mp = multiprocessing.Process(target=foo, args=("test",))  

    mp.start()  

    mp.join()  

    print("end")  

    # 输入结果:  

    # 开始 10960  

    # foo test 3884  

    # end  

注意: args=("test",)这里是一个元组,如果只有一个参数则必须在后面加上一个逗号

在windows 下必须需要加上if __name__ == '__main__': 嗯,万恶的Windows

方法二:

import multiprocessing, time, os  
    
    
class Foo(multiprocessing.Process):  
    def __init__(self, name):  
        super().__init__()   #嗯,没错还是这一行,没有的话会报错的
        self.name = name  
    
    def run(self):  
        time.sleep(1)  
        print("foo", self.name, os.getpid())  
    
    
if __name__ == '__main__':  
    print("开始", os.getpid())  
    f = Foo("test")  
    f.start()  
    f.join()  
    print("end")  
# 输出结果:  
# 开始 1840  
# foo test 8968  
# end  

 

2. 进程与线程的一些方法:

1..进程:

    预先在Ipython中输入:

In [9]: import threading, time ...: class Foo(threading.Thread): ...: def __init__(self, age): #这里先将name换掉 ...: super().__init__() ...: self.name = age ...: def run(self): ...: time.sleep(1) ...: print("foo", self.age) In [10]: f = Foo("test")

这里是为什么暂时先不用name, 这里使用name,则线程名就是传入的

方法

 

getName() name	得到线程名	In [16]: f.getName() Out[16]: 'Thread-937' In [17]: f.name Out[17]: 'Thread-937'
setName()	给线程设置线程名	In [17]: f.setName("jjjj") In [18]: f.getName() Out[18]: 'jjjj'
is_alive()	是否是活动线程	/
isDaemon() setDaemon()	设置setDaemon为true时,执行完后,主进程没有执行完会阻塞,主进程先执行完,则子进程会在主进程执行完之后KILL掉	In [10]: f.setDaemon(True) In [11]: f.isDaemon() Out[11]: True
join(time)	等待至线程中止。这阻塞调用线程直至线程的join() 方法被调用中止-正常退出或者抛出未处理的异常-或者是可选的超时发生	/
start()	启动线程活动。	/
daemon f.daemon=True	设置或获得daemon	/

三个threading的方法

threading.enumerate()	返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。	In [27]: threading.enumerate() Out[27]: [<HistorySavingThread(IPythonHistorySavingThread, started 5048)>, <_MainThread(MainThread, started 4028)>]
threading.activeCount()	返回正在运行的线程数量，与len(threading.enumerate())有相同的结果	In [31]: threading.activeCount() Out[31]: 2
threading.currentThread()	返回当前的线程变量。	In [32]: threading.current_thread() Out[32]: <_MainThread(MainThread, started 4028)>

2 进程

is_alive()	是否是活动进程	/
join()	阻塞当前上下文环境的进程程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。	/
start()	进程准备就绪，等待CPU调度	/
daemon	同线程	In [13]: f.daemon=True In [14]: f.daemon Out[14]: True
name	命名	In [20]: f.name="123412"   In [21]: f.name Out[21]: '123412'
pid	等于 os.getpid()	/
terminate()	不管任务是否完成，立即停止工作进程	/

1. 锁

   1线程.

   1. 同步锁: threading.Lock()

lock = threading.Lock() #创建一个线程锁对象

lock.acquire() #加锁,并且返回一个布尔类型, 不可多次acquire,会阻塞住

lock.release() #去锁,返回值为NONE

 

 

1. 死锁,和递归锁

在同时存在两个或者多个锁对象的时候,可能就会出现死锁现象.可以使用递归锁解决该问题.

rlock =threading.RLock()

rlock.acquire() #可以多次, acquire

rlock.release() #几次acquire,几次acquire

 

1. 信号量

这个类似于停车场,车位是一定数量的,当停满后,外面的车只能在外面等待,线程无法拿到锁,不能执行

lock=threading.BoundedSemaphore(4) #创建一个信号量对象,允许同时有4个线程可以拿到锁,其他的线程需要等有线程去锁后才可以拿到

 

2进程.

注意:进程之间是无法直接共享数据的,所以在父进程创建的锁对象,是无法在子进程中使用的.so,要么在子进程中创建锁,要么直接给他传入一个锁对象(这样就能共享数据了)

代码如下:

import multiprocessing, time, os

 

class Foo(multiprocessing.Process):

def __init__(self, name, lock):

super().__init__()

self.name = name

self.lock = lock

 

def run(self):

self.lock.acquire()

time.sleep(1)

print("foo", self.name, os.getpid())

 

print(self.pid)0

print(self.is_alive())

self.lock.release()

# print(self.g)

 

 

if __name__ == '__main__':

lock = multiprocessing.Lock()

print("开始", os.getpid())

f = Foo("test", lock)

# f.daemon=True

f.start()

f.join()

print("end")

 

 

或者使用内部创建一个锁,下面为Lock 还有Rlock

 

import multiprocessing, time, os

 

 

class Foo(multiprocessing.Process):

def __init__(self, name):

super().__init__()

self.name = name

#在内部创建一个锁对象,并且绑定到self

self.lock = multiprocessing.Lock()

 

def run(self):

self.lock.acquire()

time.sleep(1)

print("foo", self.name, os.getpid())

 

print(self.pid)

print(self.is_alive())

self.lock.release()

# print(self.g)

 

 

if __name__ == '__main__':

lock = multiprocessing.Lock()

print("开始", os.getpid())

f = Foo("test",)

# f.daemon=True

f.start()

f.join()

print("end")

信号量的用法

import multiprocessing, time, os

 

#使用外部传入一个lock

class Foo(multiprocessing.Process):

def __init__(self, name, lock):

super().__init__()

self.name = name

self.lock = lock

 

def run(self):

self.lock.acquire()

time.sleep(1)

print("foo", self.name, time.ctime())

self.lock.release()

 

 

if __name__ == '__main__':

lock = multiprocessing.BoundedSemaphore(3)

print("开始", os.getpid())

p_list = []

for i in range(4):

f = Foo("test", lock)

p_list.append(f)

for p in p_list:

p.start()

for p in p_list:

p.join()

print("end")

 

# 开始 12408

# foo test Fri Nov 10 22:36:17 2017

# foo test Fri Nov 10 22:36:17 2017

# foo test Fri Nov 10 22:36:17 2017

# foo test Fri Nov 10 22:36:18 2017

# end

 

 

使用内部创建一个lock,注意:这个lock对象不能放在__init__方法中,

class Foo(multiprocessing.Process):

lock = multiprocessing.BoundedSemaphore(3)

 

def __init__(self, name, lock):

super().__init__()

self.name = name

 

def run(self):

self.lock.acquire()

time.sleep(1)

print("foo", self.name, time.ctime())

self.lock.release()

 

 

if __name__ == '__main__':

# lock = multiprocessing.BoundedSemaphore(3)

print("开始", os.getpid())

p_list = []

for i in range(4):

f = Foo("test", "")

p_list.append(f)

for p in p_list:

p.start()

for p in p_list:

p.join()

print("end")

 

三 线程/进程之间的通信,和数据共享

1. 线程.

   数据共享:线程之间的数据是共享的,所以一般情况下要加锁,

   通信:使用到两种方法

   i条件（Condition）

   默认条件下，Condition中使用的是Rlock

   并且提供了notify(),notify_all(),wait(),wait_for()等方法

wait()：条件不满足时调用，线程会释放锁并进入等待阻塞；

notify()：条件创造后调用，通知等待池激活一个线程；

notifyAll()：条件创造后调用，通知等待池激活所有线程。

看下面的例子

import threading, time def eat(): if lock.acquire(): print("我拿到锁了") lock.wait() print("吃鱼") lock.release() def fish(): if lock.acquire(): print("准备") time.sleep(1) print("做鱼") print("做好了") lock.notify() lock.release() if __name__ == '__main__': li = [] lock = threading.Condition() print("start") e = threading.Thread(target=eat) f = threading.Thread(target=fish) li.append(e) li.append(f)   for t in li: t.start() for t in li: t.join() print("end")

# start # 我拿到锁了 # 准备 # 做鱼 # 做好了 # 123 # 吃鱼 # end

# 在没有锁的时候 # start # 吃鱼 # 准备 # 做鱼 # 做好了 # end

Ii 事件(evernt)

与Condition类似只是少了锁功能，因为条件同步设计于不访问共享资源的条件环境。event=threading.Event()：条件环境对象，初始值 为False

event.isSet()：返回event的状态值；

event.wait()：如果 event.isSet()==False将阻塞线程；

event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；

event.clear()：恢复event的状态值为False。

 

如下:

# coding=utf-8

import threading, time

def eat():

print("eat",event.is_set())

event.wait()

print("吃鱼")

event.clear()

def fish():

print("is_set?",event.is_set())

print("准备")

time.sleep(1)

print("做鱼")

print("做好了")

event.set()

if __name__ == '__main__':

li = []

event = threading.Event()

print("start")

e = threading.Thread(target=eat)

f = threading.Thread(target=fish)

li.append(e)

li.append(f)

for t in li:

t.start()

for t in li:

t.join()

print("end")

 

# start

# eat False

# is_set? False

# 准备

# 做鱼

# 做好了

# 吃鱼

# end

Iii 队列:

阻塞住了,等待一个get

阻塞住了,等待有一个put

So 应用在多线程中呢

一个线程put,一个get

Eg:

# coding=utf-8

import threading, time

import queue

def eat():

print("吃鱼")

def fish():

print("准备")

time.sleep(1)

print("做鱼")

print("做好了")

if __name__ == '__main__':

q=queue.Queue()

print("start")

e = threading.Thread(target=eat)

f = threading.Thread(target=fish)

q.put(f)

q.put(e)

for i in range(q.qsize()):

t = q.get()

t.start()

t.join()

print("end")

线程池…..下个博客再写,单独拎出来

 

 

2进程

使用Queue

# coding=utf-8

import multiprocessing

 

 

def func(q, i):

q.put(i)

 

 

if __name__ == '__main__':

# 实例化一个queue

q = multiprocessing.Queue()

# 创建进程

li = []

for i in range(3):

mp = multiprocessing.Process(target=func, args=(q, i))

mp.start()

li.append(mp)

for i in li:

i.join()

for i in range(q.qsize()):

print(q.get())

if q.empty():

print("q为空")

 

 

使用Pipes

# coding=utf-8

import multiprocessing

 

 

def f(conn):

print(conn.recv())

 

 

if __name__ == '__main__':

par_conn, sub_conn = multiprocessing.Pipe()

mp = multiprocessing.Process(target=f, args=(sub_conn,))

mp.start()

par_conn.send("hello 酷狗")

mp.join()

print("end")

 

 

 

共享数据: 使用manager

# coding=utf-8

import multiprocessing

 

 

def f(li, i):

li.append(i)

 

 

if __name__ == '__main__':

# li=list((1,2,3)) #使用这一行,不使用下面两行时[1, 2, 3]

manager = multiprocessing.Manager()

li = manager.list((1, 2, 3))

l = []

for i in range(3):

mp = multiprocessing.Process(target=f, args=(li, i))

l.append(mp)

mp.start()

for i in l:

i.join()

print("end")

print(li)

# end

# [1, 2, 3, 0, 2, 1]

进程池………………………改天写