W10_Pipe_Manager数据共享_进程池和回调函数_多线程

W10_Pipe_Manager数据共享_进程池和回调函数

• 管道
• 进程间的数据共享
• 进程池
  • 1.为什么会有进程池的概念
  • 2.进程池原理：
  • 进程池示例
  • 进程池版的socket连接效果
  • 获取进程池异步调用的返回值
  • 进程池回调函数
• 总结：
• 多线程
  • 函数式线程
  • 线程类
  • 多线程socket
  • 线程模块中的方法
  • 守护线程
  • 线程join方法
  • 互斥锁
  • 递归锁RLock--解决死锁问题
  • 线程信号量
  • 线程事件Event
  • 线程条件Condition
  • 队列，栈，优先级队列
  • 线程池
• 协程
• 协程应用:爬虫，socket
• IO多路复用

管道

from multiprocessing import Pipe,Process

def func(conn2):
    print(conn2.recv())

conn1,conn2 = Pipe()
conn1.send("Hello pipe")
p = Process(target=func, args=(conn2,))
p.start()

多进程中管道异常EOFError

from multiprocessing import Pipe, Process
import time
import random


def func_recv(conn1, conn2):
    conn2.close()
    while True:
        try:
            print(conn1.recv())
            time.sleep(random.random())
        except EOFError:
            conn1.close()
            print("recv done")
            break


def func_send(conn1, conn2):
    conn1.close()
    for i in range(4):
        conn2.send("msg %d" % i)
    conn2.close()


conn1, conn2 = Pipe()
recv_p = Process(target=func_recv, args=(conn1, conn2))
send_p = Process(target=func_send, args=(conn1, conn2))
recv_p.start()
send_p.start()
conn1.close()
conn2.close()

注意：多进程使用管道可能会出现数据不安全，需要加锁操作

返回顶部

进程间的数据共享

from multiprocessing import Manager

数据不安全，内容略

进程池

1.为什么会有进程池的概念

效率；
每开启进程，开启属于这个进程的内存空间（如寄存器，堆栈，文件都会战用内存空间）；
进程过多，操作系统调试较为耗时

2.进程池原理：

python中先创建一个属于进程的池子，这个池子指定能存放多少进程，任务存放于队列，等待进程池中的进程处理，当一个进程处理完一个任务后，并不销毁，而是放回进程池，然后继续去任务队列中拿取下一个任务，这就节省了进程被销毁和再创建的时间，也节省了过多进程调度的时间；
信号量与之相比，只是节省了调度时间，因为信号里是控制进程执行的数量，并不能控制进程创建的数量。

进程池示例

from multiprocessing import Pool
import time
import os

def func(n):
    print("start func %s,pid:%s" %(n,os.getpid()))
    time.sleep(1)
    print("end func %s,pid:%s" %(n,os.getpid()))

p = Pool(5)  #初始化进程池，设置可同时开5个进程
for i in range(10):
    p.apply_async(func,args = (i,)) #异步调用
p.close() #进程池结束接收任务
p.join()   #感知进程池中的任务执行结束

返回顶部

进程池版的socket连接效果

server.py

from socket import socket
from multiprocessing import Pool


def interactive(conn):
    conn.send("hello".encode('utf-8'))
    data = conn.recv(1024)
    print(data.decode('utf-8'))



sk = socket()
sk.bind(('127.0.0.1', 9090))
sk.listen()
p = Pool(5)  #进程数一般设置为CPU核数+1
while True:
    conn, add = sk.accept()
    p.apply_async(interactive, args=(conn,))


conn.close()
sk.close()

client.py

from socket import socket

sk = socket()
sk.connect(('127.0.0.1',9090))
data = sk.recv(1024)
print(data.decode('utf-8'))
msg = input("client:")
sk.send(msg.encode("utf-8"))
sk.close()

返回顶部

获取进程池异步调用的返回值

同步调用的返回值

from multiprocessing import Pool


def func(i):
    return i*i

p = Pool(5)
ret_list = []
for i in range(10):
    ret = p.apply(func, args=(i,))
    print(ret)

异步调用的返回值

