并发编程-单元练习题

1、简述计算机操作系统中的“中断”的作用？

中断是指在计算机执行期间，系统内发生任何非寻常的或非预期的急需处理事件，使得cpu暂时中断当前正在执行的程序，
转去执行相应的事件处理程序。待处理完毕后又返回原来被中断处继续执行或调度新的进程执行的过程。
它使计算机可以更好更快利用有限的系统资源解决系统响应速度和运行效率的一种控制技术。
实时响应，系统调度。

2、简述计算机内存中的“内核态”和“用户态”；

内核态：cpu可以访问内存的所有数据，包括外围设备，例如硬盘，网卡，cpu也可以将自己从一个程序切换到另一个程序。
用户态：只能受限的访问内存，且不允许访问外围设备，占用cpu的能力被剥夺，cpu资源可以被其他程序获取。
为什么要有用户态和内核态？
    由于需要限制不同的程序之间的访问能力，防止他们获取别的程序的内存数据，或者获取外围设备的数据，并发送到网络，
    cpu划分出两个权限等级：用户态和内核态。

3、什么是进程、线程、协程、多线程和多进程、线程池和进程池

进程：
正在进行的一个过程或者说一个任务，负责执行任务的是cpu，进程用来把资源集中到一起，进程是资源单位，或者说是资源集合
线程：
线程是cpu的执行单位
多线程和多进程的区别：
0、创建进程的开销远大于创建线程的开销。
1、开启速度 在主进程下开启线程比 开启子进程快 （线程：线程先执行，在执行进程程序）
2、pid 对比：在主进程下开启多个线程（进程和线程pid相同）；在主进程下开启子进程（pid不相同）
3、同一进程内的线程共享该进程的数据（进程之间的地址是隔离的；同一进程内的线程共享该进程的地址空间）

协程：
协程：是单线程下的并发，又称微线程，纤程。
协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。
优点如下：
1. 协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级
2. 单线程内就可以实现并发的效果，最大限度地利用cpu
缺点如下：
1. 协程的本质是单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程
2. 协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程

进程池与线程池作用：
# 为什么建池 :我们必须对服务器开启的进程数或者线程数加以控制，让机器在一个自己可以承受的范围内运行
# 这就是进程池或线程池的作用

协程与线程的区别：
协程：单线程下的并发。协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。 
　　1.python的线程是属于内核级别的，即由操作系统控制调度（如单线程遇到io或执行时间过长就会被迫交出cpu执行权限，切换其他的线程运行） 
　　2.单线程内开启协程，一旦遇到io，就会从应用程序级别（而非操作系统）控制切换，以此来提升效率（！！非io操作的切换与效率无关）

4、请简述你对join、daemon方法的理解，举出它们在生产环境中的使用场景；

join: 等待一个任务执行完毕；可以将并发变成串行。
daemon: 
    守护进程（守护线程）会等待主进程（主线程）运行完毕后被销毁。
    运行完毕：
        1.对主进程来说，运行完毕指的是主进程代码运行完毕。
        2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕。

4、threading中Lock和RLock的相同点和不同点；

Lock():互斥锁，只能被acquire一次，可能会发生死锁情况。 
RLock():递归锁，可以连续acquire多次。
RLock = Lock + counter
    counter：记录了acquire的次数，直到一个线程所有的acquire都被release,其他线程才能获得资源。 

3、进程间通信方式有哪些？

进程指：正在执行的一个程序或者说一个任务，负责执行任务的是cpu。进程是用来把资源集中到一起的，进程是资源单位，或者说资源集合。

进程间通信(IPC)

消息队列（    队列 = 管道 + 锁）

管道(使用消息传递的)

有名管道（FIFO）

信号量

共享内存

套接字（socket）

4、简述你对管道、队列的理解；

