Python基础之网络并发编程

一、任务运行的多种方式

1、同步与异步

根据进程和函数之间的通讯机制,函数可分为同步和异步

同步与异步是为了表达任务的提交方式

  • 同步:指提交任务后,直接得到任务的最终结果
    • eg:在取款机前排队取款,直到排队取到钱之后离开
  • 异步:指提交任务后,不等待任务的结果,直接去做另外的事情,任务完成后结果会反馈回来
    • eg:指在银行取钱,领到取票机的小票后,不用一直排队,可以去做其他的事情,到号时,银行会叫号

2、阻塞与非阻塞

指函数或方法调用的时候,结果是否会立即返回

  • 阻塞:执行任务后会立即返回任务的结果
  • 非阻塞:执行任务后,并不会立即返回结果,就绪态、运行态

3、综合使用

综合使用指同步与非同步、阻塞与非阻塞四种状态两两结合使用

  • 同步阻塞
    • 指调用任务后,遇到阻塞,但是过程中什么也干不了,只能等待任务的最终结果
  • 同步非阻塞
    • 指调用任务后,任务在顺利进行,运行通畅,直到获取任务结果
  • 异步阻塞
    • 指调用任务后,任务遇到阻塞,没有办法直接获取到最终结果,但是调用者可以去处理其他的事情,并不会影响调用者处理其他事情,等原任务执行结束后反馈结果即可
  • 异步非阻塞
    • 指调用者提交一个任务后,立即提交其他的任务,并且提交的多个任务都可以通畅运行,立即获得结果,这种状态的效率极高

二、创建进程的多种方式

​ 通常,用户使用计算机,创建进程都是通过双击电脑桌面上的应用图标来创建进程,但我们程序员在使用python时,可以直接通过代码来创建进程,并且可以同时创建多个进程,且进程与进程之间互不干扰,可以高效的提高任务完成率

1、双击电脑桌面图标

2、通过代码创建

须知:

  • windows
    • 以导入模块的方式创建进程
  • linux/mac
    • 以拷贝代码的方式创建进程

1、代码创建进程

方式一

​ 函数方式创建

from multiprocessing import Process
import time


def task(name):
     print('task is running',name)
     time.sleep(3)
     print('task is over',name)
        
if __name__ == '__main__':
     # p1 = Process(target=task, args=('jason',))  # 位置参数
     p1 = Process(target=task, kwargs={'name':'jason123'})  # 关键字参数
     p1.start()  # 异步 告诉操作系统创建一个新的进程 并在该进程中执行task函数
     task()  # 同步
     print('主')

方式二

​ 继承类的方式创建

from multiprocessing import Process
import time


class MyProcess(Process):
    def __init__(self, name, age):
        super().__init__()
        self.name = name
        self.age = age
        

    def run(self):
        print('run is running', self.name, self.age)
        time.sleep(3)
        print('run is over', self.name, self.age)


if __name__ == '__main__':
    obj = MyProcess('jason', 123)
    obj.start()
    print('主')

2、进程间数据隔离

​ 进程隔离是指在同一台计算机上多个同时运行的进程,进程与进程之间数据默认是隔离的

​ eg:在主进程定义一个变量,使用主进程创建多个进程,在创建的多个进程内修改、重新定义主进程的变量,每个进程内的变量互不干扰,包括主进程。

代码表现:

from multiprocessing import Process
import time

money = 1000


def task():
    global money
    money = 666
    print('子进程的task函数查看money', money)


if __name__ == '__main__':
    p1 = Process(target=task)
    p1.start()  # 创建子进程
    time.sleep(3)  # 主进程代码等待3秒
    print(money)  # 主进程代码打印money

3、进程join方法

​ 使用主进程创建多个子进程后,代码运行到主进程创建的子进程代码后,通常子进程代码是异步执行,并不会干扰主进程代码执行顺序,通过join方法可使异步执行的代码变为同步执行,主进程会等待子进程代码运行结束后向下执行,join方法需定义在子进程启动代码后,否则会影响同步执行的结果

使用方法:

​ 通过定义的子进程对象‘点’join()的方式使用

代码表现:

from multiprocessing import Process
import time


def task(name, n):
    print('%s is running' % name)
    time.sleep(n)
    print('%s is over' % name)