from multiprocessing import Pool


def func(i):
    return i*i

p = Pool(5)
ret_list = []
for i in range(10):
    ret = p.apply_async(func,args=(i,))
    print(ret)
    ret_list.append(ret)   #存储返回的对象

for ret in ret_list:
    print(ret.get())   #使用get方法获取返回对象的值,
                      #get方法会阻塞，因此不需要进程池close和join

返回顶部

进程池回调函数

callback回调函数，回调函数在主进程中执行

from multiprocessing import Pool
import os
import time

def func(i):
    print("in func,pid: %s" % (os.getpid()))
    time.sleep(1)
    return i * i

def func2(ret):
    print("pid:%s, in func2: %s" %(os.getpid(),ret))

p = Pool(3)
for i in range(10):
    p.apply_async(func, args=(i,), callback=func2)

p.close()
p.join()

总结：

返回顶部

多线程

函数式线程

进程是内存分配的最小单位
线程是操作系统调度的最小单位
线程是CPU执行的最小单位
进程内至少包含一个线程
进程中可以并行执行多个线程
开启一个线程的时间要远远小于开启一个进程；
多个线程内部有自己的数据栈，数据不共享；
全局变量在多个线程之间是共享的。

from threading import Thread
import time

def func(i):
    print(i)
    time.sleep(1)


for i in range(10):
    t = Thread(target=func, args=(i,))
    t.start()

返回顶部

线程类

class MyTread(Thread):
    def __init__(self,arg):
        super().__init__()
        self.arg = arg

    def run(self):
        time.sleep(1)
        print(self.arg)

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

返回顶部

多线程socket

server.py

from socket import socket
from multiprocessing import Pool


def interactive(conn):
    conn.send("hello".encode('utf-8'))
    data = conn.recv(1024)
    print(data.decode('utf-8'))



sk = socket()
sk.bind(('127.0.0.1', 9090))
sk.listen()

while True:
    conn, add = sk.accept()
    t = Thread(target=interactive, args=(conn,))
    t.start()


conn.close()
sk.close()

client.py

from socket import socket

sk = socket()
sk.connect(('127.0.0.1',9090))
data = sk.recv(1024)
print(data.decode('utf-8'))
msg = input("client:")
sk.send(msg.encode("utf-8"))
sk.close()

返回顶部

线程模块中的方法

1.threading.current_thread() #线程对象：线程名和线程号
2.threading.get_ident() #线程号
3.threading.active_count() #在运行的线程数
4.threading.enumerate() #线程对像的列表

print(threading.current_thread())
<_MainThread(MainThread, started 140380643608384)>

print(threading.get_ident())
print(threading.active_count())
print(threading.enumerate())

返回顶部

守护线程

无论是进程还是线程，都遵循：守护xx会等待主xx运行完毕后被销毁。需要强调的是：运行完毕并非终止运行

#1.对主进程来说，运行完毕指的是主进程代码运行完毕
#2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕

#1 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束，
#2 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

示例：

from threading import Thread
import time

def func1():
    print("in func1")
    while True:
        print("*" * 10)
        time.sleep(0.5)

def func2():
    print("in func2")
    time.sleep(3)
    print("func2 done")

t1 = Thread(target=func1,)
t1.daemon = True           #设置为守护线程，必须在start方法之前设置，守护线程等待其它的子线程结束之后才结束
t1.start()
t2 = Thread(target=func2,)
t2.start()

# 主线程会等待其它非守护子线程执行结束后才能结束，
# 因为主线程结束就意味着进程就结束了，进程整体的资源会被回收，而进程必须保证非守护线程运行完毕后才能结束

返回顶部

线程join方法

from threading import Thread
import time

def func1():
    print("in func1")
    while True:
        print("*" * 10)
        time.sleep(0.5)

def func2():
    print("in func2")
    time.sleep(3)
    print("func2 done")

t1 = Thread(target=func1,)
t1.daemon = True           #设置为守护线程，守护线程等待其它的线程结束之后才结束
t1.start()
t2 = Thread(target=func2,)
t2.start()
t2.join()  #代码执行到这里会阻塞，直到t2线程进程完成，再往下执行
print("t2执行完成")
time.sleep(2)
print("主线程执行完成")

