import time
from multiprocessing import Process

def f(name):
    print('hello', name)
    time.sleep(1)
    print('我是子進程')


if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    # p.join()
    print('我是父進程')

import os
from multiprocessing import Process

def f(x):
    print('子進程id ：',os.getpid(),'父進程id ：',os.getppid())
    return x*x

if __name__ == '__main__':
    print('主進程id ：', os.getpid())
    p_lst = []
    for i in range(5):
        p = Process(target=f, args=(i,))
        p.start()

import time
from multiprocessing import Process


def f(name):
    print('hello', name)
    time.sleep(1)


if __name__ == '__main__':
    p_lst = []
    for i in range(5):
        p = Process(target=f, args=('bob',))
        p.start()
        p_lst.append(p)
        p.join()
     # [p.join() for p in p_lst]
    print('父進程在執(zhí)行')

import os
from multiprocessing import Process


class MyProcess(Process):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        print(os.getpid())
        print('%s 正在和女主播聊天' %self.name)

p1=MyProcess('wupeiqi')
p2=MyProcess('yuanhao')
p3=MyProcess('nezha')

p1.start() # start會自動調用run
p2.start()
# p2.run()
p3.start()


p1.join()
p2.join()
p3.join()

print('主線程')

from multiprocessing import Process

def work():
    global n
    n=0
    print('子進程內(nèi): ',n)


if __name__ == '__main__':
    n = 100
    p=Process(target=work)
    p.start()
    print('主進程內(nèi): ',n)

import os
import time
from multiprocessing import Process

class Myprocess(Process):
    def __init__(self,person):
        super().__init__()
        self.person = person
    def run(self):
        print(os.getpid(),self.name)
        print('%s正在和女主播聊天' %self.person)


p=Myprocess('哪吒')
p.daemon=True # 一定要在p.start()前設置,設置p為守護進程,禁止p創(chuàng)建子進程,并且父進程代碼執(zhí)行結束,p即終止運行
p.start()
time.sleep(10)  # 在sleep時查看進程id對應的進程ps -ef|grep id
print('主')

from multiprocessing import Process

def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")


p1=Process(target=foo)
p2=Process(target=bar)

p1.daemon=True
p1.start()
p2.start()
time.sleep(0.1)
print("main-------")  # 打印該行則主進程代碼結束,則守護進程p1應該被終止.#可能會有p1任務執(zhí)行的打印信息123,因為主進程打印main----時,p1也執(zhí)行了,但是隨即被終止，p2可以打印出來。

from socket import *
from multiprocessing import Process

server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8080))
server.listen(5)

def talk(conn,client_addr):
    while True:
        try:
            msg=conn.recv(1024)
            if not msg:break
            conn.send(msg.upper())
        except Exception:
            break

if __name__ == '__main__': # windows下start進程一定要寫到這下面
    while True:
        conn,client_addr=server.accept()
        p=Process(target=talk,args=(conn,client_addr))

from multiprocessing import Process
import time
import random

class Myprocess(Process):
    def __init__(self,person):
        self.name=person
        super().__init__()

    def run(self):
        print('%s正在和網(wǎng)紅臉聊天' %self.name)
        time.sleep(random.randrange(1,5))
        print('%s還在和網(wǎng)紅臉聊天' %self.name)


p1=Myprocess('哪吒')
p1.start()

p1.terminate()#關閉進程,不會立即關閉,所以is_alive立刻查看的結果可能還是存活
print(p1.is_alive()) #結果為True

print('開始')
print(p1.is_alive()) #結果為False

class Myprocess(Process):
    def __init__(self,person):
        self.name=person   # name屬性是Process中的屬性，標示進程的名字
        super().__init__()  # 執(zhí)行父類的初始化方法會覆蓋name屬性
        # self.name = person  # 在這里設置就可以修改進程名字了
        # self.person = person  # 如果不想覆蓋進程名，就修改屬性名稱就可以了
    def run(self):
        print('%s正在和網(wǎng)紅臉聊天' %self.name)
        # print('%s正在和網(wǎng)紅臉聊天' %self.person)
        time.sleep(random.randrange(1,5))
        print('%s正在和網(wǎng)紅臉聊天' %self.name)
        # print('%s正在和網(wǎng)紅臉聊天' %self.person)


p1=Myprocess('哪吒')
p1.start()
print(p1.pid)    #可以查看子進程的進程id

import os
import time
import random
from multiprocessing import Process