管道通常指无名管道
1、它是半双工的（即数据只能在一个方向上流动），具有固定的读端和写端
2、它只能用于具有亲缘关系的进程中通信（也就是父与子进程或者兄弟进程之间）
3、数据不可反复读取了，即读了之后欢喜红区中就没有了
消息队列
1、消息队列是面向记录的，其中的消息具有特定的格式以及特定的优先级
2、消息队列独立于发送与接收进程。进程终止时，消息队列及其内容不会被删除。
3、消息队列可以实现消息随机查询。
 
    队列 = 管道 + 锁

5、请列举你知道的进程间通信方式；

队列，信号量(Semaphore),Event事件，定时器Timer,线程queue,进程池和线程池，异步调用+回调机制

6、什么是同步I/O，什么是异步I/O？

同步I/O操作：实际的I/O操作将导致请求进程阻塞，直到I/O操作完成。

异步I/O操作：实际的I/O操作不导致请求进程阻塞。

同步或者异步I/O主要是指访问数据的机制(即实际I/O操作的完成方式)，

同步 一般指主动请求并等待I/O操作完毕的方式，I/O操作未完成前，会导致应用进程挂起；

而异步 指用户进程触发IO操作以后便开始做自己的事情，而当IO操作已经完成的时候会得到IO

完成的通知（异步的特点就是通知）,这可以使进程在数据读写时也不阻塞。

7、请问multiprocessing模块中的Value、Array类的作用是什么？举例说明它们的使用场景

python 多进程通信Queue Pipe Value Array
queue和pipe用来在进程间传递消息;
Value + Array 是python中共享内存映射文件的方法;速度比较快.

8、请问multiprocessing模块中的Manager类的作用是什么？与Value和Array类相比，Manager的优缺点是什么？

Python multiprocessing.Manager（进程间共享数据）
Python中进程间共享数据，除了基本的queue，pipe和value+array外，还提供了更高层次的封装。使用multiprocessing.Manager可以简单地使用这些高级接口。
Manager支持的类型有list,dict,Namespace,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Value和Array。

 编程题

1、请写一个包含10个线程的程序，主线程必须等待每一个子线程执行完成之后才结束执行，每一个子线程执行的时候都需要打印当前线程名、当前活跃线程数量；

from threading import Thread,currentThread,activeCount
import time
def task(n):
    print('线程名：%s----%s'%(currentThread().name,n))
    time.sleep(1)
    print('数量：%s'%activeCount())
    
if __name__ == "__main__":
    t_li = []
    for i in range(10):
        t = Thread(target=task,args=(i,))
        t.start()
        t_li.append(t)
    for t in t_li:
        t.join()
        
    print('主,end----')

2、请写一个包含10个线程的程序，并给每一个子线程都创建名为"name"的线程私有变量，变量值为“Alex”；

from threading import Thread
def task(name):
    print( name,'is running')
if __name__ == '__main__':
    for i in range(10):
        name = f"alex{i}"
        t=Thread(target=task,args=(name,))
        t.start()
    print('主 end')

3、请使用协程写一个消费者生产者模型；

import gevent
from gevent import monkey;monkey.patch_all()
def consumer(i):
    print('消费',i)
def producter(i):
    print('生产',i)

if __name__ == '__main__':
    for i in range(10):
        g1 = gevent.spawn(producter,i)
        g2 = gevent.spawn(consumer,i)
        g1.join()
        g2.join()

-------------------------------------------

# 方法二、
def consumer():
    while True:
        x=yield
        print('消费',x)
def producter():
    c=consumer()
    next(c)
    for i in range(10):
        print('生产',i)
        c.send(i)
producter()

4、写一个程序，包含十个线程，子线程必须等待主线程sleep 10秒钟之后才执行，并打印当前时间；

from threading import Thread,Event,currentThread
import time,datetime

def task():
    print(f"{currentThread().getName()} 等待")
    event.wait()
    print(f"{currentThread().getName()},{datetime.datetime.now()} is running")
if __name__ == '__main__':
    event=Event()
    for i in range(10):
        t=Thread(target=task)
        t.start()
    time.sleep(10)
    event.set()
---------------------------------------------------------------------------------
from threading import Event
event.isSet()：返回event的状态值；
event.wait()：如果 event.isSet()==False将阻塞线程；
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
event.clear()：恢复event的状态值为False。

