Day9 进程同步锁 进程队列 进程池 生产消费模型 进程池 paramike模块

进程同步锁：

当运行程序的时候，有可能你的程序同时开多个进程，开进程的时候会将多个执行结果打印出来，这样的话打印的信息都是错乱的，怎么保证打印信息是有序的呢？

其实也就是相当于让进程独享资源。

from multiprocessing import Process,Lock    #引用函数
import time
def work(name,mutex):    
    mutex.acquire()    #在这里加入锁
    print('task <%s> is runing' %name)
    time.sleep(2)
    print('task <%s> is done' % name)
    mutex.release()    #加完锁以后必须需要解锁

if __name__ == '__main__':
    mutex=Lock()
    p1=Process(target=work,args=('egon',mutex))
    p2=Process(target=work,args=('alex',mutex))
    p1.start()
    p2.start()
    print('主')

比如说模拟抢票的功能：

要先写一个文本   ("count":1)   就记个数就行

import json
import os
import time
from multiprocessing import Process,Lock
def search():
    dic=json.load(open('db.txt'))
    print('33[32m[%s] 看到剩余票数<%s>33[0m' %(os.getpid(),dic['count']))
def get_ticket():
    dic = json.load(open('db.txt'))
    time.sleep(0.5) #模拟读数据库的网络延迟
    if dic['count'] > 0:
        dic['count']-=1
        time.sleep(0.5)  # 模拟写数据库的网络延迟
        json.dump(dic,open('db.txt','w'))
        print('33[31m%s 购票成功33[0m' %os.getpid())
def task(mutex):
    search()
    mutex.acquire()
    get_ticket()
    mutex.release()
if __name__ == '__main__':
    mutex=Lock()
    for i in range(10):
        p=Process(target=task,args=(mutex,))
        p.start()

---

进程队列：

共享内存的方式：

from multiprocessing import Process,Manager,Lock     #Manager共享内存函数

def task(dic,mutex):
    with mutex:
        dic['count']-=1

if __name__ == '__main__':
    mutex=Lock()
    m=Manager()
    dic=m.dict({'count':100})
    p_l=[]
    for i in range(100):
        p=Process(target=task,args=(dic,mutex))
        p_l.append(p)
        p.start()

    for p in p_l:
        p.join()
    print(dic)

队列：

进程彼此之间隔离，要实现进程间通信

from multiprocessing import Queue    #引用函数
q=Queue(3)      #意味着你队列长队最大为三

q.put('first')
q.put('second')
q.put('third')
# q.put('fourth')    #满了的话会一直卡住

print(q.get())
print(q.get())
print(q.get())
print(q.get())

#了解
# q=Queue(3)
# 
# q.put('first',block=False)
# q.put('second',block=False)
# q.put('third',block=False)    #这样的话队列满了就会抛出异常，不会卡在这里
# # q.put_nowait('fourth') 
21 #q.put('fourth',block=False)
# q.put('fourth',timeout=3)     #指定抛出时间，如果3秒后队列还是满的抛出异常

---

生产者消费者模型：

正常情况下，一般都是生产者预先生产出商品，然后等着消费者来买。

（实际情况可能是有多个生产者，多个消费者）

from multiprocessing import Process, JoinableQueue
import time, os


def producer(q, name):
    for i in range(3):
        time.sleep(1)
        res = '%s%s' % (name, i)
        q.put(res)
        print('33[45m<%s> 生产了 [%s]33[0m' % (os.getpid(), res))
    q.join()　　#对应下边的task_done

def consumer(q):
    while True:
        res = q.get()
        time.sleep(1.5)
        print('33[34m<%s> 吃了 [%s]33[0m' % (os.getpid(), res))
        q.task_done()    #代表我这个进程我已经取走了，发给生产者，对应生产者要有个jion（）