返回顶部

互斥锁

未加锁的情况，数据不安全

from threading import Lock,Thread
import time


def funn():
    global n
    tmp = n
    time.sleep(0.1)
    n = tmp - 1


n = 10
t_list = []
for i in range(10):
    t = Thread(target=funn, )
    t.start()
    t_list.append(t)
for t in t_list:
    t.join()
print(n)

加锁后结果符合预期结果

from threading import Lock, Thread
import time


def funn(lock):
    global n
    lock.acquire()
    tmp = n
    time.sleep(0.2)
    n = tmp - 1
    lock.release()


n = 10
t_list = []
lock = Lock()
for i in range(10):
    t = Thread(target=funn, args=(lock,))
    t.start()
    t_list.append(t)
for t in t_list: t.join()
print(n)

返回顶部

递归锁RLock--解决死锁问题

死锁：
在同一个线程或同一个进程里，使用两把锁及两把锁以上的锁时，都有可能产生死锁现象，为了避免，可以改为递归锁

死锁示例

from threading import Lock, Thread
import time

fork_lock = Lock()
noodle_lock = Lock()


def eat1(name):
    noodle_lock.acquire()
    print("%s got the noodle" % name)
    fork_lock.acquire()
    print("%s got the fork" % name)
    print("%s eat noodle" % name)
    fork_lock.release()
    noodle_lock.release()


def eat2(name):
    fork_lock.acquire()
    print("%s got the fork" % name)
    time.sleep(0.1)
    noodle_lock.acquire()
    print("%s got the noodle" % name)
    print("%s eat noodle" % name)
    noodle_lock.release()
    fork_lock.release()


Thread(target=eat1, args=("alex",)).start()

Thread(target=eat2, args=("guest",)).start()

Thread(target=eat1, args=("boots",)).start()

Thread(target=eat2, args=("xiaoming",)).start()

解决死锁，采用递归锁RLock

递归锁在同一个线程里，可以多次acquire,
只要有一个线程有一次acquire,其它线程就无法acquire,也就无法访问acquire中锁住的数据了

from threading import RLock, Thread
import time

fork_lock = noodle_lock = RLock()



def eat1(name):
    noodle_lock.acquire()
    print("%s got the noodle" % name)
    fork_lock.acquire()
    print("%s got the fork" % name)
    print("%s eat noodle" % name)
    fork_lock.release()
    noodle_lock.release()


def eat2(name):
    fork_lock.acquire()
    print("%s got the fork" % name)
    time.sleep(0.1)
    noodle_lock.acquire()
    print("%s got the noodle" % name)
    print("%s eat noodle" % name)
    noodle_lock.release()
    fork_lock.release()


Thread(target=eat1, args=("alex",)).start()

Thread(target=eat2, args=("guest",)).start()

Thread(target=eat1, args=("boots",)).start()

Thread(target=eat2, args=("xiaoming",)).start()

线程信号量

与进程用户相同，
示例：

from threading import Thread,Semaphore
import time


def func(sem,i):
    sem.acquire()
    print(i)
    time.sleep(1)
    sem.release()

sem = Semaphore(3)
for i in range(10):
    t = Thread(target=func,args=(sem,i))
    t.start()

返回顶部

线程事件Event

示例：起两个线程
1.第一个线程：尝试连接数据库：
1）等待一个信号，告诉我们之间的网络是通的
2）连接数据库
2.第二个线程：检测与数据库之间的网络是否连通
1）time.sleep(0,2)
2)将事件的状态设置为True

from threading import Thread,Event
import time
import  random


def connect_db(e):
    count = 0
    while count <3:
        e.wait(0.5)  #状态为False时，只等待0.5s
        if e.is_set():
            print("connected db server")
            break
        else:
            count += 1
            print(" round %s:try to connect db server" % count)

    else:
        raise TimeoutError


def check_network(e):
    time.sleep(random.randint(0,2))
    print("network fine")
    e.set()

e = Event()

Thread(target=check_network,args=(e,)).start()
Thread(target=connect_db, args=(e,)).start()

