什么是进程(process)
An executing instance of a program is called a process.
Each process provides the resources needed to execute a program. A process has a virtual address space, executable code, open handles to system objects, a security context, a unique process identifier, environment variables, a priority class, minimum and maximum working set sizes, and at least one thread of execution. Each process is started with a single thread, often called the primary thread, but can create additional threads from any of its threads.
翻译： 一个应用程序中一个可执行的实例被叫做进程。每个进程提供了所需的资源来执行一个程序。一个进程通常包含一块虚拟的地址空间、可执行代码、打开系统对象句柄、一个安全的隔离环境、一个独特的进程标识符、环境变量、优先级、最小和最大工作集大小、和至少一个线程的执行。每个流程开始一个线程,通常被称为主要的线程,但从它的任何线程可以创建额外的线程。
程序并不能单独运行，只有将程序装载到内存中，系统为它分配资源才能运行，而这种执行的程序就称之为进程。程序和进程的区别就在于：程序是指令的集合，它是进程运行的静态描述文本；进程是程序的一次执行活动，属于动态概念。
在多道编程中，我们允许多个程序同时加载到内存中，在操作系统的调度下，可以实现并发地执行。这是这样的设计，大大提高了CPU的利用率。进程的出现让每个用户感觉到自己独享CPU，因此，进程就是为了在CPU上实现多道编程而提出的。
进程的缺陷：
- 进程只能在一个时间干一件事，如果想同时干两件事或多件事，进程就无能为力了。
- 进程在执行的过程中如果阻塞，例如等待输入，整个进程就会挂起，即使进程中有些工作不依赖于输入的数据，也将无法执行。
举例说明：一个操作系统就像是一个工厂，工厂里面有很多个生产车间，不同的车间生产不同的产品，每个车间就相当于一个进程，且你的工厂又穷，供电不足，同一时间只能给一个车间供电，为了能让所有车间都能同时生产，你的工厂的电工只能给不同的车间分时供电，但是轮到你的qq车间时，发现只有一个干活的工人，结果生产效率极低，为了解决这个问题，应该怎么办呢？。。。。没错，你肯定想到了，就是多加几个工人，让几个人工人并行工作，这每个工人，就是线程！
什么是线程(thread)
线程是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务.
A thread is an execution context, which is all the information a CPU needs to execute a stream of instructions.
Suppose you're reading a book, and you want to take a break right now, but you want to be able to come back and resume reading from the exact point where you stopped. One way to achieve that is by jotting down the page number, line number, and word number. So your execution context for reading a book is these 3 numbers.
If you have a roommate, and she's using the same technique, she can take the book while you're not using it, and resume reading from where she stopped. Then you can take it back, and resume it from where you were.
Threads work in the same way. A CPU is giving you the illusion that it's doing multiple computations at the same time. It does that by spending a bit of time on each computation. It can do that because it has an execution context for each computation. Just like you can share a book with your friend, many tasks can share a CPU.
On a more technical level, an execution context (therefore a thread) consists of the values of the CPU's registers.
Last: threads are different from processes. A thread is a context of execution, while a process is a bunch of resources associated with a computation. A process can have one or many threads.
Clarification: the resources associated with a process include memory pages (all the threads in a process have the same view of the memory), file descriptors (e.g., open sockets), and security credentials (e.g., the ID of the user who started the process).
进程与线程的区别
- 所有在同一个进程里的线程是共享同一块内存空间的，
- 两个进程想通信，必须通过一个中间代理来实现
- 创建新线程很简单，创建新进程需要对其父进程进行一次克隆。
- 一个线程可以控制同一进程里的其他线程，但是进程只能操作子进程
- 主线程变更(取消、优先级变化等)可能会影响进程的其他线程的行为;父进程的变化不会影响子进程
Python GIL(Global Interpreter Lock,全局解释器锁)
全局解释器锁在CPython的,或GIL,是一个互斥锁,防止多个本地线程执行Python字节码。这把锁是必要的,主要是因为CPython的内存管理不是线程安全的。(然而,由于GIL存在,其他功能已经习惯于依赖保证执行)。无论你启多少个线程，你有多少个cpu, Python在执行的时候会淡定的在同一时刻只允许一个线程运行。
python的多线程就是调用的操作系统的原生线程。
Python threading模块
调用方法有两种，一是直接调用，如下：
import threading
import time
def sayhi(num): #定义每个线程要运行的函数
print("running on number:%s" %num)
time.sleep(3)
if __name__ == '__main__':
t1 = threading.Thread(target=sayhi,args=(1,)) #生成一个线程实例
t2 = threading.Thread(target=sayhi,args=(2,)) #生成另一个线程实例
t1.start() #启动线程
t2.start() #启动另一个线程
print(t1.getName()) #获取线程名
print(t2.getName())
第二种是继承式调用：
import threading
import time
class MyThread(threading.Thread):
def __init__(self,num):
#threading.Thread.__init__(self)
super(MyThread, self).__init__()
self.num = num
def run(self):#定义每个线程要运行的函数
print("running on number:%s" %self.num)
time.sleep(3)
if __name__ == '__main__':
t1 = MyThread(1)
t2 = MyThread(2)
t1.start()
t2.start()
jion 和 Daemon
#coding:utf-8
import threading
import time
def run(num):
print("task", num)
time.sleep(2)
print("taskdone",num)
start_time = time.time()
for i in range(50):
t = threading.Thread(target=run, args=("t%s" %i,))
t.start()
print("all thread is finished ------")
print("cost:",time.time() - start_time) #默认情况下主线程是不会等待子线程执行完毕的，他们是并行的
-----------------华丽的分割线----------------------------
想要计算时间需将代码更改如下：
import threading
import time
def run(num):
print("task", num)
time.sleep(2)
print("taskdone",num)
start_time = time.time()
tlist = [] #存线程实例
for i in range(50):
t = threading.Thread(target=run, args=("t%s" %i,))
tlist.append(t)#为了不阻塞后面的线程的启动，不在这里join，先放到一个列表里
t.start()
for t in tlist: #循环线程实例列表，等待所有线程执行完毕
t.join()
print("all thread is finished ------",threading.currentThread(),threading.active_count()) #证明此处是主线程，执行结果中可看到MainThreadstarted 140639423129344
#threading.active_count() 查看当前活动的线程的个数
print("cost:",time.time() - start_time)
# .join 逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义
-----------------华丽的分割线----------------------------
import threading
import time
def run(num):
print("task", num)
time.sleep(2)
print("taskdone",num)
start_time = time.time()
tlist = [] #存线程实例
for i in range(50):
t = threading.Thread(target=run, args=("t%s" %i,))
t.setDaemon(True) #把当前线程设置为守护线程，一定要在start之前，此时程序不会等待子线程执行完毕
tlist.append(t)#为了不阻塞后面的线程的启动，不在这里join，先放到一个列表里
t.start()
# for t in tlist: #循环线程实例列表，等待所有线程执行完毕
# t.join()
print("all thread is finished ------",threading.currentThread(),threading.active_count()) #证明此处是主线程，执行结果中可看到MainThreadstarted 140639423129344
#threading.active_count() 查看当前活动的线程的个数
print("cost:",time.time() - start_time)
'''setDaemon设置为后台线程或前台线程（默认，如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止。'''
总结：
- start 线程准备就绪，等待CPU调度
- setName 为线程设置名称
- getName 获取线程名称
- setDaemon 设置为后台线程或前台线程（默认），如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止。
- join 逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义
- run 线程被cpu调度后自动执行线程对象的run方法
线程锁（互斥锁Mutex Lock）
一个进程下可以启动多个线程，多个线程共享父进程的内存空间，也就意味着每个线程可以访问同一份数据，此时，如果2个线程同时要修改同一份数据，会出现什么状况？
由于线程之间是进行随机调度，并且每个线程可能只执行n条记录之后，当多个线程同时修改同一条数据时可能会出现脏数据，所以，出现了线程锁 - 同一时刻允许一个线程执行操作。
未加锁版本代码入下：
#coding:utf-8
import threading
import time
num = 0
def run(n):
global num
num += 1
tlist = []
for i in range(50):
t = threading.Thread(target=run, args=("t-%s" %i, ))
t.start()
tlist.append(t)
for t in tlist:
t.join()
print("----all threads is finished----")
print("num:",num)
执行结果如下：
yang@yang-virtual-machine:~/桌面/MyProject$ python threading_ex2.py
----all threads is finished----
('num:', 49)
此时在ubuntu的2.7版本下测试数据的结果为49，也就是说这中间产生了脏数据。
*注：不要在3.x上运行，不知为什么，3.x上的结果总是正确的，可能是自动加了锁
加锁版版本如下：
#coding:utf-8
import threading
import time
num = 0
def run(n):
lock.acquire() #获取一把锁
global num
num += 1
time.sleep(0.5) #加锁之后整个程序变串行了，这个时候就需要立即释放这个锁
lock.release()#释放锁
lock = threading.Lock() #生成一个锁的实例
tlist = []
for i in range(50):
t = threading.Thread(target=run, args=("t-%s" %i, ))
t.start()
tlist.append(t)
for t in tlist:
t.join()
print("----all threads is finished----")
print("num:",num)
PS:++全局解释器锁是保证同一个时间只有一个线程在执行，这里的用户态lock是保证同一时间只有一个线程在修改数据。++
RLock（递归锁）
连续锁好几次的时候，就一定要用递归锁，示例代码如下：
import threading,time
def run1():
print("grab the first part data")
lock.acquire()
global num
num +=1
lock.release()
return num
def run2():
print("grab the second part data")
lock.acquire()
global num2
num2+=1
lock.release()
return num2
def run3():
lock.acquire()
res = run1()
print('--------between run1 and run2-----')
res2 = run2()
lock.release()
print(res,res2)
if __name__ == '__main__':
num,num2 = 0,0
lock = threading.RLock() #此时如果换成Lock 将会进入死循环，它找不到出来的锁了
for i in range(10):
t = threading.Thread(target=run3)
t.start()
while threading.active_count() != 1:
print(threading.active_count())
else:
print('----all threads done---')
print(num,num2)
信号量（Semaphore）
互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。实际用例例如MySQL的线程池。
import threading,time
def run(n):
semaphore.acquire() #申请 跟锁是一样的 只是一次放进去5个线程，5个同时改数据 也是会出错的
time.sleep(1)
print("run the thread: %s
" %n)
semaphore.release() #释放
if __name__ == '__main__':
num= 0
semaphore = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
for i in range(20):
t = threading.Thread(target=run,args=(i,))
t.start()
while threading.active_count() != 1:
pass #print threading.active_count()
else:
print('----all threads done---')
print(num)
Event(事件)
threading.Event机制类似于一个线程向其它多个线程发号施令的模式，其它线程都会持有一个threading.Event的对象，这些线程都会等待这个事件的“发生”，如果此事件一直不发生，那么这些线程将会阻塞，直至事件的“发生”。
对此，我们可以考虑一种应用场景（仅仅作为说明），例如，我们有多个线程从Redis队列中读取数据来处理，这些线程都要尝试去连接redis的服务，一般情况下，如果Redis连接不成功，在各个线程的代码中，都会去尝试重新连接。如果我们想要在启动时确保Redis服务正常，才让那些工作线程去连接Redis服务器，那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作：主线程中会去尝试连接Redis服务，如果正常的话，触发事件，各工作线程会尝试连接Redis服务。
import threading
import time
import logging
logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-10s) %(message)s',)
def worker(event):
logging.debug('Waiting for redis ready...')
event.wait()
logging.debug('redis ready, and connect to redis server and do some work [%s]', time.ctime())
time.sleep(1)
readis_ready = threading.Event()
t1 = threading.Thread(target=worker, args=(readis_ready,), name='t1')
t1.start()
t2 = threading.Thread(target=worker, args=(readis_ready,), name='t2')
t2.start()
logging.debug('first of all, check redis server, make sure it is OK, and then trigger the redis ready event')
time.sleep(3) # simulate the check progress
readis_ready.set()
事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。
clear：将“Flag”设置为False
set：将“Flag”设置为True
通过Event来实现两个或多个线程间的交互，下面是一个红绿灯的例子，即起动一个线程做交通指挥灯，生成几个线程做车辆，车辆行驶按红灯停，绿灯行的规则。
a client thread can wait for the flag to be set
event.wait()
a server thread can set or reset it
event.set()
event.clear()
If the flag is set, the wait method doesn’t do anything.
标志位被设定，wait就不会在阻塞。代表绿灯。
If the flag is cleared, wait will block until it becomes set again.
标志位被清空，就会被阻塞，wait等待绿灯。
Any number of threads may wait for the same event.
import threading,time
import random
def light():
if not event.isSet():
event.set() #wait就不阻塞 #绿灯状态
count = 0
while True:
if count < 10:
print(' 33[42;1m--green light on--- 33[0m')
elif count <13:
print(' 33[43;1m--yellow light on--- 33[0m')
elif count <20:
if event.isSet():
event.clear()
print(' 33[41;1m--red light on--- 33[0m')
else:
count = 0
event.set() #打开绿灯
time.sleep(1)
count +=1
def car(n):
while 1:
time.sleep(random.randrange(10))
if event.isSet(): #绿灯
print("car [%s] is running.." % n)
else:
# event.wait()
print("car [%s] is waiting for the red light.." %n)
if __name__ == '__main__':
event = threading.Event()
Light = threading.Thread(target=light)
Light.start()
for i in range(3):
t = threading.Thread(target=car,args=(i,))
t.start()
queue 对列
队列主要有两个作用，1是解耦合，2是提高效率，队列中的数据只有一份，取走了就没了。队列的数据在内存中。
对列就是一个容器，用来放数据的，是有顺序的。
class queue.Queue(maxsize=0) #先入先出
class queue.LifoQueue(maxsize=0) #后进先出
class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列
>>> import queue
>>> q = queue.Queue()
>>> q.put(1)
>>> q.get()
1
>>> q.get_nowait()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/lib/python3.5/queue.py", line 192, in get_nowait
return self.get(block=False)
File "/usr/lib/python3.5/queue.py", line 161, in get
raise Empty
queue.Empty
>>> q.put(2)
>>> q.get_nowait()
2
>>> q.get_nowait()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/lib/python3.5/queue.py", line 192, in get_nowait
return self.get(block=False)
File "/usr/lib/python3.5/queue.py", line 161, in get
raise Empty
queue.Empty
>>>
#q.get() 会阻塞，这个时候用q.get_nowait()会抛出异常，此时我们在程序中就可以抓住异常来进行处理。等同于
q.get(baock=Flase, timeout=1)
后进先出
import queue
q = queue.LifoQueue()
q.put("d1")
q.put("d2")
q.put("d3")
print(q.get())
print(q.get())
print(q.get())
生产者消费者模型
在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。
为什么要使用生产者和消费者模式
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
#coding:utf-8
import threading
import time
import queue
q = queue.Queue(maxsize=10)
def Producer(name):
count = 1
while True:
q.put("包子%s" %(count))
print("生产了包子",count)
count += 1
def Consumer(name):
while q.qsize() > 0:
print("[%s] 取到了 [%s] 并且吃了它......"%(name, q.get()))
time.sleep(1)
p = threading.Thread(target=Producer, args=("Abc",))
c = threading.Thread(target=Consumer, args=("Def",))
p.start()
c.start()
PS:I/O操作不占用cpu,计算占用CPU，Python多线程不适合CPU密集操作型的任务，适合I/O操作密集型的任务。
multiprocessing(多进程)
#coding:utf-8
import multiprocessing
import time,threading
def thread_run():
print(threading.get_ident()) #获取线程号
def run(name):
time.sleep(2)
print("hello", name)
t = threading.Thread(target=thread_run,)
t.start()
if __name__ == "__main__":
for i in range(10):
p = multiprocessing.Process(target=run, args=("BOB %s"%i,))
p.start()
#开启了10个进程，每个进程又开启了一个线程。
在Linux上每一个进程都是由父进程执行的。每一个进程默认都会有一个父进程。
from multiprocessing import Process
import os
def info(title):
print(title)
print('module name:', __name__)
print('parent process:', os.getppid())
print('process id:', os.getpid())
print("
")
def f(name):
info(' 33[31;1mfunction f 33[0m')
print('hello', name)
if __name__ == '__main__':
info(' 33[32;1mmain process line 33[0m')
p = Process(target=f, args=('bob',))
p.start()
p.join()
执行结果如下：
C:Python36python.exe C:/Users/ly/Desktop/MyProject/dayeight/多进程2.py
main process line
module name: __main__
parent process: 6380
process id: 2892
function f
module name: __mp_main__
parent process: 2892
process id: 8276
hello bob
进程间通信
不同进程间内存是不共享的，要想实现两个进程间的数据交换，必须寻找一个中间商，可以用以下方法：
- Queues
from multiprocessing import Process, Queue
def f(q):
q.put([42, None, "Hello"])
if __name__ == '__main__':
q = Queue()
p = Process(target=f, args=(q,))
p.start()
print(q.get())
p.join()
#这里的queue跟线程的queue是不同的，线程的q传入之后不能够序列化，这里实现的子进程和主进程的通信。
#父进程里的Q 传给了子进程，这个时候就实现了数据共享。其实是克隆了一个Q。然后交给了子进程。这里不是同时修改了数据，只是实现了数据的传递。
- Pipes
from multiprocessing import Process, Pipe
def f(conn):
conn.send([42, None, 'hello'])
conn.send([42, None, 'hello2'])
conn.close()
if __name__ == '__main__':
parent_conn, child_conn = Pipe() #管道一生成就会生成两个返回对象，分别赋值
p = Process(target=f, args=(child_conn,)) #传给了child_conn，就开始传输数据
p.start()
print(parent_conn.recv())
print(parent_conn.recv())
#print(parent_conn.recv())# prints "[42, None, 'hello']" #在管道的另一头直接收消息
#接收多次的时候就会阻塞，这个时候parent也是可以发送的，这类似于socket，互相交互
p.join()
PS:Queue 和 Pipe只是实现进程间数据的传递,还没有实现数据的共享。这个时候需要用到Managers。
- Manager
可以实现list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array的共享。
from multiprocessing import Process, Manager
import os
def f(d, l):
d[1] = '1'
d['2'] = 2
d[0.25] = None
l.append(os.getpid())
print(l)
if __name__ == '__main__':
with Manager() as manager:
d = manager.dict() #生成一个字典，在多个进程间共享和传递
l = manager.list(range(5)) #生成一个列表
p_list = []
for i in range(10):
p = Process(target=f, args=(d, l))
p.start()
p_list.append(p)
for res in p_list:
res.join()
print(d)
print(l)
- 进程同步
from multiprocessing import Process, Lock
def f(l, i):
l.acquire()
try:
print('hello world', i)
finally:
l.release() #虽然每个进程都是独立的 因为屏幕是共享的，控制这个锁，在屏幕上打印的时候，避免处乱。
if __name__ == '__main__':
lock = Lock()
for num in range(10):
Process(target=f, args=(lock, num)).start()
进程池
同一个时间有多少个进程在cpu上运行。
进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。
进程池中有两个方法：apply，apply_async
from multiprocessing import Process, Pool, freeze_support
import time,os
def Foo(i):
time.sleep(2)
print("in process",os.getpid())
return i + 100
def Bar(arg):
print('-->exec done:', arg,os.getpid())
if __name__ == '__main__':
pool = Pool(5)#允许进程池里同时放入5个进程，交给CPU运行
print("主进程：",os.getpid())
for i in range(10):
pool.apply_async(func=Foo, args=(i,), callback=Bar)
#apply_async是并行，callback是回调函数，执行完foo的时候再执行bar，并将返回值传给bar,主进程执行的回调。
#pool.apply(func=Foo, args=(i,)) #apply 是串行
print('end')
pool.close()
pool.join() # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。