def work(n):
    print('%s: %s is running' % (n, os.getpid()))
    time.sleep(random.random())
    print('%s:%s is done' % (n, os.getpid()))


if __name__ == '__main__':
    for i in range(3):
        p = Process(target=work, args=(i,))
        p.start()

# 0: 15620 is running
# 1: 19688 is running
# 2: 15892 is running
# 1:19688 is done
# 0:15620 is done
# 2:15892 is done

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

def work(lock,n):
    lock.acquire()
    print('%s: %s is running' % (n, os.getpid()))
    time.sleep(random.random())
    print('%s: %s is done' % (n, os.getpid()))
    lock.release()

if __name__ == '__main__':
    lock=Lock()
    for i in range(3):
        p=Process(target=work,args=(lock,i))
        p.start()
        
# 1: 24776 is running
# 1: 24776 is done
# 0: 23588 is running
# 0: 23588 is done
# 2: 27308 is running
# 2: 27308 is done

# 文件db的內(nèi)容為：{"count":5}
# 注意一定要用雙引號，不然json無法識別
# 并發(fā)運行，效率高，但競爭寫同一文件，數(shù)據(jù)寫入錯亂
from multiprocessing import Process, Lock
import time, json, random


def search():
    dic = json.load(open('db'))
    print('剩余票數(shù)%s' % dic['count'])


def get():
    dic = json.load(open('db'))
    time.sleep(random.random())  # 模擬讀數(shù)據(jù)的網(wǎng)絡延遲
    if dic['count'] > 0:
        dic['count'] -= 1
        time.sleep(random.random())  # 模擬寫數(shù)據(jù)的網(wǎng)絡延遲
        json.dump(dic, open('db', 'w'))
        print('購票成功')
    else:
        print('購票失敗')


def task(lock):
    search()
    lock.acquire()
    get()
    lock.release()


if __name__ == '__main__':
    lock = Lock()
    for i in range(10):  # 模擬并發(fā)10個客戶端搶票
        p = Process(target=task, args=(lock,))
        p.start()

'''
multiprocessing模塊支持進程間通信的兩種主要形式:管道和隊列
都是基于消息傳遞實現(xiàn)的,但是隊列接口
'''

from multiprocessing import Queue

q = Queue(3)

# put ,get ,put_nowait,get_nowait,full,empty
q.put(3)
q.put(2)
q.put(1)
# q.put(3)   # 如果隊列已經(jīng)滿了，程序就會停在這里，等待數(shù)據(jù)被別人取走，再將數(shù)據(jù)放入隊列。
# 如果隊列中的數(shù)據(jù)一直不被取走，程序就會永遠停在這里。
try:
    q.put_nowait(3)  # 可以使用put_nowait，如果隊列滿了不會阻塞，但是會因為隊列滿了而報錯。
except:  # 因此我們可以用一個try語句來處理這個錯誤。這樣程序不會一直阻塞下去，但是會丟掉這個消息。
    print('隊列已經(jīng)滿了')

# 因此，我們再放入數(shù)據(jù)之前，可以先看一下隊列的狀態(tài)，如果已經(jīng)滿了，就不繼續(xù)put了。
print(q.full())  # True

print(q.get())  # 3
print(q.get())  # 2
print(q.get())  # 1
# print(q.get()) # 同put方法一樣，如果隊列已經(jīng)空了，那么繼續(xù)取就會出現(xiàn)阻塞。
try:
    q.get_nowait(3)  # 可以使用get_nowait，如果隊列滿了不會阻塞，但是會因為沒取到值而報錯。
except:  # 因此我們可以用一個try語句來處理這個錯誤。這樣程序不會一直阻塞下去。
    print('隊列已經(jīng)空了')

print(q.empty())  # True,空了

import time
from multiprocessing import Process, Queue


def f(q):
    q.put([time.asctime(), 'from Eva', 'hello'])  # 調用主函數(shù)中p進程傳遞過來的進程參數(shù) put函數(shù)為向隊列中添加一條數(shù)據(jù)。


if __name__ == '__main__':
    q = Queue()  # 創(chuàng)建一個Queue對象
    p = Process(target=f, args=(q,))  # 創(chuàng)建一個進程
    p.start()
    print(q.get())
    p.join()

# ['Mon Dec  9 18:27:08 2019', 'from Eva', 'hello']

import os
import time
import multiprocessing