if __name__ == '__main__':
    p1 = Process(target=task, args=('jason1', 1))
    p2 = Process(target=task, args=('jason2', 2))
    p3 = Process(target=task, args=('jason3', 3))
    # p.start()  # 异步
    '''主进程代码等待子进程代码运行结束再执行'''
    # p.join()
    # print('主')
    start_time = time.time()
    p1.start()
    p1.join()
    p2.start()
    p2.join()
    p3.start()
    p3.join()
    # p1.join()
    # p2.join()
    # p3.join()
    print(time.time() - start_time)  # 3秒多

4、IPC机制

IPC>>>:进程间通讯机制

相当于消息队列,储存数据的地方,同一主进程下的所有子进程共同存、取数据的地方,大家都可以使用

使用方法:

​ 1、导入multiprocessing模块下的Queue

​ 2、通过赋值变量的方式创建消息队列

​ 3、消息队列后方括号内为储存数据的个数

​ 4、对象‘点’put():存入

​ 5、对象‘点’get():取

代码用法:

from multiprocessing import Queue


q = Queue(3)  # 括号内可以指定存储数据的个数
# 往消息队列中存放数据
q.put(111)
# print(q.full())  # 判断队列是否已满
q.put(222)
q.put(333)
# print(q.full())  # 判断队列是否已满
# 从消息队列中取出数据
print(q.get())
print(q.get())
# print(q.empty())  # 判断队列是否为空
print(q.get())
# print(q.empty())  # 判断队列是否为空
# print(q.get())
print(q.get_nowait())

"""
full() empty() 在多进程中都不能使用!!!
"""


from multiprocessing import Process, Queue


def product(q):
    q.put('子进程p添加的数据')

def consumer(q):
    print('子进程获取队列中的数据', q.get())


if __name__ == '__main__':
    q = Queue()
    # 主进程往队列中添加数据
    # q.put('我是主进程添加的数据')
    p1 = Process(target=consumer, args=(q,))
    p2 = Process(target=product, args=(q,))
    p1.start()
    p2.start()
    print('主')

三、生产者消费者模型

  • 生产者:

    • 负责生产数据的代码

      eg: 回想使用爬虫,在网页上获取数据

  • 消费者:

    • 负责处理数据的代码

    ​ eg:类似于正则,对数据进行处理,分类

该模型除了有生产者和消费者之外还必须有消息队列(只要是能够提供数据保存服务和提取服务的理论上都可以)

四、进程对象的多种方法

1、查看进程号

方式一:

1、导入multiprocessing模块下current_process方法
2、在进程下使用currnet_process().pid
3、该方法可直接查看进程的 '父' 进程,以及进程号

方式二:

1、导入os模块
2、os.getpid()
	查看当前进程的进程号
3、os.getppid()
	查看进程的 '父' 进程

2、终止进程

1、通过进程对象名‘点’terminate()
	可直接结束进程
ps: 计算机操作系统有很多命令可以直接杀死(结束)进程

3、判断进程是否存活

1、通过进程名‘点’is_alive()
	存活:Trun
	死亡:False

4、守护进程

​ 守护进程是主进程创建的进程

  • 守护进程会在主进程代码运行结束后就立即终止
  • 守护进程内无法创建子进程,否则报错
  • 守护进程和主进程之间是相互独立的

代码用法:

from multiprocessing import Process
import time


def task(name):
    print('德邦总管:%s' % name)
    time.sleep(3)
    print('德邦总管:%s' % name)


if __name__ == '__main__':
    p1 = Process(target=task, args=('大张红',))
    p1.daemon = True
    p1.start()
    time.sleep(1)
    print('恕瑞玛皇帝:小吴勇嗝屁了')

5、僵死进程与孤儿进程

5、1.僵尸进程

​ 僵尸进程是指进程运行结束后,并不会立刻销毁所有的数据,而是等一段时间后才会销毁,

​ eg:进程号、进程执行时间、进程执行结果,会将这些结果反馈给父进程查看

所有的进程都会经理僵尸进程这个阶段

5、2.孤儿进程

​ 指子进程正常运行,父进程以外死亡,操作系统会针对孤儿进程派出福利院进行接收、接管这些进程

​ 福利院:__ init __进程/进程号为1

​ 福利院会对孤儿进程进行接管、负责收集孤儿进程的数据、结果、执行时间

五、互斥锁

1、多进程错乱问题

​ 多进程错乱问题指,当我们使用主进程创建了多个子进程后,这些子进程,在同一时间查询并需要使用主进程内IPC(消息队列)内数据,但是因为是同一时间,所以会产生消息错乱问题