5、写一个程序，包含十个线程，同时只能有五个子线程并行执行；

方法二、用信号量的方法（Semaphore）

from threading import Thread, Semaphore, currentThread
import time


def task():
    sm.acquire()
    print(currentThread().getName())
    time.sleep(2)
    sm.release()


if __name__ == '__main__':
    sm = Semaphore(5)
    start = time.time()
    for i in range(10):
        t = Thread(target=task)
        t.start()
    t.join()
    stop = time.time()
    print(stop - start)

Thread-1
Thread-2
Thread-3
Thread-4
Thread-5
Thread-6
Thread-7
Thread-9
Thread-8
Thread-10
4.001999855041504

6、写一个程序 ，包含一个名为hello的函数，函数的功能是打印字符串“Hello, World!”，该函数必须在程序执行30秒之后才开始执行(不能使用time.sleep())；

from threading import Timer
def hello(name):
    print('%s say '%name,'Hello World!')
    
if __name__ == "__main__":
    t = Timer(5,hello,args=('alice',))
    t.start()

---------------------------------------------

from threading import Thread,Timer,currentThread,Event
def task():
    event.wait()
    print('Hello, World!')
if __name__ == '__main__':
        event=Event()
        t=Thread(target=task)
        t.start()
        event.wait(5)
        event.set()

7、写一个程序，利用queue实现进程间通信；

# from multiprocessing import Process,Queue
# import time,random,os
# def consumer(q,name):
#     while True:
#         res=q.get()
#         if res is None:break
#         time.sleep(random.randint(1,3))
#         print(f"{name},{res}")
#
# def producer(q,name,food):
#     for i in range(3):
#
#         res=f"{food},{i}"
#         q.put(res)
#         print(f"{name}生产了{i}")
#
# if __name__ == '__main__':
#     q=Queue()
#     #生产者们:即厨师们
#     p1=Process(target=producer,args=(q,'egon','包子'))
#
#     #消费者们:即吃货们
#     c1=Process(target=consumer,args=(q,'alex'))
#
#     #开始
#     p1.start()
#     c1.start()
#     p1.join()
#     q.put(None)
#     print('主')

# ---------------------------------------------------------------------
from multiprocessing import Process,JoinableQueue
import time,random,os
def producer(q,name):
    for i in range(3):
        res=f"{name},{i}"
        q.put(res)
        print(res)
def consumer(q,name):

    while True:
        res=q.get()
        print(f'{name} eating---',res)
        q.task_done()

if __name__ == '__main__':
    q=JoinableQueue()
    p1=Process(target=producer,args=(q,'aa'))
    c1=Process(target=consumer,args=(q,'bb'))
    c1.daemon=True
    p1.start()
    c1.start()
    p1.join()
    print('主')

8、写一个程序，利用pipe实现进程间通信；

from multiprocessing import Process,Pipe
def task(conn):
    conn.send('hello world')
    conn.close()
    
if __name__ == "__main__":
    parent_conn,child_conn = Pipe()
    p = Process(target=task,args=(child_conn,))
    p.start()
    p.join()
    print(parent_conn.recv())

9、使用selectors模块创建一个处理客户端消息的服务器程序；

# server  blocking IO  
import socket
server = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
server.bind(('127.0.0.1',8080))
server.listen(5)
while True:
    conn,addr = server.accept()
    print(addr)
    while True:
        try:
            data = conn.recv(1024)
            if not data: break
            conn.send(data.upper())
        except Exception as e:
            print(e)
            break
            
# server  IO多路复用 selectors 会根据操作系统选择select poll epoll           
import socket
import selectors
sel = selectors.DefaultSelector()
def accept(server_fileobj,mask):
    conn,addr = server_fileobj.accept()
    print(addr)
    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())
    except Exception:
        print('closeing...',conn)
        sel.unregister(conn)
        conn.close()
        