# 向queue中輸入數(shù)據(jù)的函數(shù)
def inputQ(queue):
    info = str(os.getpid()) + '(put):' + str(time.asctime())
    queue.put(info)


# 向queue中輸出數(shù)據(jù)的函數(shù)
def outputQ(queue):
    info= queue.get()
    print('%s%s%s' % (str(os.getpid()), '(get):', info))


# Main
if __name__ == '__main__':
    multiprocessing.freeze_support()
    record1 = []  # store input processes
    record2 = []  # store output processes
   
 queue = multiprocessing.Queue(3)

    # 輸入進程
    for i in range(10):
        process = multiprocessing.Process(target=inputQ, args=(queue,))
        process.start()
        record1.append(process)

    # 輸出進程
    for i in range(10):
        process = multiprocessing.Process(target=outputQ, args=(queue,))
        process.start()
        record2.append(process)

    for p in record1:
        p.join()

    for p in record2:
        p.join()

# 17568(get):3208(put):Mon Dec  9 18:29:17 2019
# 27620(get):24024(put):Mon Dec  9 18:29:17 2019
# 19780(get):21716(put):Mon Dec  9 18:29:17 2019
# 27576(get):27608(put):Mon Dec  9 18:29:17 2019
# 11304(get):10668(put):Mon Dec  9 18:29:18 2019
# 19732(get):20548(put):Mon Dec  9 18:29:18 2019
# 18120(get):25360(put):Mon Dec  9 18:29:18 2019
# 24752(get):21764(put):Mon Dec  9 18:29:18 2019
# 19848(get):7604(put):Mon Dec  9 18:29:18 2019
# 13888(get):10376(put):Mon Dec  9 18:29:18 2019

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


def producer(q):
    for i in range(10):
        time.sleep(random.randint(1, 3))
        res = '包子%s' % i
        q.put(res)
        print('%s 生產(chǎn)了 %s' % (os.getpid(), res))


def consumer(q):
    while True:
        res = q.get()
        time.sleep(random.randint(1, 3))
        print('%s 吃 %s' % (os.getpid(), res))


if __name__ == '__main__':
    q = Queue()
    # 生產(chǎn)者們:即廚師們
    p1 = Process(target=producer, args=(q,))

    # 消費者們:即吃貨們
    c1 = Process(target=consumer, args=(q,))

    # 開始
    p1.start()
    c1.start()
    print('主')

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


def producer(q):
    for i in range(10):
        time.sleep(random.randint(1, 3))
        res = '包子%s' % i
        q.put(res)
        print('%s 生產(chǎn)了 %s' % (os.getpid(), res))
    q.put(None)# 發(fā)送結束信號
def consumer(q):
    while True:
        res = q.get()
        if res is None: break  # 收到結束信號則結束
        time.sleep(random.randint(1, 3))
        print('%s 吃 %s' % (os.getpid(), res))


if __name__ == '__main__':
    q = Queue()
    # 生產(chǎn)者們:即廚師們
    p1 = Process(target=producer, args=(q,))

    # 消費者們:即吃貨們
    c1 = Process(target=consumer, args=(q,))

    # 開始
    p1.start()
    c1.start()
    print('主')

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


def producer(q):
    for i in range(2):
        time.sleep(random.randint(1, 3))
        res = '包子%s' % i
        q.put(res)
        print('%s 生產(chǎn)了 %s' % (os.getpid(), res))


def consumer(q):
    while True:
        res = q.get()
        if res is None: break  # 收到結束信號則結束
        time.sleep(random.randint(1, 3))
        print('%s 吃 %s' % (os.getpid(), res))


if __name__ == '__main__':
    q = Queue()
    # 生產(chǎn)者們:即廚師們
    p1 = Process(target=producer, args=(q,))

    # 消費者們:即吃貨們
    c1 = Process(target=consumer, args=(q,))

    # 開始
    p1.start()
    c1.start()

    p1.join()
    q.put(None)# 發(fā)送結束信號
    print('主')

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


def producer(name, q):
    for i in range(2):
        time.sleep(random.randint(1, 3))
        res = '%s%s' % (name, i)
        q.put(res)
        print('%s 生產(chǎn)了 %s' % (os.getpid(), res))


def consumer(q):
    while True:
        res = q.get()
        if res is None: break  # 收到結束信號則結束
        time.sleep(random.randint(1, 3))
        print('%s 吃 %s' % (os.getpid(), res))