返回顶部

线程条件Condition

Condition是一个更复杂点的锁
除了提供acquire,release方法之外，还提供wait,notify

wait:一个条件被创建之初，默认有一个False状态，该状态会使wait一直处于等待状态
notify: int数据类型，表示创建的一串钥匙有几把钥匙，钥匙是一次性的

from threading import Thread, Condition

def func(c, i):
    c.acquire()
    c.wait()  # 等钥匙
    print("round:%s" % i)
    c.release()

c = Condition()
for i in range(10):
    t = Thread(target=func, args=(c, i))
    t.start()

while t.is_alive():
    num = input("input key number:").strip()
    c.acquire()
    c.notify(int(num))  # 造钥匙
    c.release()

返回顶部

队列，栈，优先级队列

q = queue.Queue() #队列：先进选出
q = queue.LifoQueue() #栈：先进后出
q = queue.PriorityQueue() #优先级队列：根据优先极，优先极值越小，优先极越高；相同优先极，则比较两个字符串的assic码值，值越小，越优先

q.put()
q.get()
q.put_nowait()

具体方法查询进程一节内容

import queue

# q = queue.Queue()
q = queue.LifoQueue()

q.put('c')
q.put('a')
q.put('b')
print(q.get())
print(q.get())
print(q.get())

q_pr = queue.PriorityQueue()
q_pr.put((10,"a"))
q_pr.put((30,"b"))
q_pr.put((20,"d"))
q_pr.put((20,"c"))
print(q_pr.get())
print(q_pr.get())
print(q_pr.get())
print(q_pr.get())

返回顶部

线程池

方法参见：
http://www.cnblogs.com/Eva-J/articles/8306047.html#_label16

#1 介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.

#2 基本方法
#submit(fn, *args, **kwargs)
异步提交任务

#map(func, *iterables, timeout=None, chunksize=1)
取代for循环submit的操作

#shutdown(wait=True)
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

#result(timeout=None)
取得结果

#add_done_callback(fn)
回调函数

基本使用示例：

from concurrent.futures import ThreadPoolExecutor
import time


def func(n):
    time.sleep(2)
    print(n)
    return n*n


tpool = ThreadPoolExecutor(max_workers=5)  # 默认不要超过CPU个数*5
t_list = []
for i in range(10):
    t = tpool.submit(func, i)   #往线程池中提交任务
    t_list.append(t)
tpool.shutdown()     # 阻塞,等待任务执行完成再往下执行，相当于以前的close() + join()
print('this is in main thread')
for t in t_list:
    print("result: %s " % t.result())

回调函数

from concurrent.futures import ThreadPoolExecutor
import time


def func(n):
    time.sleep(2)
    print(n)
    return n*n

def call_back(m):
    print("call back result:%s" % m.result())

tpool = ThreadPoolExecutor(max_workers=5)  # 默认不要超过CPU个数*5
for i in range(10):
    t = tpool.submit(func, i).add_done_callback(call_back)   #往线程池中提交任务,后面调用回调函数

返回顶部

协程

协程： 能够在一个线程中实现并发效果的概念；能够规避一些任务中的IO操作；在任务执行中，检测到IO就切换到其它任务
协程与线程相比：协程在一个线程上，提高了CPU利用率，切换的效率更快

gevent模块安装：
pip3 install gevent

一个4核CPU的最大并发：
进程 × 线程 × 协程
5 * 20 * 500 = 5W

进程和线程的任务切换由操作系统完成
协程任务之间的切换，由程序（代码）完成，只有遇到协程模块能识别的IO操作的时候，程序才会进行任务切换，实现并发效果

示例一：

from gevent import monkey;monkey.patch_all()  #这个必须写在其它导入的模块之前，以便gevent能正确识别IO操作
import gevent
import time
import threading


def eat():
    print(threading.current_thread())
    print("eating start")
    time.sleep(1)
    print("eating end")


def play():
    print(threading.current_thread())
    print("playing start")
    time.sleep(1)
    print("playing end")


g1 = gevent.spawn(eat)
g2 = gevent.spawn(play)
g1.join()
g2.join()

