linux建立程序

1. Linux建立子程序的原理:

#1).  父程序和子程序, 如果父程序結束， 子程序也隨之結束;
#2).  x先有父程序， 再有子程序， 通過fork函式實現;

2. fork函式的返回值:呼叫該方法一次， 返回兩次;

#- 產生的子程序返回一個0
#- 父程序返回子程序的pid;

3. Window也能使用fork函式麼?

#Windows沒有fork函式， Mac有fork函式(Unix -> Linux, Unix-> Mac),
#封裝了一個模組multiprocessing

4. 常用方法:

#- os.fork()
#- os.getpid(): 獲取當前程序的pid;
#- os.getppid(): parent process id, 獲取當前程序的父程序的id號；
 

import multiprocessing
import os
import time
print('當前程序(pid=%d)正在執行...' %(os.getpid()))
print('當前程序的父程序(pid=%d)正在執行...' %(os.getppid()))
print('正在建立子程序...')
pid=os.fork()
pid2=os.fork()
print('第一個:',pid)
print('第二個:',pid2)
if pid==0:
    print('這是建立的子程序,子程序的id為%s,父程序id為%s' %(os.getpid(),os.getppid()))
else:
    print('當前父程序[%s]的返回值是%s' %(os.getpid(),pid))
time.sleep(10)
 

建立子程序_通過例項化process物件建立多程序

import multiprocessing
import threading
import time

def job():
    print('當前子程序的名稱為%s' %(multiprocessing.current_process()))
#建立一個程序物件
p1=multiprocessing.Process(target=job)
#執行多程序，執行任務
p1.start()
#建立一個程序物件
p2=multiprocessing.Process(target=job)
p2.start()
#等待所有的子程序執行結束，在執行主程序內容
p1.join()
p2.join()
print('任務執行結束...')
 

派生多程序，通過繼承的方式建立子程序

import multiprocessing
import time

class JobProcess(multiprocessing.Process):
    #重寫Process的構造方法，獲取新的屬性
    def __init__(self,queue):
        super(JobProcess, self).__init__()
        self.queue=queue
    #重寫run方法，將執行的任務放在裡面即可
    def run(self):
        print('當前子程序名稱為%s' %(multiprocessing.current_process()))
processes=[]
#啟動10個程序，來執行需要執行的任務
for i in range(10):
    #例項化物件
    p=JobProcess(queue=3)
    processes.append(p)
    #啟動多程序，處理需要執行的任務
    p.start()
#等待所有的子程序執行結束，在繼續執行主程序
[process.join() for process in processes]
time.sleep(3)
#執行主程序
print('任務執行結束...')

mytimeit

import time
def timeit(f):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        res = f(*args, **kwargs)
        end_time = time.time()
        print("%s執行時間為%s" %(f.__name__, end_time-start_time))
        return  res
    return  wrapper

多程序案例

import multiprocessing
import threading
from mytimeit import timeit
class JobProcess(multiprocessing.Process):
    def __init__(self,li):
        super(JobProcess, self).__init__()
        self.li=li
    def run(self):
        for item in self.li:
            sum(item)

class JobThread(threading.Thread):
    def __init__(self,li):
        super(JobThread, self).__init__()
        self.li=li
    def run(self):
        sum(self.li)

@timeit
def use_multi_process():
    # 1). 開啟的程序數有瓶頸， 取決於CPU的個數;
    # 2). 如果處理的資料比較小， 不建議使用多程序， 因為程序的建立和銷燬需要時間，
    #     開啟的程序數越多，不一定效率越高;;
    # 3). 如果處理的資料量足夠大， 0<開啟的程序數<cpu個數, 開啟的程序數越多， 效率越高;
    lists = [[1, 2, 3, 4, 5], [2, 3, 4, 5, 6],
             [2, 3, 4, 5, 6, 7, 8], [2, 3, 4, 5, 6]] * 10
    processes=[]
    for li in lists:
        p=JobProcess(li)
        processes.append(p)
        p.start()
    # # 對於處理的資料進行分片， 每個程序處理一部分任務;
    # for i in range(0, len(lists), 10):
    #     # print(i, i+1000)
    #     p = JobProcess(lists[i:i + 10])
    #     processes.append(p)
    #     p.start()

    #等待所有的子程序執行結束
    [process.join() for process in processes]
    print('多程序執行結束')
@timeit
def use_multi_thread():
    lists = [[1, 2, 3, 4, 5], [2, 3, 4, 5, 6],
             [2, 3, 4, 5, 6, 7, 8], [2, 3, 4, 5, 6]] * 10
    processes = []
    for li in lists:
        p = JobThread(li)
        processes.append(p)
        p.start()
    # 等待所有的子程序執行結束
    [process.join() for process in processes]
    print("多執行緒執行結束.......")
if __name__=='__main__':
    use_multi_process()
    use_multi_thread()

守護程序

# 守護執行緒:
    #setDeamon:
        #True: 主執行緒執行結束， 子執行緒不再繼續執行;
        #Flase:

import multiprocessing
import time
def job():
    name=multiprocessing.current_process()
    print('%s開始執行' %(name))
    time.sleep(3)
    print('%s結束程序' %(name))
if __name__=='__main__':
    #啟動一個子程序
    p1=multiprocessing.Process(target=job,name='use_daemon')
    p1.daemon=True
    p1.start()
    #join等待所有子程序執行結束，在執行主程序

    p1.join(3)
    #執行主程序
    print('%s程式執行結束'%(multiprocessing.current_process()) )

終止程序

import time
import  multiprocessing

def job():
    print("start.....")
    time.sleep(1)
    print('end.......')
if __name__ == '__main__':
    p = multiprocessing.Process(target=job)
    print("Before:", p.is_alive())
    p.start()  # 啟動子程序
    print("During:", p.is_alive())
    p.terminate()   # 終止子程序
    print('terminate:', p.is_alive())
    p.join()        #等待子程序徹底終止
    print("joined:", p.is_alive())

程序池之ProcessPoolExecutor

from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import ProcessPoolExecutor
def job(id):
    print('start %d' %(id))
    print('end %d' % (id))
    return id
# 第2種方式: submit
pool = ProcessPoolExecutor(max_workers=4)
# 分配任務給子程序， 並且返回一個Future物件;
f1 = pool.submit(job, 5)
# 獲取程序是否執行結束；
f1.done()
# 獲取子程序執行的結果
f1.result()

# 第三種方式:map函式
# pool = ProcessPoolExecutor(max_workers=4)
# for res in pool.map(job, range(1,100)):
#     print(res)

程序間的通訊之生產者和消費者

import multiprocessing
from multiprocessing import Queue
import time
class Producer(multiprocessing.Process):
    #往佇列寫內容
    def __init__(self,queue):
        super(Producer,self).__init__()
        self.queue=queue
    def run(self):
        for i in range(10):
            self.queue.put(i)
            time.sleep(2)
            print('傳遞訊息,內容為:%s' %(i))

class Consumer(multiprocessing.Process):
    def __init__(self,queue):
        super(Consumer, self).__init__()
        self.queue=queue
    def run(self):
        # 判斷佇列是否為空， 如果是， 跳出迴圈， 不會再去從佇列獲取資料;
        # while not self.queue.empty():
        while True:
            time.sleep(0.1)
            print("讀取程序傳遞的訊息:%s" % (self.queue.get()))
if __name__=='__main__':
    q=Queue()
    p1=Producer(q)
    c1=Consumer(q)
    p1.start()
    c1.start()
    p1.join()
    c1.terminate()
    c1.join()
    print('all done')