if __name__ == '__main__':
    q = Queue()
    # 生產(chǎn)者們:即廚師們
    p1 = Process(target=producer, args=('包子', q))
    p2 = Process(target=producer, args=('骨頭', q))
    p3 = Process(target=producer, args=('泔水', q))

    # 消費者們:即吃貨們
    c1 = Process(target=consumer, args=(q,))
    c2 = Process(target=consumer, args=(q,))

    # 開始
    p1.start()
    p2.start()
    p3.start()
    c1.start()

    p1.join()  # 必須保證生產(chǎn)者全部生產(chǎn)完畢,才應該發(fā)送結束信號
    p2.join() 
    p3.join()
    q.put(None) # 有幾個消費者就應該發(fā)送幾次結束信號None 
    q.put(None) # 發(fā)送結束信號     
    print('主')

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


def producer(name, q):
    for i in range(10):
        time.sleep(random.randint(1, 3))
        res = '%s%s' % (name, i)
        q.put(res)
        print('%s 生產(chǎn)了 %s' % (os.getpid(), res))
    q.join()  # 生產(chǎn)完畢，使用此方法進行阻塞，直到隊列中所有項目均被處理。


def consumer(q):
    while True:
        res = q.get()
        time.sleep(random.randint(1, 3))
        print('%s 吃 %s' % (os.getpid(), res))
        q.task_done()  # 向q.join()發(fā)送一次信號,證明一個數(shù)據(jù)已經(jīng)被取走了


if __name__ == '__main__':
    q = JoinableQueue()
    # 生產(chǎn)者們:即廚師們
    p1 = Process(target=producer, args=('包子', q))
    p2 = Process(target=producer, args=('骨頭', q))
    p3 = Process(target=producer, args=('泔水', q))

    # 消費者們:即吃貨們
    c1 = Process(target=consumer, args=(q,))
    c2 = Process(target=consumer, args=(q,))
    c1.daemon = True
    c2.daemon = True

    # 開始
    p_l = [p1, p2, p3, c1, c2]
    for p in p_l:
        p.start()

    p1.join()
    p2.join()
    p3.join()
    print('主')

    # 主進程等--->p1,p2,p3等---->c1,c2
    # p1,p2,p3結束了,證明c1,c2肯定全都收完了p1,p2,p3發(fā)到隊列的數(shù)據(jù)
    # 因而c1,c2也沒有存在的價值了,不需要繼續(xù)阻塞在進程中影響主進程了。應該隨著主進程的結束而結束,所以設置成守護進程就可以了。

import os,time
from multiprocessing import Pool

def work(n):
    print('%s run' %os.getpid())
    time.sleep(3)
    return n**2

if __name__ == '__main__':
    p=Pool(3)
 #進程池中從無到有創(chuàng)建三個進程,以后一直是這三個進程在執(zhí)行任務
    res_l=[]
    for i in range(10):
        res=p.apply(work,args=(i,))
 # 同步調用，直到本次任務執(zhí)行完畢拿到res，等待任務work執(zhí)行的過程中可能有阻塞也可能沒有阻塞
                                    # 但不管該任務是否存在阻塞，同步調用都會在原地等著
    print(res_l)

import os
import time
import random
from multiprocessing import Pool


def work(n):
    print('%s run' % os.getpid())
    time.sleep(random.random())
    return n ** 2


if __name__ == '__main__':
    p = Pool(3)  # 進程池中從無到有創(chuàng)建三個進程,以后一直是這三個進程在執(zhí)行任務
    res_l = []
    for i in range(10):
        res= p.apply_async(work, args=(i,))
  # 異步運行，根據(jù)進程池中有的進程數(shù)，每次最多3個子進程在異步執(zhí)行
        # 返回結果之后，將結果放入列表，歸還進程，之后再執(zhí)行新的任務
        # 需要注意的是，進程池中的三個進程不會同時開啟或者同時結束
        # 而是執(zhí)行完一個就釋放一個進程，這個進程就去接收新的任務。
        res_l.append(res)

    # 異步apply_async用法：如果使用異步提交的任務，主進程需要使用join，等待進程池內(nèi)任務都處理完，然后可以用get收集結果
    # 否則，主進程結束，進程池可能還沒來得及執(zhí)行，也就跟著一起結束了
    p.close()
    p.join()
    for res in res_l:
        print(res.get()
 )  # 使用get來獲取apply_aync的結果,如果是apply,則沒有get方法,因為apply是同步執(zhí)行,立刻獲取結果,也根本無需get