返回顶部
示例二：

from gevent import monkey;monkey.patch_all()  #这个必须写在其它导入的模块之前，以便gevent能正确识别IO操作
import gevent
import time

def task():
    time.sleep(1)
    print("hello")

def sync():               # 同步调用
    for i in range(10):
        task()


def async():            #异步调用
    g_list = []
    for i in range(10):
        g = gevent.spawn(task)
        g_list.append(g)       #为了让主线程等待所有子线程执行完再结束，这里把子线程放入列表，后续调用join方法

    gevent.joinall(g_list)   # 等价于： for g in g_list:g.join()


#sync()     
async()   #对比异步调用，更节省时间

返回顶部

协程应用:爬虫，socket

爬虫应用示例:

from gevent import monkey;monkey.patch_all()
import gevent
import requests
import time
from urllib.request import urlopen


def get_urlopen(url):
    response = urlopen(url)
    content = response.read().decode("utf-8")
    print(url,len(content))
    return len(content)

def get_requests(url):
    response = requests.get(url)
    try:
        content = response.content.decode("utf-8")
    except UnicodeDecodeError as e:
        content = response.content.decode("gb2312")
    print(url,len(content))
    return len(content)


url_list = ["baidu","sougou","taobao","126","youku"]


start_time = time.time()
g_list = []
for url in url_list:
    full_url = "http://www." + url +".com"
    #g = gevent.spawn(get_urlopen,full_url)
    g = gevent.spawn(get_requests,full_url)
    g_list.append(g)
gevent.joinall(g_list)
print("done!")
tast_time = time.time() - start_time
print(tast_time)

返回顶部
socket应用

协程版socket,server.py

from gevent import monkey;monkey.patch_all()
import gevent
from socket import socket


def talk(conn):
    data = conn.recv(1024).decode("utf-8")
    print(data)
    msg = input(">>>").encode("utf-8")
    conn.send(msg)
    conn.close()


sk = socket()
sk.bind(("127.0.0.2",8989))
sk.listen()
while True:
    conn,add = sk.accept()

    gevent.spawn(talk,conn)

返回顶部

协程版socket,client.py

from socket import socket

sk = socket()
sk.connect(("127.0.0.2",8989))
msg = input("client:").encode("utf-8")
sk.send(msg)
data = sk.recv(1024)
print(data.decode("utf-8"))

返回顶部

IO多路复用

import select

返回顶部

#同步:提交一个任务之后要等待这个任务执行完毕
#异步:只管提交任务,不等待这个任务执行完毕就可以做其他事情
#阻塞 recv recvfrom accept
#非阻塞
#阻塞 线程 运行状态->阻塞状态->就结
#非阻塞
10多路复用
select机制 Windows 1inux都是操作系统轮询每一个被监听的项,看是否有读操作
po11机制 1inux  它可以监听的对象比select机制监听的多
                随着监听项的增多,导致效率降低
#epoll机制 1imux

返回顶部

epoll:

selector_demo.py

import selectors
from socket import *

def accept(sk,mask):
    conn,addr=sk.accept()
    sel.register(conn,selectors.EVENT_READ,read)

def read(conn,mask):
    try:
        data=conn.recv(1024)
        if not data:
            print('closing',conn)
            sel.unregister(conn)
            conn.close()
            return
        conn.send(data.upper()+b'_SB')
    except Exception:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()

sk=socket()
sk.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
sk.bind(('127.0.0.1',8088))
sk.listen(5)
sk.setblocking(False) #设置socket的接口为非阻塞
sel=selectors.DefaultSelector()   # 选择一个适合我的IO多路复用的机制
sel.register(sk,selectors.EVENT_READ,accept)
#相当于网select的读列表里append了一个sk对象,并且绑定了一个回调函数accept
# 说白了就是 如果有人请求连接sk,就调用accrpt方法

while True:
    events=sel.select() #检测所有的sk,conn，是否有完成wait data阶段
    for sel_obj,mask in events:  # [conn]
        callback=sel_obj.data #callback=read
        callback(sel_obj.fileobj,mask) #read(conn,1)