python 线程与进程

python多进程与多线程

##############################
#多进程
##############################
#import os
'''
print('1Process start, pid:',os.getpid())
print('Process start, ppid:',os.getppid())
'''
'''
#unix环境支持
pid = os.fork()
if pid == 0:
    print('child process,id:', os.getpid(), 'parent id is:', os.getppid())
else:
    print('pid:', os.getpid(), 'create child process:', pid)
'''

#################
#multiprocessing
#################
'''
from multiprocessing import Process
print('2Process start, pid:',os.getpid())
def run_proc(name):
#    while(1):
#        pass
    print('run child process %s (%s)...' %(name, os.getpid()))

print('4Process start, pid:',os.getpid())
if __name__ == '__main__':
    print('1Parent process %s.' % os.getpid())
    p = Process(target=run_proc, args=('test',))
    print('Child process will start.', os.getpid())
    #启动子进程
    p.start()
    print('3Process start, pid:',os.getpid())
    #等待子进程结束， 类似于wait()
    p.join()
    print('Child process end.', os.getpid())
'''

#由上面这个例子可以看出，python的代码是顺序执行的。
#执行p.start()方法后，开始执行子进程
#子进程的执行范围是2、4之间的代码行
#p.join()等待子进程返回
#单纯从打印的顺序，无法准确的判断父子进程的执行顺序


'''
from multiprocessing import Pool
import os, time, random
def long_time_task(name):
    print('Run task %s (%s)...' % (name, os.getpid()))
    start = time.time()
    time.sleep(random.random()*3)
    end = time.time()
    print('Task %s runs %0.2f seconds.' % (name, (end - start)))

if __name__=='__main__':
    print('Parent process %s.' % os.getpid())
    p = Pool(5)#设置最大线程格个数,默认值是cpu核心数
    for i in range(5):
        p.apply_async(long_time_task, args=(i,))#创建线程
    print('Waiting for all subprocesses done...')
    p.close()#防止将任何其他任务提交到池中。完成所有任务后，工作进程将退出。
    p.join()#调用前，必须调用close() or terminate()
    print('All subprocess done.')
'''
#win7下cmd输出内容如下
'''
Parent process 25380.
Waiting for all subprocesses done...
Run task 0 (26008)...
Run task 1 (26216)...
Run task 2 (28108)...
Run task 3 (24732)...
Run task 4 (25008)...
Task 3 runs 0.15 seconds.
Task 4 runs 1.79 seconds.
Task 2 runs 2.45 seconds.
Task 0 runs 2.69 seconds.
Task 1 runs 2.98 seconds.
All subprocess done.
请按任意键继续. . .
'''

###########
#子进程
###########
'''
import subprocess #类似于exec()?

print ('$ nslookup www.python.org')
r = subprocess.call(['nslookup', 'www.python.org'])
print('Exit code:', r)
'''
''' win7输出
$ nslookup www.python.org
服务器:  clou-ad.szclou.com
Address:  10.98.94.5

非权威应答:
名称:    dualstack.python.map.fastly.net
Addresses:  2a04:4e42:36::223
          151.101.228.223
Aliases:  www.python.org

Exit code: 0
请按任意键继续. . .
'''
'''
import subprocess
print('$ nslookup')
p = subprocess.Popen(['nslookup'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, err = p.communicate(b'set q=mx
python.org
exit
')
s='utf-8'#unix
s1='gbk'#win7系统
print(output.decode(s))
print('Exit code:', p.returncode)
'''
#相当于命令行下执行nslookup,然后输入
#set q=mx
#python.org
#exit

'''win7输出
$ nslookup
默认服务器:  clou-ad.szclou.com
Address:  10.98.94.5

> > 服务器:  clou-ad.szclou.com
Address:  10.98.94.5

python.org      MX preference = 50, mail exchanger = mail.python.org

mail.python.org internet address = 188.166.95.178
mail.python.org AAAA IPv6 address = 2a03:b0c0:2:d0::71:1
>
Exit code: 0
请按任意键继续. . .
'''

###########
#进程间通信
###########
'''
from multiprocessing import Process, Queue
import os, time, random

def write(q):
    print('Process to write: %s' % os.getpid())
    for value in ['A', 'B', 'C']:
        print('put %s to queue.' % value)
        q.put(value)
        time.sleep(random.random())

def read(q):
    print('Process to read: %s' % os.getpid())
    while True:
        value = q.get(True)
        print('Get %s from queue.' % 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()
'''
'''
Process to write: 28536
Process to read: 27968
put A to queue.
Get A from queue.
put B to queue.
Get B from queue.
put C to queue.
Get C from queue.
请按任意键继续. . .
'''

##############################
#多线程：python的多线程是真正的Posix Thread,而不是模拟出来的(linux)
##############################

