百万年薪python之路 -- 并发编程之 多线程 一

Posted 2022-08-30 zhangchaoyin

多线程

1.进程: 生产者消费者模型

一种编程思想,模型,设计模式,理论等等,都是交给你一种编程的方法,以后遇到类似的情况,套用即可

生产者与消费者模型的三要素:

1. 生产者:产生数据的
2. 消费者:接收数据做进一步处理的
3. 容器: 缓存区(队列) 起到缓冲的作用,平衡生产力与消费者,解耦

2.线程的理论知识

1. 什么是线程

   一条流水线的工作流程

   进程: 在内存中开启一个进程空间,然后将主进程的所有的资源复制一份,然后调用cpu去执行这些代码.

   进程是资源调度的基本单位,而线程是cpu的最小执行单位

   

   进程的开启: 进程会在内存中开辟一个进程空间,将主进程的数据全部复制一份,线程会执行里面的代码

2. 线程vs进程

   1. 开启进程的开销非常大,比线程的开销大很多.
   2. 开启线程的速度非常快,要比进程快几十倍到上百倍.
   3. 同一个进程内,线程与线程之间可以共享数据, 进程与进程之间需要借助队列等方法实现通信.

3. 线程的应用

   并发: 一个cpu看起来像同时执行多个任务

   单个进程开启三个线程,并发的执行任务.

   开启三个进程并发的执行任务.

   开启多线程的优点: 数据共享,开销小,速度快.

主线程和子线程没有主次之分

那么一个进程谁在干活?

? 一个主线程在干活,当主线程执行完代码后,还得等待其他线程执行完,才能退出进程.

3.开启线程的两种方式

**线程不需要在if _ _ name _ _ == ‘_ _ main _ _‘:语句下**

第一种:

from threading import Thread
import time

def task(name):
    print(f"name is running")
    time.sleep(1)
    print(f"name is gone")


if __name__ == '__main__':
    t1 = Thread(target=task,args=("zcy",))
    t1.start()
    print("==main Threading==") # 线程没有主次之分

第二种:

from threading import Thread
import time

class MyThread(Thread):

    def __init__(self,name,lst,s):
        super(MyThread, self).__init__()
        self.name = name
        self.lst =lst
        self.s = s

    def run(self):
        print(f"self.name is running")
        time.sleep(1)
        print(f"self.name is gone")

if __name__ == '__main__':
    t1 = MyThread("zdr",[1,2,3],"180")
    t1.start()
    print("==main thread==")

4.线程vs进程的代码对比

1. 开启速度对比

   # 多进程
   from multiprocessing import Process
   
   def work():
       print('hello')
   
   def task():
       print('bye')
   
   if __name__ == '__main__':
       # 在主进程下开启线程
       t1 = Process(target=work)
       t2 = Process(target=task)
       t1.start()
       t2.start()
       print('main thread/process')

   # 多线程
   from threading import Thread
   import time
   
   def task(name):
       print(f"name is running")
       time.sleep(1)
       print(f"name is gone")
   
   
   
   if __name__ == '__main__':
       t1 = Thread(target=task,args=("zdr",))
       t2 = Thread(target=task,args=("zcy",))
       t3 = Thread(target=task,args=("zfy",))
       t4 = Thread(target=task,args=("lfz",))
       t1.start()
       t2.start()
       t3.start()
       t4.start()
       print('==main thread==') # 线程是没有主次之分

2. 对比pid

   # 进程
   from multiprocessing import Process
   import time
   import os
   
   def task():
       print(f"子进程:os.getpid()")
       print(f"主进程:os.getppid()")
   
   
   if __name__ == '__main__':
       p1 = Process(target=task)
       p2 = Process(target=task)
       p1.start()
       p2.start()
   
       print(f"==mainos.getpid()")

   # 主线程
   from threading import Thread
   import os
   
   def task():
       print(os.getpid())
   
   
   if __name__ == '__main__':
       t1 = Thread(target=task)
       t2 = Thread(target=task)
       t1.start()
       t2.start()
       print(f"===main thread:os.getpid()")

3. 同一个进程内线程共享内部数据

   from threading import Thread
   import os
   
   
   x = 3
   def task():
       global x
       x = 100
   
   
   if __name__ == '__main__':
       t1 = Thread(target=task)
       t1.start()
       print(f"===main thread:x")
   
   
   # 同一个进程内的资源数据对于这个进程的多个线程来说是共享的.

小练习:

import multiprocessing
import threading

import socket
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.bind(('127.0.0.1',8080))
s.listen(5)

def action(conn):
    while True:
        data=conn.recv(1024)
        print(data)
        conn.send(data.upper())

if __name__ == '__main__':

    while True:
        conn,addr=s.accept()


        p=threading.Thread(target=action,args=(conn,))
        p.start()

多线程并发的socket服务端

import socket

s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect(('127.0.0.1',8080))

