Python（12）进程与线程（二）

#-*- coding: utf-8 -*-
from multiprocessing import Process,Queue
import os,time,random
def write(q):
    print('写进程为：%s' % os.getpid())
    for value in  ['A','B','C']:
        print('将 %s 放入队列' % value)
        q.put(value)
        time.sleep(random.random())
def read(q):
    print('读进程为：%s' % os.getpid())
    while True:
        value = q.get(True)
        print('在队列中获取 %s' % value)
if __name__ == '__main__':
    q = Queue()
    pw = Process(target=write,args=(q,))
    pr = Process(target=read,args=(q,))
    pw.start()
    pr.start()
    pw.join()
    pr.terminate()  #杀死子进程
#输出结果：
写进程为：3576
将 A 放入队列
读进程为：880
在队列中获取 A
将 B 放入队列
在队列中获取 B
将 C 放入队列
在队列中获取 C
#解析
q是Queue的实例，Queue的方法put()可以将元素放入item队列，get()可以查看队列中的元素
get(True)可以控制是否使用报错，如果使用False的话，在队列没有值的话会报错，反之则不会，默认为True

#-*- coding: utf-8 -*-
import time, threading
# 新线程执行的代码:
def loop():
    print('线程 %s 正在运行' % threading.current_thread().name)
    n = 0
    while n < 5:
        n = n + 1
        print('线程 %s >>> %s' % (threading.current_thread().name, n))
        time.sleep(1)
    print('线程 %s 结束' % threading.current_thread().name)
print('线程 %s 正在运行' % threading.current_thread().name)
t = threading.Thread(target=loop, name='LoopThread')
t.start()
t.join()
print('线程 %s 已经结束' % threading.current_thread().name)
#输出：
线程 MainThread 正在运行
线程 LoopThread 正在运行
线程 LoopThread >>> 1
线程 LoopThread >>> 2
线程 LoopThread >>> 3
线程 LoopThread >>> 4
线程 LoopThread >>> 5
线程 LoopThread 结束
线程 MainThread 已经结束
#解析
第一行输出的名称'MainThread'，其实就是当前进程创建的主线程名称，然后创建了线程，并且命名为'LoopThread'，当线程结束后，最后主线程也结束

#-*- coding: utf-8 -*-
import time, threading
# 假定这是你的银行存款:
balance = 0
def change_it(n):
    # 先存后取，结果应该为0:
    global balance
    balance = balance + n
    balance = balance - n
def run_thread(n):
    for i in range(20000000):
        change_it(n)