#Pool內(nèi)的進程數(shù)默認是cpu核數(shù)，假設為4（查看方法os.cpu_count()）
#開啟6個客戶端，會發(fā)現(xiàn)2個客戶端處于等待狀態(tài)
#在每個進程內(nèi)查看pid，會發(fā)現(xiàn)pid使用為4個，即多個客戶端公用4個進程
from socket import *
from multiprocessing import Pool
import os

server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8080))
server.listen(5)

def talk(conn):
    print('進程pid: %s' %os.getpid())
    while True:
        try:
            msg=conn.recv(1024)
            if not msg:break
            conn.send(msg.upper())
        except Exception:
            break

if __name__ == '__main__':
    p=Pool(4)
    while True:
        conn,*_=server.accept()
        p.apply_async(talk,args=(conn,))
        # p.apply(talk,args=(conn,client_addr)) #同步的話，則同一時間只有一個客戶端能訪問

from socket import *

client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))


while True:
    msg=input('>>: ').strip()
    if not msg:continue

    client.send(msg.encode('utf-8'))
    msg=client.recv(1024)
    print(msg.decode('utf-8'))

from multiprocessing import Pool
import requests
import json
import os


def get_page(url):
    print('<進程%s> get %s' % (os.getpid(), url))
    respone = requests.get(url)
    if respone.status_code == 200:
        return {'url': url, 'text': respone.text}


def pasrse_page(res):
    print('<進程%s> parse %s' % (os.getpid(), res['url']))
    parse_res = 'url:<%s> size:[%s]\n' % (res['url'], len(res['text']))
    with open('db.txt', 'a') as f:
        f.write(parse_res)


if __name__ == '__main__':
    urls = [
        'https://www.baidu.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://help.github.com/',
        'http://www.sina.com.cn/'
    ]

    p = Pool(3)
    res_l = []
    for url in urls:
        res =p.apply_async(get_page, args=(url,), callback=pasrse_page)
        res_l.append(res)

    p.close()
    p.join()
    print([res.get() for res in res_l])  # 拿到的是get_page的結果,其實完全沒必要拿該結果,該結果已經(jīng)傳給回調函數(shù)處理了

'''
打印結果:
<進程3388> get https://www.baidu.com
<進程3389> get https://www.python.org
<進程3390> get https://www.openstack.org
<進程3388> get https://help.github.com/
<進程3387> parse https://www.baidu.com
<進程3389> get http://www.sina.com.cn/
<進程3387> parse https://www.python.org
<進程3387> parse https://help.github.com/
<進程3387> parse http://www.sina.com.cn/
<進程3387> parse https://www.openstack.org
[{'url': 'https://www.baidu.com', 'text': '<!DOCTYPE html>\r\n...',...}]
'''

import re
from urllib.request import urlopen
from multiprocessing import Pool

def get_page(url,pattern):
    response=urlopen(url).read().decode('utf-8')
    return pattern,response

def parse_page(info):
    pattern,page_content=info
    res=re.findall(pattern,page_content)
    for item in res:
        dic={
            'index':item[0].strip(),
            'title':item[1].strip(),
            'actor':item[2].strip(),
            'time':item[3].strip(),
        }
        print(dic)
if __name__ == '__main__':
    regex = r'<dd>.*?<.*?class="board-index.*?>(\d+)</i>.*?title="(.*?)".*?class="movie-item-info".*?<p class="star">(.*?)</p>.*?<p class="releasetime">(.*?)</p>'
    pattern1=re.compile(regex,re.S)

    url_dic={
        'http://maoyan.com/board/7':pattern1,
    }

    p=Pool()
    res_l=[]
    for url,pattern in url_dic.items():
        res=p.apply_async(get_page,args=(url,pattern),
         callback=parse_page) res_l.append(res) for i in res_l: i.get()

from multiprocessing import Pool
import time,random,os

def work(n):
    time.sleep(1)
    return n**2
if __name__ == '__main__':
    p=Pool()

    res_l=[]
    for i in range(10):
        res=p.apply_async(work,args=(i,))
        res_l.append(res)

    p.close()
    p.join() #等待進程池中所有進程執(zhí)行完畢

    nums=[]
    for res in res_l:
        nums.append(res.get() ) #拿到所有結果
    print(nums) #主進程拿到所有的處理結果,可以在主進程中進行統(tǒng)一進行處理