python （高级消息队列）普通、进程、进程池的消息队列

一、普通消息队列

from queue import Queue 
这个是普通的队列模式，类似于普通列表，先进先出模式，get方法会阻塞请求，直到有数据get出来为止。

import Queue

q = Queue.Queue()

for i in range(5):
    q.put(i)

while not q.empty():
    print q.get()

输出：
0
1
2
3
4

二、多进程消息队列

from multiprocessing.Queue import Queue（各子进程共有）
这个是多进程并发的Queue队列，用于解决多进程间的通信问题。普通Queue实现不了。例如来跑多进程对一批IP列表进行运算，运算后的结果都存到Queue队列里面，这个就必须使用multiprocessing提供的Queue来实现。

创建

import multiprocessing
queue = multiprocessing.Queue(队列长度)

因为进程间不共享全局变量，所以使用Queue进行数据通信，可以在父进程中创建两个字进程，一个往Queue里写数据，一个从Queue里取出数据。 

import multiprocessing
import time


def write_queue(queue):
    # 循环写入数据
    for i in range(10):
        if queue.full():
            print("队列已满!")
            break
        # 向队列中放入消息
        queue.put(i)
        print(i)
        time.sleep(0.5)


def read_queue(queue):
    # 循环读取队列消息
    while True:
        # 队列为空，停止读取
        if queue.empty():
            print("---队列已空---")
            break

        # 读取消息并输出
        result = queue.get()
        print(result)


if __name__ == '__main__':

    # 创建消息队列
    queue = multiprocessing.Queue(3)
    # 创建子进程
    p1 = multiprocessing.Process(target=write_queue, args=(queue,))
    p1.start()
    # 等待p1写数据进程执行结束后，再往下执行
    p1.join()
    p1 = multiprocessing.Process(target=read_queue, args=(queue,))
    p1.start()

三、进程池消息队列

        初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务。 

创建

import multiprocessing
pool = multiprocessing.Pool(最大进程数)

进程池内通信 
创建进程池内Queue消息队列通信 
import multiprocessing 
Queue：queue = multiprocessing.Manager().Queue() 
例： 
import multiprocessing 
import time

事例：

def write_data(queue): 
    # for循环 向消息队列中写入值 
    for i in range(5): 
    # 添加消息 
        queue.put(i) 
        print(i) 
        time.sleep(0.2) 
        print(“队列已满~”)

def read_data(queue):
    # 循环读取数据
    while True:
        # 判断队列是否为空
        if queue.qsize() == 0:
            print("队列为空～")
            break
        # 从队列中读取数据
        result = queue.get()
        print(result)


if __name__ == '__main__':
    # 创建进程池
    pool = multiprocessing.Pool(2)
    # 创建进程池队列
    queue = multiprocessing.Manager().Queue()
    # 在进程池中的进程间进行通信
    # 使用线程池同步的方式，先写后读
    # pool.apply(write_data, (queue, ))
    # pool.apply(read_data, (queue, ))
    # apply_async() 返回ApplyResult 对象
    result = pool.apply_async(write_data, (queue, ))
    # ApplyResult对象的wait() 方法，表示后续进程必须等待当前进程执行完再继续
    result.wait()
    pool.apply_async(read_data, (queue, ))
    pool.close()
    # 异步后，主线程不再等待子进程执行结束，再结束
    # join() 后，表示主线程会等待子进程执行结束后，再结束
    pool.join()