Python进程与线程

进程与线程

1 进程与线程相关概念

1.1 进程

进程定义：

　　进程就是一个程序在一个数据集上的一次动态执行过程。进程一般由程序、数据集、进程控制块三部分组成，是最小的资源管理单元

　　程序：用来描述进程要完成哪些功能以及如何完成；

　　数据集：则是程序在执行过程中所需要使用的资源；

　　进程控制块：用来记录进程的外部特征，描述进程的执行变化过程，系统可以利用它来控制和管理进程，它是系统感知进程存在的唯一标志。

1.2 线程

　　线程的出现是为了降低上下文切换的消耗，提高系统的并发性，并突破一个进程只能干一样事的缺陷，使到进程内并发成为可能。

　　线程也叫轻量级进程，它是一个基本的CPU执行单元，也是程序执行过程中的最小单元，由线程ID、程序计数器、寄存器集合和堆栈共同组成。线程的引入减小了程序并发执行时的开销，提高了操作系统的并发性能。线程没有自己的系统资源。

1.3 进程与线程的关系

　　进程是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。或者说进程是具有一定独立功能的程序关于某个数据集合上的一次运行活动,进程是系统进行资源分配和调度的一个独立单位。
线程则是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位。

进程和线程的关系：

　　(1)一个线程只能属于一个进程，而一个进程可以有多个线程，但至少有一个线程。
　　(2)资源分配给进程，同一进程的所有线程共享该进程的所有资源。
　　(3)CPU分给线程，即真正在CPU上运行的是线程。

1.4 并行和并发

切换：切换的操作者是操作系统

进程/线程切换的原则：

　　1、时间片

　　2、遇到IO操作切换

　　3、优先级切换

并行处理（Parallel Processing）

　　是计算机系统中能同时执行两个或更多个处理的一种计算方法。并行处理可同时工作于同一程序的不同方面。并行处理的主要目的是节省大型和复杂问题的解决时间。

并发处理(concurrency Processing)：

　　指一个时间段中有几个程序都处于已启动运行到运行完毕之间，且这几个程序都是在同一个处理机(CPU)上运行，但任一个时刻点上只有一个程序在处理机(CPU)上运行

　　并发的关键是你有处理多个任务的能力，不一定要同时。并行的关键是你有同时处理多个任务的能力。所以说，并行是并发的子集

　　在python中给每个进程添加一把锁，使每个进程只有一个线程被执行，即在python中的进程中没有线程的并行

1.5 同步与异步

 　　在计算机领域，同步就是指一个进程在执行某个请求的时候，若该请求需要一段时间才能返回信息，那么这个进程将会一直等待下去，直到收到返回信息才继续执行下去；

　　异步是指进程不需要一直等下去，而是继续执行下面的操作，不管其他进程的状态。当有消息返回时系统会通知进程进行处理，这样可以提高执行的效率。

　　举个例子，打电话时就是同步通信，发短息时就是异步通信。

2.threading模块

2.1用Thread类创建子线程

import threading,time
def foo(n):
    print('>>>>>>>>>>>%s'%n)
    time.sleep(n)
def bar(n):
    print('>>>>>>>>>>>%s' % n)
    time.sleep(n)
s=time.time()
t1=threading.Thread(target=foo,args=(2,))
t1.start()
t2=threading.Thread(target=bar,args=(5,))
t2.start()

print('ending！')

　　

2.2 Thread类继承式创建

#继承Thread式创建

import threading
import time

class MyThread(threading.Thread):

    def __init__(self,num):
        threading.Thread.__init__(self)
        self.num=num

    def run(self):
        print("running on number:%s" %self.num)
        time.sleep(3)

t1=MyThread(56)
t2=MyThread(78)

t1.start()
t2.start()
print("ending")

　　

2.3 Thread类的实例方法

# join()：在子线程完成运行之前，这个子线程的父线程将一直被阻塞。