server_fileobj = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
server_fileobj.bind(('127.0.0.1',8080))
server_fileobj.listen(5)
server_fileobj.setblocking(False)
sel.register(server_fileobj,selectors.EVENT_READ,accept)
while True:
    events = sel.select()
    for sel_obj,mask in events:
        callback = sel_obj.data
        callback(sel_obj.fileobj,mask)
        
# client
# -*- coding:utf-8 -*-
import socket
client = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
client.connect(('127.0.0.1',8080))
while True:
    msg = input('>>>:').strip()
    if not msg:continue
    client.send(msg.encode('utf-8'))
    data = client.recv(1024)
    print(data.decode('utf-8'))

10、使用socketserver创建服务器程序时，如果使用fork或者线程服务器，一个潜在的问题是，恶意的程序可能会发送大量的请求导致服务器崩溃，请写一个程序，避免此类问题；

# server socketserver 模块内部使用IO多路复用 和多进程/多线程
import socketserver
class Handler(socketserver.BaseRequestHandler):
    def handle(self):
        print('new connection:',self.client_address)
        while True:
            try:
                data = self.request.recv(1024)
                if not data:break
                print('client data:',data.decode())
                self.request.send(data.upper())
            except Exception as e:
                print(e)
                break
                
if __name__ == "__main__":
    server = socketserver.ThreadingTCPServer(('127.0.0.1',8080),Handler)
    server.serve_forever()

11、写一个程序，要求用户输入用户名和密码，要求密码长度不少于6个字符，且必须以字母开头，如果密码合法，则将该密码使用md5算法加密后的十六进制概要值存入名为password.txt的文件，超过三次不合法则退出程序；

def start():
    count = 3
    while count > 0:
        name = input('>>').strip()
        pwd = input('>>').strip()
        if len(pwd) >= 6 and pwd[0].isalpha():# isalpha() 方法检测字符串是否只由字母组成。
            pwd_md5 = hashlib.md5(pwd.encode('utf-8')).hexdigest()
            print('输入成功')
            obj = {'user_name': name, 'pass_word': pwd_md5}
            with open('password.txt','a',encoding='utf-8')as f:
                json.dump(obj,f)
            exit()
        else:
            if count == 0:
                print('三次错误')
                exit()
            count -=1
            print(f'输入错误还有{count}次机会')
            continue

if __name__ == '__main__':
    import hashlib,json
    start()

 12、写一个程序，使用socketserver模块，实现一个支持同时处理多个客户端请求的服务器，要求每次启动一个新线程处理客户端请求；

python中的socketserver模块，主要是用来提供服务器类，并且提供异步处理的能力。

服务端：
import socketserver
class Handler(socketserver.BaseRequestHandler):
    def handle(self):
        print('new connection:', self.client_address)
        while True:
            try:
                data = self.request.recv(1024)
                if not data: break
                print('client data:', data.decode())
                self.request.send(data.upper())
            except Exception as e:
                print(e)
                break


if __name__ == "__main__":
    server = socketserver.ThreadingTCPServer(('127.0.0.1', 8082), Handler)
    server.serve_forever()

客户端
模拟多个线程 ---客户端发送消息

 def start(ip_port):
    client = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    client.connect(ip_port)
    client.send(f"{currentThread().getName()} hello".encode('utf-8'))
    data = client.recv(1024)
    print(data.decode('utf-8'))

if __name__ == '__main__':
    import socket
    from threading import Thread,currentThread
    ip_port = ('127.0.0.1',8082)
    for i in range(20):
        t = Thread(target=start,args=(ip_port,))
        t.start()

 13、请使用协程写一个消费者生产者模型；

# def consumer():
#     while True:
#         x = yield
#         print('消费：',x)
#
# def producter():
#     c = consumer()
#     next(c)
#     for i in range(10):
#         print('生产：',i)
#         c.send(i)
#
# if __name__ == '__main__':
#     producter()

import gevent
from gevent import monkey;monkey.patch_all()
def consumer(i):
    print('消费',i)
def producter(i):
    print('生产',i)

if __name__ == '__main__':
    for i in range(10):
        g1 = gevent.spawn(producter,i)
        g2 = gevent.spawn(consumer,i)
        g1.join()
        g2.join()