学习python，从入门到放弃（35）

学习python，从入门到放弃（35）

消息队列

from multiprocessing import Queue


q = Queue(5)  # 自定义队列的长度
# 朝队列中存放数据
q.put(111)
q.put(222)
q.put(333)
print(q.full())  # False  判断队列是否满了
q.put(444)
q.put(555)
print(q.full())  # True
# q.put(666)  # 超出最大长度 原地阻塞等待队列中出现空位
print(q.get())
print(q.get())
print(q.empty())  # False  判断队列是否空了
print(q.get())
print(q.get())
print(q.get())
print(q.empty())  # True
# print(q.get())  # 队列中没有值 继续获取则阻塞等待队列中给值
# print(q.get_nowait())  # 队列中如果没有值 直接报错

full()、empty()、get_nowait() 等方法不可以在并发的场景下使用。

队列可以支持进程间数据通信。

IPC机制(进程间通信)

一个操作系统不同的进程，有自己的进程内存空间，其中的数据不共享，因此进程间的通信就需要采用一定的机制。

from multiprocessing import Process, Queue


def producer(q):
    # print('子进程producer从队列中取值>>>:', q.get())
    q.put('子进程producer往队列中添加值')


def consumer(q):
    print('子进程consumer从队列中取值>>>:', q.get())


if __name__ == '__main__':
    q = Queue()
    p = Process(target=producer, args=(q,))
    p1 = Process(target=consumer, args=(q,))
    p.start()
    p1.start()
    # q.put(123)  # 主进程往队列中存放数据123
    print('主进程')

生产者消费者模型

生产者负责生产/制作数据。

消费者负责消费/处理数据。

比如在爬虫领域中，会先通过代码爬取网页数据(爬取网页的代码就可以称之为是生产者)，之后针对网页数据做筛选处理(处理网页的代码就可以称之为消费者)。

如果使用进程来演示，除了有至少两个进程之外，还需要一个媒介(消息队列)。

以后遇到该模型需要考虑的问题其实就是供需平衡的问题，生产力与消费力要均衡。

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


def producer(name, food, q):
    for i in range(5):
        data = f'{name}生产了{food}{i}'
        print(data)
        time.sleep(random.randint(1, 3))  # 模拟产生过程
        q.put(data)


def consumer(name, q):
    while True:
        food = q.get()
        if food == None:
            print('完蛋了 没得吃了 要饿死人了')
            break
        time.sleep(random.random())
        print(f'{name}吃了{food}')
        q.task_done()  # 每次去完数据必须给队列一个反馈


if __name__ == '__main__':
    # q = Queue()
    q = JoinableQueue()
    p1 = Process(target=producer, args=('大厨jason', '韭菜炒蛋', q))
    p2 = Process(target=producer, args=('老板kevin', '秘制小汉堡', q))
    c1 = Process(target=consumer, args=('涛涛', q))
    c2 = Process(target=consumer, args=('龙龙', q))
    c1.daemon = True
    c2.daemon = True
    p1.start()
    p2.start()
    c1.start()
    c2.start()
    # 生产者生产完所有数据之后 往队列中添加结束的信号
    p1.join()
    p2.join()
    # q.put(None)  # 结束信号的个数要跟消费者个数一致才可以
    # q.put(None)
    """队列中其实已经自己加了锁 所以多进程取值也不会冲突 并且取走了就没了"""
    q.join()  # 等待队列中数据全部被取出(一定要让生产者全部结束才能判断正确)

线程理论

进程是资源单位，而线程是执行单位。进程仅仅是在内存中开辟一块空间(提供线程工作所需的资源)，线程真正被CPU执行，线程需要的资源跟所在进程的要，一个进程中至少有一个线程。

开设线程的消耗远远小于进程

开进程需要申请内存空间再拷贝代码。

开线程无需申请内存空间、拷贝代码，而且一个进程内可以开设多个线程，一个进程内的多个线程数据是共享的。

多种功能应该开设多线程而不是多进程

开设线程的两种方式

from threading import Thread
import time


def task(name):
    print(f'{name} is running')
    time.sleep(3)
    print(f'{name} is over')


# 创建线程无需在__main__下面编写 但是为了统一 还是习惯在子代码中写
t = Thread(target=task, args=('jason',))
t.start()  # 创建线程的开销极小 几乎是一瞬间就可以创建
print('主线程')


class MyThread(Thread):
    def __init__(self, username):
        super().__init__()
        self.username = username

    def run(self):
        print(f'{self.username} jason is running')
        time.sleep(3)
        print(f'{self.username} is over')


t = MyThread('jasonNB')
t.start()
print('主线程')

线程实现TCP服务端的并发

import socket
from threading import Thread

server = socket.socket()
server.bind(('127.0.0.1', 8080))
server.listen()


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


while True:
    sock, addr = server.accept()
    # 每类一个客户端就创建一个线程做数据交互
    t = Thread(target=talk, args=(sock,))
    t.start()

线程join方法

因为主线程结束也就标志着整个进程的结束，要确保子线程运行过程中所需的各项资源，所以主线程要等着子线程结束才会结束整个进程。

from threading import Thread
import time


def task(name):
    print(f'{name} is running')
    time.sleep(3)
    print(f'{name} is over')


t = Thread(target=task, args=('jason', ))
t.start()
t.join()  # 主线程代码等待子线程代码运行完毕之后再往下执行
print('主线程')

同一个进程内的多个线程数据共享

from threading import Thread

money = 6666666


def task():
    global money
    money = 1


t = Thread(target=task)
t.start()
t.join()
print(money)

线程对象属性和方法

一个进程下的多个线程处于一个进程

active_count() 可以统计进程下活跃的线程数。

current_thread().name 和 self.name 可以获取线程的名字。

守护线程

from threading import Thread
import time


def task(name):
    print(f'{name} is running')
    time.sleep(3)
    print(f'{name} is over')


t1 = Thread(target=task, args=('jason',))
t2 = Thread(target=task, args=('kevin',))
t1.daemon = True
t1.start()
t2.start()
print('主线程')

GIL全局解释器锁

GIL并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念，而CPython是大部分环境下默认的Python执行环境。GIL 全称 gloabl interpreter lock (全局解释器锁) ，官方解释：

In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple native threads from executing Python bytecodes at once. This lock is necessary mainly because CPython’s memory management is not thread-safe. (However, since the GIL exists, other features have grown to depend on the guarantees that it enforces.)

这主要是针对cpython解释器来说的，其他解释器不一样。

GIL遵循的原则：“一个线程运行 Python ，而其他 N 个睡眠或者等待 I/O.”（即保证同一时刻只有一个线程对共享资源进行存取）。

由于GIL涉及到底层实现，比较复杂，想要完全搞明白还是很困难的。但是只要记住2点：

  1. 在IO密集型型操作下，多线程还是可以的。比如在网络通信，time.sleep() 延时的时候。

  2. 在CPU密集型操作下，多线程性能反而不如单线程，此时只能用多进程。

相关

标签