​ eg:主进程消息队列内只有一个数据,但是所有子进程在同一时间调用get方法后都可以查询并取到数据

​ 解决办法:可以使用互斥锁,将并发变为穿行,牺牲了效率,保证了数据的准确性和安全

多进程错乱问题代码表现:

模拟抢票软件

from multiprocessing import Process
import time
import json
import random


# 查票
def search(name):
    with open(r'data.json', 'r', encoding='utf8') as f:
        data = json.load(f)
    print('%s在查票 当前余票为:%s' % (name, data.get('ticket_num')))


# 买票
def buy(name):
    # 再次确认票
    with open(r'data.json', 'r', encoding='utf8') as f:
        data = json.load(f)
    # 模拟网络延迟
    time.sleep(random.randint(1, 3))
    # 判断是否有票 有就买
    if data.get('ticket_num') > 0:
        data['ticket_num'] -= 1
        with open(r'data.json', 'w', encoding='utf8') as f:
            json.dump(data, f)
        print('%s买票成功' % name)
    else:
        print('%s很倒霉 没有抢到票' % name)


def run(name):
    search(name)
    buy(name)


if __name__ == '__main__':
    for i in range(10):
        p = Process(target=run, args=('用户%s'%i, ))
        p.start()

2、互斥锁代码实操

​ 互斥锁可以解决多进程错乱问题

用法:

​ 1、在主进程生成互斥锁对象

​ 2、在子进程需要加互斥锁的代码上下方加入关键词

​ 3、只有第一个抢到互斥锁的进程对象才有资格运行代码,在代码运行结束后下一个进程按照上述流程运行

注意事项:

​ 建议将锁加在实际数据操作部分,否则会降低程序运行效率

关键词 作用
Lock 模块名
acquire 抢锁
release 释放锁

代码实操:

from multiprocessing import Process, Lock
import time
import json
import random


def search(name):
    with open(r'data.json', 'r', encoding='utf8') as f:
        data = json.load(f)
    print('%s查看票 目前剩余:%s' % (name, data.get('ticket_num')))


def buy(name):
    # 先查询票数
    with open(r'data.json', 'r', encoding='utf8') as f:
        data = json.load(f)
    # 模拟网络延迟
    time.sleep(random.randint(1, 3))
    # 买票
    if data.get('ticket_num') > 0:
        with open(r'data.json', 'w', encoding='utf8') as f:
            data['ticket_num'] -= 1
            json.dump(data, f)
        print('%s 买票成功' % name)
    else:
        print('%s 买票失败 非常可怜 没车回去了!!!' % name)


def run(name, mutex):
    search(name)
    mutex.acquire()  # 抢锁
    buy(name)
    mutex.release()  # 释放锁


if __name__ == '__main__':
    mutex = Lock()  # 产生一把锁
    for i in range(10):
        p = Process(target=run, args=('用户%s号' % i, mutex))
        p.start()

六、多进程TCP服务端并发

1、代码用法

import socket
from multiprocessing import Process


def get_server():
    server = socket.socket()
    server.bind(('127.0.0.1', 8080))
    server.listen(5)
    return server


def get_talk(sock):
    while True:
        data = sock.recv(1024)
        print(data.decode('utf8'))
        sock.send(data.upper())


if __name__ == '__main__':
    server = get_server()
    while True:
        sock, addr = server.accept()
        # 开设多进程去聊天
        p = Process(target=get_talk, args=(sock,))
        p.start()

原文地址:http://www.cnblogs.com/kangssssh/p/16912131.html

1. 本站所有资源来源于用户上传和网络,如有侵权请邮件联系站长! 2. 分享目的仅供大家学习和交流,请务用于商业用途! 3. 如果你也有好源码或者教程,可以到用户中心发布,分享有积分奖励和额外收入! 4. 本站提供的源码、模板、插件等等其他资源,都不包含技术服务请大家谅解! 5. 如有链接无法下载、失效或广告,请联系管理员处理! 6. 本站资源售价只是赞助,收取费用仅维持本站的日常运营所需! 7. 如遇到加密压缩包,默认解压密码为"gltf",如遇到无法解压的请联系管理员! 8. 因为资源和程序源码均为可复制品,所以不支持任何理由的退款兑现,请斟酌后支付下载 声明:如果标题没有注明"已测试"或者"测试可用"等字样的资源源码均未经过站长测试.特别注意没有标注的源码不保证任何可用性