基于进程的并发

from multiprocessing import Process
import time

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

if __name__ == '__main__':
    p_list = []
    for i in range(3):
        p = Process(target=f,args=('alvin',))   #必须加，号
        p_list.append(p)
        p.start()
    for i in p_list:
        i.join()
    print('end...')

from multiprocessing import Process
import time

class MyProcess(Process):

    def run(self):
        time.sleep(1)
        print('hello',self.name,time.ctime())

if __name__ == '__main__':
    p_list = []
    for i in range(3):
        p = MyProcess()
        p_list.append(p)
        p.start()
    for i in p_list:
        i.join()
    print('end...')

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))
        p.start()

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 Process
import time
def foo():
    print(123)
    time.sleep(5)
    print("end123")

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

if __name__ == '__main__':

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

    p1.daemon=True  #守护进程必须要在start之前设定
    p1.start()
    p2.start()
    print("main-------")

q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。
q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.
 
q.get_nowait():同q.get(False)
q.put_nowait():同q.put(False)

q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。
q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。
q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样

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('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))

def consumer(q):   #消费者
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))

if __name__ == '__main__':
    q=Queue()

    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('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))
    q.put(None) #发送结束信号

def consumer(q):    #消费者
    while True:
        res=q.get()
        if res is None:break #收到结束信号则结束
        time.sleep(random.randint(1,3))
        print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))

if __name__ == '__main__':
    q=Queue()
    p1=Process(target=producer,args=(q,))

    c1=Process(target=consumer,args=(q,))

    p1.start()
    c1.start()
    print('主')

from multiprocessing import Process,Pipe
import time,os

def producer(seq,p):    #生产者
    left,right=p
    right.close()
    for i in seq:
        left.send(i)
    else:
        left.close()

def consumer(p,name):   #消费者
    left,right=p
    left.close()
    while True:
        try:
            baozi=right.recv()
            print('%s 收到包子:%s' %(name,baozi))
        except EOFError:
            right.close()
            break

if __name__ == '__main__':
    left,right=Pipe()

    c1=Process(target=consumer,args=((left,right),'c1'))
    c1.start()

    seq=(i for i in range(10))
    producer(seq,(left,right))

    right.close()
    left.close()

    c1.join()
    print('主进程')

p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()
p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。
   
p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
P.jion():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用

from multiprocessing import Process,Pool
import time,os

def Foo(i):
    time.sleep(1)
    print(i)

if __name__ == '__main__':

    pool = Pool(5)
    for i in range(100):
        pool.apply(func=Foo,args=(i,))   #同步调用

    pool.close()
    pool.join()
    print('end')

from multiprocessing import Process,Pool
import time,os

def Foo(i):
    time.sleep(1)
    print(i)

if __name__ == '__main__':

    pool = Pool(5)
    for i in range(100):
        pool.apply_async(func=Foo,args=(i,))

    pool.close()
    pool.join()
    print('end')

from multiprocessing import Process,Pool
import time,os

def Foo(i):
    time.sleep(1)
    print(i)
    return i+100

def Bar(arg):
    print('logger:',arg)

if __name__ == '__main__':

    pool = Pool(5)

    for i in range(100):
        pool.apply_async(func=Foo,args=(i,),callback=Bar)

    pool.close()  #必须放在join前
    pool.join()   
    print('end')