Day34Pyhotn之路——网络编程

Posted 2020-10-16

• 守护进程

　　　　（1）守护进程在主进程代码运行结束后，就自动终止

　　　　（2）守护进程内无法开启子进程，否则报错AssertionError: daemonic processes are not allowed to have children

　　　　注意：进程之间相互独立的，主进程运行完毕后，守护进程随即结束

 

from multiprocessing import Process
import os,time, random
def func():
    print(‘%s ruing ‘%os.getpid())
    time.sleep(0.2)
    print(‘%s end‘%os.getpid())
    # p = Process(target=func) #在子进程内创建以个子进程，报错
    # p.start()

if __name__ == ‘__main__‘:
    P = Process(target=func)
    P.daemon = True
    P.start()
    # P.join()# 使主进程必须等到子进程运行完毕
    print(‘主‘)

 

from multiprocessing import Process
from threading import Thread
import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

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

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

    p1.daemon=True
    p1.start()
    p2.start()
    print("main-------")#打印出“main”意味着主进程结束，守护进程也就结束了，然而p2并不是守护进行，主进程需等到p2运行完毕，才能结束，否则会产生孤儿进程

迷惑人的例子

• 守护线程

守护线程需要等到其他非守护线程执行完，才结束。

 

from threading import Thread
import time, os, random

def func():
    print(‘%s is ruing‘% os.getpid())
    time.sleep(2)
    print("%s end"% os.getpid())
    t = Thread(target=time.sleep,args=(3,))
    t.start()

if __name__ == ‘__main__‘:
    t = Thread(target=func)
    t.start()
    print(‘主‘)

 

from multiprocessing import Process
from threading import Thread
import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

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

if __name__ == ‘__main__‘:
    t1=Thread(target=foo)
    t2 = Thread(target=bar)

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

迷惑人的例子

• 互斥锁：把并发变成了串行，牺牲效率提高安全性

进程同步锁

def func():
    # mutex.acquire()
    print(‘%s is runing‘%os.getpid())
    time.sleep(2)
    print(‘%s end‘%os.getpid())
    time.sleep(random.randint(1,3))
    # mutex.release()

if __name__ == ‘__main__‘:
    p = Process(target=func)
    p1 = Process(target=func)
    p2= Process(target=func)
    p3 = Process(target=func)
    p.start()
    p1.start()
    p2.start()
    p3.start()
    # mutex = Lock()
    # p = Process(target=func,args=(mutex,))
    # p1 = Process(target=func,args=(mutex,))
    # p2 = Process(target=func,args=(mutex,))
    # p3 = Process(target=func,args=(mutex,))
    # p.start()
    # p1.start()
    # p2.start()
    # p3.start()

模拟抢票：

from multiprocessing import Process,Lock
import os,time
import json
from random import randint
# def sehch():
#     with open(‘text.txt‘,encoding=‘utf-8‘)as f:
#         dic = json.load(f)
#     print(‘%s 剩余票数%s‘%(os.getpid(),dic[‘count‘]))
# def get():
#
#     with open(‘text.txt‘,encoding=‘utf-8‘)as f:
#         dic = json.load(f)
#     if dic[‘count‘] > 0:
#         dic[‘count‘] -=1
#         time.sleep(randint(1,3))
#         with open("text.txt",‘w‘,encoding=‘utf-8‘)as f:
#             json.dump(dic,f)
#             print(‘购票成功%s‘%os.getpid())
# def task(mutex):
#     sehch()
#     mutex.acquire()
#     get()
#     mutex.release()
# if __name__ == ‘__main__‘:
#     mutex = Lock()
#     for i in range(20):
#         p = Process(target=task,args=(mutex,))
#         p.start()

模拟抢票

•  互斥锁——线程

 

from threading import Thread
import time, os, random
n = 100
def func():
    # global n
    # mutex.acquire()
    # tem = n
    # n = tem -1
    # mutex.release()
    global n
    # with mutex:
    tem = n
    time.sleep(0.1)
    n = tem -1
if __name__ == ‘__main__‘:
    li = []
    # mutex = Lock()
    for i in range(100):
        t = Thread(target=func)
        li.append(t)
        t.start()  #在创建进程和线程时，使用的Thread, Process 时需要使用start进行启动
    for t in li:        #使用with时，可以省略使用acquire，和release 两个函数
        t.join()
    print(n)

结果为99

 

from threading import Thread
import time, os, random
n = 100
def func():
    # global n
    # mutex.acquire()
    # tem = n
    # n = tem -1
    # mutex.release()
    global n
    with mutex:
        tem = n
        n = tem -1
