10-线程，进程，协程，IO多路复用

标签：soc   最小   puts   自动   home   manager   阻塞   join   poll   

- 线程进程介绍
    1. 工作最小单元是线程
    2. 应用程序 -> 至少有一个进程  -> 至少有一个线程
    3. 应用场景：
         IO密集型：线程
        计算密集型：进程
    4. GIL，全局解释器锁。
        - 保证同一个进程中只有一个线程同时被调度
- 线程
  1. 基本使用

def task(arg):
    time.sleep(arg)
    print(arg)
for i in range(5):
    t = threading.Thread(target=task,args=[i,])
    # t.setDaemon(True) # 主线程终止，不等待子线程
    # t.setDaemon(False)
    t.start()
    # t.join()  # 一直等
    # t.join(1) # 等待最大时间

  2. 锁

# 1. 只能有一个人使用锁
    # lock = threading.Lock() # 只能开一把
    # lock = threading.RLock()# 可以开多把
# 2. 多个人同时使用锁
    # lock = threading.BoundedSemaphore(3)
# 3. 所有的解脱锁的限制
    # lock = threading.Event()
# 4. 肆意妄为
    # lock = threading.Condition()

3. 线程池
    模式一：直接处理

        3. 线程池
            模式一：直接处理
def task(url):
    """
    任务执行两个操作：下载；保存本地
    """
    # response中封装了Http请求响应的所有数据
    # - response.url            请求的URL
    # - response.status_code    响应状态码
    # - response.text           响应内容（字符串格式）
    # - response.content        响应内容（字节格式）
    # 下载
    response = requests.get(url)

    # 下载内容保存至本地
    f = open(‘a.log‘,‘wb‘)
    f.write(response.content)
    f.close()
pool = ThreadPoolExecutor(2)
url_list = [
    ‘http://www.oldboyedu.com‘,
    ‘http://www.autohome.com.cn‘,
    ‘http://www.baidu.com‘,
]
for url in url_list:
    print(‘开始请求‘,url)
    # 去连接池中获取链接
    pool.submit(task,url)

模式二：分步处理

            模式二：分步处理
def save(future):
    """
    只做保存  # future中包含response
    """
    response = future.result()

    # 下载内容保存至本地
    f = open(‘a.log‘,‘wb‘)
    f.write(response.content)
    f.close()
def task(url):
    """
    只做下载 requests
    """
    # response中封装了Http请求响应的所有数据
    # - response.url            请求的URL
    # - response.status_code    响应状态码
    # - response.text           响应内容（字符串格式）
    # - response.content        响应内容（字节格式）
    # 下载
    response = requests.get(url)
    return response
pool = ThreadPoolExecutor(2)
url_list = [
    ‘http://www.oldboyedu.com‘,
    ‘http://www.autohome.com.cn‘,
    ‘http://www.baidu.com‘,
]
for url in url_list:
    print(‘开始请求‘,url)
    # 去连接池中获取链接
    # future中包含response
    future = pool.submit(task,url)
    # 下载成功后，自动调用save方法
    future.add_done_callback(save)

- 进程
    1. 基本使用

from multiprocessing import Process
import time
def task(arg):
    time.sleep(arg)
    print(arg)
if __name__ == ‘__main__‘:
    for i in range(10):
        p = Process(target=task,args=(i,))
        p.daemon = True
        # p.daemon = False
        p.start()
        p.join(1)

    print(‘主进程最后...‘)

进程锁和线程锁是一样的（使用方式一样）

2. 进程之间的数据共享
     特殊的东西
     - Array(‘类型’,长度)
     - Manager().list() / Manager().dict()
        - 报错是因为，主进程走完了，子进程就就会报错
            - 写一个input或者什么的，让主进程等，就能防止报错
            - 或者jion成为串行。
            - 使用进程池，就可以避免
     - 第三方工具

from multiprocessing import Process
from threading import Thread

"""
# 验证进程之间数据不共享
def task(num,li):
    li.append(num)
    print(li)   # 数据不是共享的，打印出来就是一个个数字 [1] [2] 。。。

if __name__ == ‘__main__‘:
    v = []
    for i in range(10):
        # p = Process(target=task,args=(i,v,))
        p = Thread(target=task,args=(i,v,))
        p.start()
"""

"""
# 方式一：进程数据共享
from multiprocessing import Process,Array
from threading import Thread

def task(num,li):
    li[num] = 1
    print(list(li))

if __name__ == ‘__main__‘:
    v = Array(‘i‘,10) # v[0]
    for i in range(10):
        p = Process(target=task,args=(i,v,))
        p.start()
"""
from multiprocessing import Process,Manager
from threading import Thread

def task(num,li):
    li.append(num)
    print(li)