while True:
    msg=input('>>: ').strip()
    if not msg:continue

    s.send(msg.encode('utf-8'))
    data=s.recv(1024)
    print(data)

客户端

5.线程的相关其他方法

线程对象的方法:

? 线程.isAlive() # 判断线程是否存活

? 线程.getname() # 获取线程名

? 线程.setname() # 设置线程名 ***

threading模块的方法:

? threading.currentThread() # 获取当前进程的对象

? threading.enumerate() # 返回一个列表,包括所有的线程对象

? threading.activeCount() # 返回一个数字,表示有多少个线程还存活

from threading import Thread, currentThread, enumerate,activeCount
import os
import time

x = 3
def task():
    # print(currentThread())
    # time.sleep(1)
    print("123")


if __name__ == '__main__':
    t1 = Thread(target=task,name="xc-1")
    t2 = Thread(target=task,name="xc-2")
    # name 设置线程名
    t1.start()
    t2.start()
    # time.sleep(2)
    # print(t1.isAlive()) # 判断线程是否存活
    # print(t1.getName()) # 获取线程名
    # t1.setName("zcy-01")
    # print(t1.name)      # ***



    # threading方法
    # print(currentThread())     # 获取当前线程的对象
    # print(currentThread().name)     # 获取当前线程的对象
    print(enumerate())      # 返回一个列表,包含所有的线程对象
    print(activeCount())
    print(f"===main thread:os.getpid()")

6.join和守护线程

? join: 阻塞 告知主线程要等待子线程执行完毕之后再执行主线程

# 线程join
from threading import Thread
import time

def task(name):
    print(f"name is running")
    time.sleep(1)
    print(f'name is gone')


if __name__ == '__main__':
    start_time = time.time()
    t1 = Thread(target=task,args=("zdr",))
    t2 = Thread(target=task,args=("zcy",))
    t3 = Thread(target=task,args=("zfy",))

    t1.start()
    t1.join()
    t2.start()
    t2.join()
    t3.start()
    t3.join()

    print(f"===main thread:time.time() - start_time")

守护线程:

无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁

需要强调的是：运行完毕并非终止运行

#1.对主进程来说，运行完毕指的是主进程代码运行完毕

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

详细解释：

#1 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束，

#2 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

先对比一下守护进程:

from multiprocessing import Process
import time

def foo():
    print(123)
    time.sleep(1)
    print("end123")



def bar():
    print(456)
    time.sleep(2)
    print("end456")


if __name__ == '__main__':
    p1 = Process(target=foo)
    p2 = Process(target=bar)

    p1.daemon = True
    p1.start()
    p2.start()
    print('====main====')

守护线程:

from threading import Thread
import time


def sayhi(name):
    print('bye~')
    time.sleep(2)
    print(f'name say hello ')


if __name__ == '__main__':
    t = Thread(target=sayhi,args=('zcy',))
    # t.setDaemon(True)
    t.daemon = True
    t.start()

    print('主线程')

from threading import Thread
import time


def foo():
    print(123)  # 1
    time.sleep(1)
    print('end123') # 4


def bar():
    print(456)  # 2
    time.sleep(3)
    print('en456')  # 3


t1 = Thread(target=foo)
t2 = Thread(target=bar)


t1.daemon = True
t1.start()
t2.start()
print('=====main====')  # 3



结果:
123
456
=====main====
end123
en456   

# 主线程什么时候结束?
# 主线程等待非守护子线程结束之后,结束

from threading import Thread
import time

def foo():
    print(123)  # 1
    time.sleep(3)
    print("end123")  

def bar():
    print(456)   # 2
    time.sleep(1)
    print("end456")  # 4


t1=Thread(target=foo)
t2=Thread(target=bar)

t1.daemon=True
t1.start()
t2.start()
print("main-------")  # 3


结果:
123
456
main-------
end456

7.互斥锁

from threading import Thread
import time
import random

x = 100

def task():
    time.sleep(random.randint(1,2))
    global x
    temp = x
    time.sleep(random.randint(1,3))
    temp = temp - 1
    x = temp


if __name__ == '__main__':
    l = []
    for i in range(100):
        t = Thread(target=task)
        l.append(t)
        t.start()

    for i in l:
        i.join()

    print(f"main:x")
    
    
    
    
# 多个任务共抢一个数据,要保证数据的安全性,要让他们串行
# 给线程加锁
from threading import Thread
from threading import Lock
import time
import random
x = 100

def task(lock):

    lock.acquire()
    # time.sleep(random.randint(1,2))
    global x
    temp = x
    time.sleep(0.01)
    temp = temp - 1
    x = temp
    lock.release()


if __name__ == '__main__':
    mutex = Lock()
    l1 = []
    for i in range(100):
        t = Thread(target=task,args=(mutex,))
        l1.append(t)
        t.start()

    time.sleep(3)
    print(f'主线程x')