# setDaemon(True)：
        '''
         将线程声明为守护线程，必须在start() 方法调用之前设置，如果不设置为守护线程程序会被无限挂起。 当我们在程序运行中，执行一个主线程，如果主线程又创建一个子线程，主线程和子线程 就分兵两路，分别运行，那么当主线程完成
         想退出时，会检验子线程是否完成。如果子线程未完成，则主线程会等待子线程完成后再退出。但是有时候我们需要的是只要主线程
         完成了，不管子线程是否完成，都要和主线程一起退出，这时就可以 用setDaemon方法啦''

Thread实例对象的方法
  # isAlive(): 返回线程是否活动的。
  # getName(): 返回线程名。
  # setName(): 设置线程名。

threading模块提供的一些方法：
  # threading.currentThread(): 返回当前的线程变量。
  # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
  # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。'

 举例：

import threading,time

def foo(n):
    print('>>>>>>>>>>>%s'%n)
    time.sleep(n)
    print(threading.active_count())

def bar(n):
    print('>>>>>>>>>>>%s' % n)
    time.sleep(n)
    print(threading.active_count())

s=time.time()
t1=threading.Thread(target=foo,args=(2,))
# t1.setDaemon(True)
t1.start()

t2=threading.Thread(target=bar,args=(5,))
# t2.setDaemon(True)
t2.start()

t1.join()
t2.join()

print('++++++++++',threading.active_count())
print('ending！')
print('cost time:',time.time()-s)

 

进程与线程中的lock（线程中互斥锁、递归锁、信号量、Event对象、队列queue）

1、同步锁 (Lock)

　　当全局资源(counter)被抢占的情况，问题产生的原因就是没有控制多个线程对同一资源的访问，对数据造成破坏，使得线程运行的结果不可预期。这种现象称为“线程不安全”。在开发过程中我们必须要避免这种情况，那怎么避免？这就用到了互斥锁了。

例如：

import threading,time
def sub():
    global num         #对全局变量进行操作

    temp=num
    time.sleep(0.001)    #模拟线程执行中出现I/o延迟等
    num=temp-1           #所有线程对全局变量进行减一

    time.sleep(1)

num=100
l=[]

for i in range(100):
    t=threading.Thread(target=sub,args=())
    t.start()
    l.append(t)

for obj in l:
    obj.join()

print(num)          

#执行结果不可预期：
>>:90
>>:93
>>:92

互斥锁概念

　　Python编程中，引入了对象互斥锁的概念，来保证共享数据操作的完整性。每个对象都对应于一个可称为” 互斥锁” 的标记，这个标记用来保证在任一时刻，只能有一个线程访问该对象。在Python中我们使用threading模块提供的Lock类。

　　我们对上面的程序进行整改，为此我们需要添加一个互斥锁变量lock = threading.Lock()，然后在争夺资源的时候之前我们会先抢占这把锁lock.acquire()，对资源使用完成之后我们在释放这把锁mutex.release()。

代码如下：

import threading,time
def sub():
    global num

    lock.acquire()
    temp=num
    time.sleep(0.01)
    num=temp-1
    lock.release()

    time.sleep(1)

num=100
l=[]
lock=threading.Lock()
for i in range(100):
    t=threading.Thread(target=sub,args=())
    t.start()
    l.append(t)

for obj in l:
    obj.join()

print(num)

2、死锁与递归锁

　　所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程。　

会产生死锁的例子：

class MyThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        self.foo()


    def foo(self):
        LockA.acquire()
        print('I am %s GET LOCKA---------%s'%(self.name,time.ctime()))
        LockB.acquire()
        print('I am %s GET LOCKB---------%s' % (self.name, time.ctime()))

        LockB.release()

        LockA.release()

LockA=threading.Lock()
LockB=threading.Lock()

for i in range(10):
    t=MyThread()
    t.start()

使用递归锁解决：

　　在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：

class MyThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        self.foo()
        self.bar()

    def foo(self):
        RLock.acquire()
        print('I am %s GET LOCKA---------%s'%(self.name,time.ctime()))
        RLock.acquire()
        print('I am %s GET LOCKB---------%s' % (self.name, time.ctime()))

        RLock.release()
        RLock.release()

    def bar(self):

        RLock.acquire()
        print('I am %s GET LOCKB---------%s' % (self.name, time.ctime()))
        time.sleep(1)
        RLock.acquire()
        print('I am %s GET LOCKA---------%s' % (self.name, time.ctime()))

        RLock.release()
        RLock.release()

RLock=threading.RLock()

for i in range(10):
    t=MyThread()
    t.start()

　　

3、Semaphore（信号量）

Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

实例：(同时只有5个线程可以获得semaphore,即可以限制最大连接数为5)：

import threading
import time

semaphore = threading.Semaphore(5)

def func():
    if semaphore.acquire():
        print (threading.currentThread().getName() + ' get semaphore')
        time.sleep(2)
        semaphore.release()

for i in range(20):
  t1 = threading.Thread(target=func)
  t1.start()

4、Event对象

　　线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就 会变得非常棘手。

　　为了解决这些问题,我们需要使用threading库中的Event对象。Event对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。

　　在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行

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

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

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

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

用evnt对象模拟红绿灯：

import queue,threading,time
import random

event = threading.Event()
def light():
    while True:
        event.set()
        for i in range(10):
            print('light green')
            time.sleep(1)
        event.clear()
        for i in range(10,13):
            print('light yellow')
            time.sleep(1)
        for i in range(13,21):
            print('light red')
            time.sleep(1)

def car(i):
    while True:
        time.sleep(random.randint(1,5))
        if event.isSet():
            print('car %s is runing'%i)
        else:
            print('car %s is waiting'%i)

if __name__ == '__main__':
    l1=threading.Thread(target=light)
    l1.start()

    for i in range(5):
        i = threading.Thread(target=car,args=(i,))
        i.start()

4、队列(queue)

'''

