线程 以及 thread类

Posted 2020-12-23 usherwang

1.多进程实现并发的socket

# server.py
?
import socket
from multiprocessing import Process
def chat(conn):
    while True:
        try:
            ret = conn.recv(1024).decode(‘utf-8‘)
            conn.send(ret.upper().encode(‘utf-8‘))
        except ConnectionResetError:
            break
?
if __name__ == ‘__main__‘:
    sk = socket.socket()
    sk.bind((‘127.0.0.1‘, 9000))
    sk.listen()
    while True:
        conn,_ = sk.accept()
        Process(target=chat,args=(conn,)).start()
        
# client.py
?
import socket
?
sk = socket.socket()
sk.connect((‘127.0.0.1‘,9000))
while True:
    sk.send(b‘hello‘)
    msg = sk.recv(1024)
    print(msg)

# 面试题
# 5个进程 p1-p5,p1,p2,结束之后p3-p5才开始,p3必须随着p4p5结束
from multiprocessing import Process
def func():
    pass
if __name__ == ‘__main__‘:
    p1 = Process(target=func)
    p2 = Process(target=func)
    p3 = Process(target=func)
    p4 = Process(target=func)
    p5 = Process(target=func)
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    p3.daemon = True
    p3.start()
    p4.start()
    p5.start()
    p4.join()
    p5.join()

2.生产者消费者模型

# 进程
    # 一个进程就是一个生产者
    # 一个进程就是一个消费者
# 队列
    # 生产者和消费者之间的容器就是队列
import time
import random
from multiprocessing import Process,Queue
?
def producer(q,name,food):
    for i in range(10):
        time.sleep(random.random())
        fd = ‘%s%s‘%(food,i)
        q.put(fd)
        print(‘%s生产了一个%s‘%(name,food))
?
def consumer(q,name):
    while True:
        food = q.get()
        if not food:break
        time.sleep(random.randint(1,3))
        print(‘%s吃了%s‘%(name,food))
?
?
def cp(c_count,p_count):
    q = Queue(10)
    for i in range(c_count):
        Process(target=consumer, args=(q, ‘alex‘)).start()
    p_l = []
    for i in range(p_count):
        p1 = Process(target=producer, args=(q, ‘wusir‘, ‘泔水‘))
        p1.start()
        p_l.append(p1)
    for p in p_l:p.join()
    for i in range(c_count):
        q.put(None)
if __name__ == ‘__main__‘:
    cp(2,3)

3.生产者消费者实现的爬虫示例

from multiprocessing import Process,Queue
import requests
?
import re
import json
def producer(q,url):
    response = requests.get(url)
    q.put(response.text)
?
def consumer(q):
    while True:
        s = q.get()
        if not s:break
        com = re.compile(
            ‘<div class="item">.*?<div class="pic">.*?<em .*?>(?P<id>d+).*?<span class="title">(?P<title>.*?)</span>‘
            ‘.*?<span class="rating_num" .*?>(?P<rating_num>.*?)</span>.*?<span>(?P<comment_num>.*?)评价</span>‘, re.S)
        ret = com.finditer(s)
        for i in ret:
            print({
                "id": i.group("id"),
                "title": i.group("title"),
                "rating_num": i.group("rating_num"),
                "comment_num": i.group("comment_num")}
            )
?
if __name__ == ‘__main__‘:
    count = 0
    q = Queue(3)
    p_l = []
    for i in range(10):
        url = ‘https://movie.douban.com/top250?start=%s&filter=‘%count
        count+=25
        p = Process(target=producer,args=(q,url,)).start()
        p_l.append(p)
    p = Process(target=consumer, args=(q,)).start()
    for p in p_l:p.join()
    q.put(None)

4.joinablequeue

import time
import random
from  multiprocessing import JoinableQueue,Process
?
def producer(q,name,food):
    for i in range(10):
        time.sleep(random.random())
        fd = ‘%s%s‘%(food,i)
        q.put(fd)
        print(‘%s生产了一个%s‘%(name,food))
    q.join()
?
def consumer(q,name):
    while True:
        food = q.get()
        time.sleep(random.random())
        print(‘%s吃了%s‘%(name,food))
        q.task_done()
?
if __name__ == ‘__main__‘:
    jq = JoinableQueue()
    p =Process(target=producer,args=(jq,‘wusir‘,‘泔水‘))
    p.start()
    c = Process(target=consumer,args=(jq,‘alex‘))
    c.daemon = True
    c.start()
    p.join()

5.进程之间的数据共享

from multiprocessing import Manager,Process,Lock
?
def func(dic,lock):
    with lock:
        dic[‘count‘] -= 1