if __name__ == ‘__main__‘:
    li = []
    mutex = Lock()
    for i in range(100):
        t = Thread(target=func)
        li.append(t)
        t.start()  #在创建进程和线程时，使用的Thread, Process 时需要使用start进行启动
    for t in li:        #使用with时，可以省略使用acquire，和release 两个函数
        t.join()
    print(n)

jion 和Lock的区别：lock 好处想锁那一块就锁那一块，可以只锁修改共享数据那一块，让局部变成串行。

互斥锁：把并发变成串行，保证大家修改同一块数据的安全，

互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去，如果指定信号量为3，那么来一个人获得一把锁，计数加1，当计数等于3时，后面的人均需要等待。一旦释放，就有人可以获得一把锁 信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念

 

from multiprocessing import Process,Semaphore
import time, os, random
def  task(sm):
    with sm:
        print("%s 快递"%os.getpid())
        time.sleep(1)
if __name__ == ‘__main__‘:
    sm = Semaphore(3)
    for i in range(10):
        p =Process(target=task, args=(sm,))
        p.start()

• 进程Queue 与线程queue

Queue（3）可以限制放入的数量，但是也可以限制取出的数量，如果取出的数量大于放入的数量，则会处于停顿状态，一直等待放入数据

（2）Queue 是先进先出

 

from multiprocessing import    Queue
# q = Queue(3)
# q.put({"a":1})
# q.put({"a":2})
# print(q.get())

 

• 线程queue队列

import queue
q = queue.Queue(3)
q.put(‘a‘)
print(q.get())

• 优先级队列：

import queue
q = queue.PriorityQueue(3)
q.put((1,21))               #put()参数为元组，元组第一个为优先级，数越小，优先级越高

print(q.get())

“”“
(1, 21)
”“”

• 堆栈：后进先出

 

import queue
q = queue.LifoQueue(3)
q.put((1,21))

print(q.get())
"""(1, 21)"""

•  生产者消费者模型

 

from multiprocessing import Process, Queue
import time, os, random

def gieve(q):
    for i in range(10):
        res = ‘包子%s‘%i
        time.sleep(0.5)
        q.put(res)
        print(‘%s 生产了 包子%s‘%(os.getpid(),res))
    q.put(None)
def get_e(q):
    while True:
        res = q.get()
        if res == None:
            break
        print(‘%s 吃了 %s‘%(os.getpid(),res))
        time.sleep(random.randint(1,3))

if __name__ == ‘__main__‘:
    q = Queue()
    a = Process(target=gieve,args=(q,))
    s = Process(target=get_e,args=(q,))
    a.start()
    s.start()
    # a.join()
    # q.put(None)
    print(‘主‘)

 

 

 

 

 

from multiprocessing import Process, Queue
import time, os, random

def gieve(q):
    for i in range(10):
        res = ‘包子%s‘%i
        time.sleep(0.5)
        q.put(res)
        print(‘%s 生产了 包子%s‘%(os.getpid(),res))
    q.put(None)
def get_e(q):
    while True:
        res = q.get()
        if res == None:
            break
print(‘%s 吃了 %s‘%(os.getpid(),res))
        time.sleep(random.randint(1,3))

if __name__ == ‘__main__‘:
    q = Queue()
    a = Process(target=gieve,args=(q,))
    s = Process(target=get_e,args=(q,))
    a.start()
    s.start()
    # a.join()
    # q.put(None)
print(‘主‘)

• 信号量

信号量。信号量是用于维持有限资源访问的信号。它们和锁类似，除了它们可以允许某个限制下的多个访问。
它就像电梯一样只能够容纳几个人。一旦达到了限制，想要使用资源的进程就必须等待。其它进程释放了信号量之后，它才可以获得。

例如，假设有许多进程需要读取中心数据库服务器的数据。如果过多的进程同时访问它，它就会崩溃，所以限制连接数量就是个好主意。
如果数据库只能同时支持N=2的连接，我们就可以以初始值N=2来创建信号量。

 

>>> from threading import Semaphore
>>> db_semaphore = Semaphore(2) # set up the semaphore
>>> database = []
>>> def insert(data):
        db_semaphore.acquire() # try to acquire the semaphore
        database.append(data)  # if successful, proceed
        db_semaphore.release() # release the semaphore
>>> insert(7)
>>> insert(8)
>>> insert(9)

信号量的工作机制是，所有进程只在获取了信号量之后才可以访问数据库。只有N=2个进程可以获取信号量，其它的进程都需要等到其中一个进程释放了信号量，之后在访问数据库之前尝试获取它。