t1 = threading.Thread(target=run_thread, args=(5,))
t2 = threading.Thread(target=run_thread, args=(8,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance)
#执行结果：
电脑速度过快的话，这里的输出可能还是0，可以把循环增加几百倍，我这里使用虚拟机用centos执行的，结果都是大于0的
#解析：
这里定义了共享变量'balance'，初始值为0，并且启动两个线程去执行，从逻辑来看，现存后取，最终结果肯定也是0，但是由于线程的调度是由操作系统决定的，所以当两个线程交替执行并且循环次数足够多使，最终的结果就不一定为0了

#!/usr/bin/env python3
#-*- coding: utf-8 -*-
import time, threading
# 假定这是你的银行存款:
balance = 0
lock = threading.Lock()   #创建实例
def change_it(n):
    # 先存后取，结果应该为0:
    global balance
    balance = balance + n
    balance = balance - n
def run_thread(n):              #修改run_thread方法
    for i in range(2000000):
        lock.acquire()          #先获取锁
        try:
            change_it(n)
        finally:
            lock.release()      #执行完后要释放锁，否则会造成死锁
t1 = threading.Thread(target=run_thread, args=(5,))
t2 = threading.Thread(target=run_thread, args=(8,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance)
#执行结果：
0
#解析:
每次在执行change_it()函数时，需要先获取锁，然后等待执行完毕之后通过finally释放锁，这样在执行函数时就会只有一个线程执行，从而避免变量被多个线程修改

def process_student(name):
    std = Student(name)   #编写了一个调用其他两个函数的函数，并且创建std实例，然后把实例传入其他函数，这里的std就相当于局部变量
    do_task_1(std)
    do_task_2(std)
def do_task_1(std):
    do_subtask_1(std)
    do_subtask_2(std)
def do_task_2(std):
    do_subtask_2(std)
    do_subtask_2(std)

global_dict = {}  #全局字典
def std_thread(name):
    std = Student(name)
    global_dict[threading.current_thread()] = std   #写入键值，键值为  '当前线程名称':'std实例'
    do_task_1()
    do_task_2()
def do_task_1():
    std = global_dict[threading.current_thread()]  #根据当前线程名称取出std实例
    ...
def do_task_2():
    std = global_dict[threading.current_thread()]  #根据当前线程名称取出std实例
    ...

import threading
# 创建全局ThreadLocal对象:
local_school = threading.local()
def process_student():
    # 获取当前线程关联的student:
    std = local_school.student
    print('Hello, %s (in %s)' % (std, threading.current_thread().name))    #std是process_thread函数传入的name变量，第二个是线程名称
def process_thread(name):
    # 绑定ThreadLocal的student:
    local_school.student = name   
    process_student()
t1 = threading.Thread(target= process_thread, args=('Alice',), name='Thread-A')
t2 = threading.Thread(target= process_thread, args=('Bob',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()
#执行结果：
Hello, Alice (in Thread-A)
Hello, Bob (in Thread-B)

使用multiprocessing和queue模块
- 使用multiprocessing.managers中的BaseManager创建分布式管理器
- 使用Queue创建队列，用于多个进程之间的通信

# -*- coding:utf-8 -*-
import random, queue
from multiprocessing.managers import BaseManager
from multiprocessing import freeze_support
#1.定义两个Queue队列，tesk_queue用于发送任务，result_queue用于接收结果
task_queue = queue.Queue()
result_queue = queue.Queue()
#创建QueueManager类,继承BaseManager,用于后面创建管理器
class QueueManager(BaseManager):
    pass
#定义两个函数，返回结果是Queue队列
def return_task_queue():
    global task_queue       #定义全局变量
    return task_queue       #返回发送任务的队列
def return_result_queue():
    global result_queue
    return result_queue     #返回接收结果的队列
#2.利用register方法把上面创建的两个队列（也就是两个函数）注册在网络上
#callable参数关联了Queue对象，将Queue对象在网络中暴露
#QueueManager.register的第一个参数是注册在网络上队列的名称
def test():
    QueueManager.register('get_task_queue', callable=return_task_queue)
    QueueManager.register('get_result_queue', callable=return_result_queue)
    #3.绑定端口8001，设置验证口令,这个相当于对象的初始化
    #绑定端口并填写验证口令，windows下需要填写IP地址，Linux下默认为本地，地址为空
    manager = QueueManager(address=('127.0.0.1', 8001), authkey=b'abc') #口令必须写成类似b'abc'形式，只写'abc'运行错误，其中口令为'abc'
    #4.启动管理器，启动Queue队列，监听信息通道
    manager.start()
    #5.通过管理实例的方法获访问网络中的Queue对象,即通过网络访问获取任务队列和结果队列,创建了两个Queue实例，
    task = manager.get_task_queue()  #发送
    result = manager.get_result_queue()  #接收
    #6.添加任务，获取返回的结果,将任务放到Queue队列中
    for i in range(10):
        n = random.randint(0, 10) #返回0到10之间的随机数
        print("Put task %s ..." % n)
        task.put(n) # 将n放入到任务队列中
    #从结果队列中取出结果
    print("Try get results...")
    for i in range(11): # 注意，这里结果队列中取结果设置为11次，总共只有10个任务和10个结果，多1次主要是确认队列中是不是已经空了
        #总共循环10次，上面放入了10个数字作为任务
        #加载一个异常捕获
        try:
            r = result.get(timeout=5) # 每次等待5秒，取结果队列中的值
            print("Result: %s" % r)
        except queue.Empty:   #如果上面代码出现异常则输出这个
            print("result queue is empty.")
    #7.关闭服务
    #一定要关闭，否则会报管道未关闭的错误
    manager.shutdown()
    print("master exit.")
if __name__ == '__main__':
    freeze_support()       #Windows下多进程可能出现问题，添加这个可以缓解
    print("Start!")
    # 运行服务进程
    test()

# coding: utf-8
# 定义具体的任务进程，具体的工作任务是什么
import time, sys, queue
from multiprocessing.managers import BaseManager
#创建QueueManager类,继承BaseManager,用于后面创建管理器
class QueueManager(BaseManager):
    pass
#1.使用QueueManager注册用于获取Queue的方法名称
#前面服务进程已经将队列名称暴露到网络中，该任务进程注册时只需要提供名称即可，与服务进程中队列名称一致
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')
#2.连接到服务器,也就是运行服务进程代码的机器
server_addr = '127.0.0.1'
print("Connet to server %s..." % server_addr)
#创建一个管理器实例，端口和验证口令保持与服务进程中完全一致
m = QueueManager(address=(server_addr, 8001), authkey=b'abc')
#连接到网络服务器
m.connect()
#3.从网络上获取Queue对象，并进行本地化，与服务进程是同一个队列
task = m.get_task_queue()
result = m.get_result_queue()
#4.从task队列获取任务，并把结果写入到resul队列，这里也循环10次，因为上面的服务队列也是计算10次
for i in range(10):
    try:
        #前面服务进程向task队列中放入了n,这里取出n，计算n和n相乘，并将相乘的算式和结果放入到result队列中去
        n = task.get(timeout=1)  #每次等待1秒后取出任务
        print("run task %d * %d..." % (n, n))  
        r = '%d * %d = %d' % (n, n, n*n)  #进行计算
        time.sleep(1)
        result.put(r)  #把r的值返回到result队列中
    except queue.Empty:
        print("task queue is empty.")
# 任务处理结束
print("worker exit.")