?
if __name__ == ‘__main__‘:
    # m = Manager()
    with Manager() as m:
        l = Lock()
        dic = m.dict({‘count‘:100})
        p_l = []
        for i in range(100):
            p = Process(target=func,args=(dic,l))
            p.start()
            p_l.append(p)
        for p in p_l:p.join()
        print(dic)
?
?
# mulprocessing中有一个manager类
# 封装了所有和进程相关的 数据共享 数据传递
# 相关的数据类型
# 但是对于 字典 列表这一类的数据操作的时候会产生数据不安全
# 需要加锁解决问题，并且需要尽量少的使用这种方式

6.线程的理论

# 线程 开销小 数据共享 是进程的一部分
# 进程 开销大 数据隔离 是一个资源分配单位
?
# cpython解释器 不能实现多线程利用多核
?
?
# 锁 ：GIL 全局解释器锁
    # 保证了整个python程序中，只能有一个线程被CPU执行
    # 原因：cpython解释器中特殊的垃圾回收机制
    # GIL锁导致了线程不能并行，可以并发
# 所以使用所线程并不影响高io型的操作
# 只会对高计算型的程序由效率上的影响
# 遇到高计算 ： 多进程 + 多线程
              # 分布式
?
# cpython pypy jpython iron python
?
# 遇到IO操作的时候
    # 5亿条cpu指令/s
    # 5-6cpu指令 == 一句python代码
    # 几千万条python代码
# web框架 几乎都是多线程

7.thread类

import os
import time
from threading import Thread
#multiprocessing 是完全仿照这threading的类写的
def func():
    print(‘start son thread‘)
    time.sleep(1)
    print(‘end son thread‘,os.getpid())
# 启动线程 start
Thread(target=func).start()
print(‘start‘,os.getpid())
time.sleep(0.5)
print(‘end‘,os.getpid())
?
#开启多个子线程
def func(i):
    print(‘start son thread‘,i)
    time.sleep(1)
    print(‘end son thread‘,i,os.getpid())
?
for i in range(10):
    Thread(target=func,args=(i,)).start()
print(‘main‘)
#主线程什么时候结束？等待所有子线程结束之后才结束
#主线程如果结束了，主进程也就结束了
?
#join方法  阻塞 直到子线程执行结束
def func(i):
    print(‘start son thread‘,i)
    time.sleep(1)
    print(‘end son thread‘,i,os.getpid())
t_l = []
for i in range(10):
    t = Thread(target=func,args=(i,))
    t.start()
    t_l.append(t)
for t in t_l:t.join()
print(‘子线程执行完毕‘)
?
#使用面向对象的方式启动线程
class MyThread(Thread):
    def __init__(self,i):
        self.i = i
        super().__init__()
    def run(self):
        print(‘start‘,self.i,self.ident)
        time.sleep(1)
        print(‘end‘,self.i)
?
for i in range(10):
    t = MyThread(i)
    t.start()
    print(t.ident)
?
#线程里的一些其他方法
from threading import current_thread,enumerate,active_count
def func(i):
    t = current_thread()
    print(‘start son thread‘,i,t.ident)
    time.sleep(1)
    print(‘end son thread‘,i,os.getpid())
?
t = Thread(target=func,args=(1,))
t.start()
print(t.ident)
print(current_thread().ident)   # 水性杨花 在哪一个线程里，current_thread()得到的就是这个当前线程的信息
print(enumerate())
print(active_count())   # =====len(enumerate())
?
#terminate 结束进程
#在线程中不能从主线程结束一个子线程
?
#测试
    #进程和线程的效率差
def func(a,b):
    c = a+b
import time
from multiprocessing import Process
from threading import Thread
if __name__ == ‘__main__‘:
    start = time.time()
    p_l = []
    for  i in range(500):
        p = Process(target=func,args=(i,i*2))
        p.start()
        p_l.append(p)
    for p in p_l:p.join()
    print(‘process :‘,time.time() - start)
?
    start = time.time()
    p_l = []
    for i in range(500):
        p = Thread(target=func, args=(i, i * 2))
        p.start()
        p_l.append(p)
    for p in p_l: p.join()
    print(‘thread :‘,time.time() - start)
?
    #数据隔离还是共享？
from threading import Thread
n = 100
def func():
    global n    # 不要在子线程里随便修改全局变量
    n-=1
t_l = []
for i in range(100):
    t = Thread(target=func)
    t_l.append(t)
    t.start()
for t in t_l:t.join()
print(n)
?
#守护线程
import time
from threading import Thread
def son1():
    while True:
        time.sleep(0.5)
        print(‘in son1‘)
def son2():
    for i in range(5):
        time.sleep(1)
        print(‘in son2‘)
t =Thread(target=son1)
t.daemon = True
t.start()
Thread(target=son2).start()
time.sleep(3)
# 守护线程一直等到所有的非守护线程都结束之后才结束
# 除了守护了主线程的代码之外也会守护子线程
?