创建一个“队列”对象

import queueq
q = queue.Queue(maxsize = 10)
　　　　#queue.Queue类即是一个队列的同步实现。队列长度可为无限或者有限。可通过Queue的构造函数的可选参数
　　　　#maxsize来设定队列长度。如果maxsize小于1就表示队列长度无限。

q.put()    将一个值22放入队列中

　　　　#调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目的值；第二个block为可选参数，默认为1。如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。如果block为0，put方法将引发Full异常。

q.get()    将一个值从队列中取出　　　　

　　　　#调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常。

'''

queue的常用方法

'''

此包中的常用方法(q = Queue.Queue()):

q.qsize() 返回队列的大小
q.empty() 如果队列为空，返回True,反之False
q.full() 如果队列满了，返回True,反之False
q.full 与 maxsize 大小对应
q.get([block[, timeout]]) 获取队列，timeout等待时间
q.get_nowait() 相当q.get(False)非阻塞 
q.put(item) 写入队列，timeout等待时间
q.put_nowait(item) 相当q.put(item, False)
q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号
q.join() 实际上意味着等到队列为空，再执行别的操作

'''

join与task_done方法

'''
join() 阻塞进程，直到所有任务完成，需要配合另一个方法task_done。

    def join(self):
     with self.all_tasks_done:
      while self.unfinished_tasks:
       self.all_tasks_done.wait()

task_done() 表示某个任务完成。每一条get语句后需要一条task_done。


import queue
q = queue.Queue(5)
q.put(10)
q.put(20)
print(q.get())
q.task_done()
print(q.get())
q.task_done()

q.join()

print("ending!")
'''

queue的三种模式：

1、queue.Queue()  先进先出模式

2、queue.LifoQueue()    先进后出，类似栈

3、queue.PriorityQueue()   优先级模式，优先级越高越先出，数字月底代表优先级越高

import queue

#######################先进后出
q=queue.LifoQueue()

q.put(34)
q.put(56)
q.put(12)

#####################优先级
q=queue.PriorityQueue()
q.put([5,100])
q.put([7,200])
q.put([3,"hello"])
q.put([4,{"name":"alex"}])

while 1:
  data=q.get()
  print(data)

队列的应用：生产者消费者模型

　　在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

　　生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

　　这就像，在餐厅，厨师做好菜，不需要直接和客户交流，而是交给前台，而客户去饭菜也不需要不找厨师，直接去前台领取即可，这也是一个结耦的过程。

import queue,threading,time
import random

q = queue.Queue(50)

def Producer():
    while True:
        if q.qsize() < 20:
            n = random.randint(1, 100)
            q.put(n)
            print(" has made baozi %s" % n)
            time.sleep(1)

def Consumer(id):
    while True:
         s = q.get()
         print("Consumer"+id+"has eat %s" % s)
         time.sleep(2)

for i in range(5):
    t1=threading.Thread(target=Producer,args=())
    t1.start()

for i in range(2):
    t=threading.Thread(target=Consumer,args=(str(i),))
    t.start()