if __name__ == ‘__main__‘:
    v = Manager().list()
    # v = Manager().dict()
    for i in range(10):
        p = Process(target=task,args=(i,v,))
        p.start()
        # p.join()
    input(‘>>>‘)

3. 进程池

 - from concurrent.futures import ProcessPoolExecutor
   def call(arg):
         data = arg.result()
       print(data)
   def task(arg):
       print(arg)
       return arg + 100
   if __name__ == ‘__main__‘:
       pool = ProcessPoolExecutor(5)
       for i in range(10):
           obj = pool.submit(task,i)
           obj.add_done_callback(call)

================== 结论 ==================
IO密集：线程
计算密集：进程

- 协程 
        pip3 install greenlet
        协程永远是一个线程在执行，对线程的一个分片处理。
    
from greenlet import greenlet

def test1():
    print(12)
    gr2.switch()
    print(34)
    gr2.switch()

def test2():
    print(56)
    gr1.switch()
    print(78)

gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()

    二次加工：
        自定义:
            select实现
        现成 :
            pip3 install gevent

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

def f(url):
    response = requests.get(url)
    print(response.url,response.status_code)

gevent.joinall([
        gevent.spawn(f, ‘http://www.oldboyedu.com/‘),
        gevent.spawn(f, ‘http://www.baidu.com/‘),
        gevent.spawn(f, ‘http://github.com/‘),
])

- IO多路复用
    监听多个socket对象是否有变化（可读，可写，发送错误）
    - 示例一：

import socket
import select

# IO多路复用：8002,8001
#
############################### 基于select实现服务端的“伪”并发 ###############################
sk1 = socket.socket()
sk1.bind((‘127.0.0.1‘,8001,))
sk1.listen(5)

sk2 = socket.socket()
sk2.bind((‘127.0.0.1‘,8002,))
sk2.listen(5)
inputs = [sk1,sk2,]
w_inputs = []
while True:
    # IO多路复用,同时监听多个socket对象
    #    - select,内部进行循环操作(1024)  主动查看
    #    - poll, 内部进行循环操作         主动查看
    #    - epoll,                        被动告知
    r,w,e = select.select(inputs,w_inputs,inputs,0.05)  # r: readble w: writable e:异常
    # r = [sk2,]
    # r = [sk1,]
    # r = [sk1,sk2]
    # r = []
    # r = [conn,]
    # r = [sk1,Wconn]
    #######？
    for obj in r:
        if obj in [sk1,sk2]:
            # 新连接捡来了...
            print(‘新连接来了:‘,obj)
            conn,addr = obj.accept()
            inputs.append(conn)
        else:
            # 有连接用户发送消息来了..
            print(‘有用户发送数据了:‘,obj)
            try:
                data = obj.recv(1024)
            except Exception as ex:
                data = ""
            if data:
                w_inputs.append(obj)
                # obj.sendall(data)
            else:
                obj.close()
                inputs.remove(obj)
                w_inputs.remove(obj)

    for obj in w:
        obj.sendall(b‘ok‘)
        w_inputs.remove(obj)

#  Socket对象
sk1 = socket.socket()
sk1.bind((‘127.0.0.1‘,8001,))
sk1.listen(5)
while True:
    # conn Socket对象,
    conn,addr = sk.accept()
    conn.recv()
    conn.sendall()

- socketserverIO
     - IO多路复用
     - 线程

import socket
import select
import threading

# IO多路复用：8002,8001
#
############################### 基于select实现服务端的“伪”并发 ###############################
"""
def process_request(conn):
    while True:
        v = conn.recv(1024)
        conn.sendall(b‘1111‘)

sk1 = socket.socket()
sk1.bind((‘127.0.0.1‘,8001,))
sk1.listen(5)
inputs=[sk1,]
while True:
    # IO多路复用,同时监听多个socket对象
    #    - select,内部进行循环操作(1024)  主动查看
    #    - poll, 内部进行循环操作         主动查看
    #    - epoll,                        被动告知
    r,w,e = select.select(inputs,[],inputs,0.05)

    for obj in r:
        if obj in sk1:
            # conn客户端的socket
            conn,addr = obj.accept()
            t = threading.Thread(target=process_request,args=(conn,))
            t.start()
"""
# import socketserver
#
# class MyHandler(socketserver.BaseRequestHandler):
#     def handle(self):
#         pass
#
#
# server = socketserver.ThreadingTCPServer((‘127.0.0.1‘,8001),MyHandler)
# server.serve_forever()

- 自定义异步非阻塞的框架

10-线程，进程，协程，IO多路复用

标签：soc   最小   puts   自动   home   manager   阻塞   join   poll   

原文地址：http://www.cnblogs.com/can-H/p/6610683.html