#Python的标准库提供了两个模块：_thread和threading，_thread是低级模块，
#threading是高级模块，对_thread进行了封装。绝大多数情况下，
#我们只需要使用threading这个高级模块。
'''
import time, threading

print('thread %s is running...' % threading.current_thread().name)

def loop():
    print('thread %s is running...' % threading.current_thread().name)
    n = 0
    while n < 5:
        n = n + 1
        print('thread %s >>> %s' % (threading.current_thread().name, n))
        time.sleep(1)
    print('thread %s ended.' % threading.current_thread().name)
    

print('print something')
t=threading.Thread(target=loop, name='LoopThread')
t.start()
t.join()
print('thread %s ended.' % threading.current_thread().name)
'''
'''
thread MainThread is running...
thread LoopThread is running...
thread LoopThread >>> 1
thread LoopThread >>> 2
thread LoopThread >>> 3
thread LoopThread >>> 4
thread LoopThread >>> 5
thread LoopThread ended.
thread MainThread ended.
请按任意键继续. . .
'''
'''
任何进程默认启动一个进程，名字是MainThread。
threading.current_thread()返回当前线程的实例
线程创建时，如果不指定名称，默认是Thread-1，.......
'''
#定义锁
'''
lock = threading.Lock()
lock.acquire()
lock.release()
'''

import time, threading, multiprocessing
'''
def loop():
    x = 0
    while True:
        x = x ^ 1

print('cpu:%s' % multiprocessing.cpu_count())
for i in range(multiprocessing.cpu_count()):
    t = threading.Thread(target=loop)
    t.start()
'''


###########
#ThreadLocal：管理线程数据的全局变量
###########
'''
import threading
#全局变量local_school,一个ThreadLocal对象
#但每个属性，如local_school.student都是线程的局部变量，不会相互干扰

#ThreadLocal最常用的地方就是为每个线程绑定一个数据库连接，HTTP请求，
#用户身份信息等，这样一个线程的所有调用到的处理函数都可以非常方便地访问这些资源。
local_school = threading.local()

def process_student():
    std = local_school.student
    print('Hello, %s (in %s)' % (std, threading.current_thread().name))

def process_thread(name):
    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()
'''

'''
计算密集型：cpu利用率高，C语言的运行效率最高，脚本语言效率低
IO密集型：脚本语言的开发效率高，c语言最差
协程：单线程的一步编程模型。有了协程的支持，就可以基于事件驱动编写高效的多任务程序
'''


###########
#分布式进程
###########

分布式进程示例：

#task_master.py
import random, time, queue
from multiprocessing.managers import BaseManager
#发送任务队列
task_queue = queue.Queue()
#接收结果队列
result_queue = queue.Queue()

#从BaseManager继承的QueueManager
class QueueManager(BaseManager):
    pass

# 把两个Queue都注册到网络上, callable参数关联了Queue对象:
QueueManager.register('get_task_queue', callable=lambda:task_queue)
QueueManager.register('get_result_queue', callable=lambda:result_queue)

#创建实例对象
#绑定端口5000， 设置验证码 'abc'
manager = QueueManager(address=('', 5000), authkey=b'abc')
manager.start()
#获得通过网络访问的Queue对象
task = manager.get_task_queue()
result = manager.get_result_queue()

time.sleep(10)
#队列中添加任务
for i in range(10):
    n = random.randint(0, 10000)
    print('Put task %d...' % n)
    task.put(n)
#读取结果
print('Try get results...')
for i in range(10):
    r = result.get(timeout=10)
    print('Result:%s' % r)

manager.shutdown()
print('master exit.')

#task_worker.py
import queue, time, sys
from multiprocessing.managers import BaseManager

class QueueManager(BaseManager):
    pass

QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')

server_addr = '127.0.0.1'
print('Connect to server %s...' % server_addr)
m = QueueManager(address=(server_addr, 5000), authkey=b'abc')
m.connect()
task = m.get_task_queue()
result = m.get_result_queue()

'''
for i in range(10):
    try:
        n = task.get(timeout=1)
        print('run task %d * %d...' % (n, n))
        r = '%d * %d = %d' % (n, n, n*n)
        time.sleep(1)
        result.put(r)
    except queue.Empty:
        print('task queue is empty.')
'''
t1 = time.time()
while 1:
#    if not isinstance(task, queue.Queue()):
#        raise ValueError('invalid Queue:%s' % task)
    try:
        if task.empty():
            time.sleep(1)
            print('queue is empty')
            continue
        n = task.get(timeout=1)
        print('run task %d * %d...' % (n, n))
        r = '%d * %d = %d' % (n, n, n*n)
        result.put(r)
        time.sleep(1)
        if time.time() - t1 > 60:
            print('over 60s')
            break
    except BaseException as f:
        print('Error:', f)
        break;

print('worker exit.')

可以多次执行work文件，创建多个进程，各个work进程中处理的数据之和是master进程中的数据，且各work处理的数据不会重复。

目前只能在Ubuntu12.04上执行，win7下可能由于权限问题，无法正常执行。