if __name__ == '__main__':
    q = JoinableQueue()

    # 生产者们：即厨师们
    p1 = Process(target=producer, args=(q, '包子'))
    p2 = Process(target=producer, args=(q, '饺子'))
    p3 = Process(target=producer, args=(q, '馄饨'))

    # 消费者们：即吃货们
    c1 = Process(target=consumer, args=(q,))
    c2 = Process(target=consumer, args=(q,))

    c1.daemon=True    #如果消费者不取完的话，程序无法结束
    c2.daemon=True　　 #这里主进程运行完，子进程要结束掉
    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()


    p1.join()

    print('主')

---

进程池：

from multiprocessing import Pool
import os,time

def work(n):
    print('task <%s> is runing' %os.getpid())
    time.sleep(2)
    return n**2
if __name__ == '__main__':
    # print(os.cpu_count())
    p=Pool(4)    #定义进程池的大小，如果不定义一般是内核数
    # for i in range(10):
    #     res=p.apply(work,args=(i,))    #向进程池里边添加进程，同步提交
    #     print(res)

    res_l=[]
    for i in range(10):
        res=p.apply_async(work,args=(i,))    #向进程池异步提交，只管扔进去，不管是否执行完成
        res_l.append(res)

    p.close()
    p.join()
    #
    # for res in res_l:
    #     print(res.get())

进程池之回调函数：

需要回调函数的场景，进程池中任何一个任务一旦处理完成，就立即告诉主进程，我好了，你可以处理我的结果了，主进程调用一个函数去处理该结果，该函数即回调函数。

import requests #pip3 install requests
import os,time
from multiprocessing import Pool
def get_page(url):
    print('<%s> get :%s' %(os.getpid(),url))
    respone = requests.get(url)
    if respone.status_code == 200:
        return {'url':url,'text':respone.text}

def parse_page(dic):
    print('<%s> parse :%s' %(os.getpid(),dic['url']))
    time.sleep(0.5)
    res='url:%s size:%s
' %(dic['url'],len(dic['text'])) #模拟解析网页内容
    with open('db.txt','a') as f:
        f.write(res)


if __name__ == '__main__':
    p=Pool(4)
    urls = [
        'http://www.baidu.com',
        'http://www.baidu.com',
        'http://www.baidu.com',
        'http://www.baidu.com',
        'http://www.baidu.com',
        'http://www.baidu.com',
        'http://www.baidu.com',
    ]


    for url in urls:
        p.apply_async(get_page,args=(url,),callback=parse_page)


    p.close()
    p.join()
    print('主进程pid:',os.getpid())

---

paramike模块：

paramike模块是一个用作做远程控制的模块，使用该模块可以对远程服务器进行命令或者文件操作。值得一提的是，fabric和ansible内部的远程管理就是使用的paramike模块。

#需要下载安装

#pip3 install paramiko

远程连接功能：

import paramiko

# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname='120.92.84.249', port=22, username='root', password='123QWEasd')

# 执行命令
stdin, stdout, stderr = ssh.exec_command('df')
# 获取命令结果
result = stdout.read()
print(result.decode('utf-8'))
# 关闭连接
ssh.close()

这是基于账号密码来访问客户端的。

另外一种方式是基于秘钥的，

现在服务端制作一个秘钥，ssh-keygen制作秘钥，sz可以下载到window桌面。

客户端要用，肯定要基于服务端有认证文件,利用 ssh-copy-id -i 用户@ip

import paramiko

private_key = paramiko.RSAKey.from_private_key_file('id_rsa')   #秘钥的文件位置

# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname='120.92.84.249', port=22, username='root', pkey=private_key)

# 执行命令
stdin, stdout, stderr = ssh.exec_command('df')
# 获取命令结果
result = stdout.read()
print(result.decode('utf-8'))
# 关闭连接
ssh.close()

 上传下载：

import paramiko

transport = paramiko.Transport(('120.92.84.249', 22))
transport.connect(username='root', password='123QWEasd')

sftp = paramiko.SFTPClient.from_transport(transport)
# 将location.py 上传至服务器 /tmp/test.py
sftp.put('id_rsa', '/tmp/test.rsa')
# 将remove_path 下载到本地 local_path
# sftp.get('remove_path', 'local_path')

transport.close()