import time
from multiprocessing import Pipe, Process
 
# 发送消息的进程 
def proc1(pipe):
    for i in xrange(1, 10):
        pipe.send(i)
        print ("send {0} to pipe".format(i))
        time.sleep(1)
 
# 接收消息的进程
def proc2(pipe):
    n = 9
    while n > 0:
        result = pipe.recv()
        print ("recv {0} from pipe".format(result))
        n -= 1
 
 
def main():
    pipe = Pipe(duplex=False)  # 设置半双工模式，p1只负责发送消息，p2只负责接收消息，pipe是一个tuple类型
    p1 = Process(target=proc1, args=(pipe[1],))
    p2 = Process(target=proc2, args=(pipe[0],)) #接收写0
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    pipe[0].close()
    pipe[1].close()
 
 
if __name__ == ‘__main__‘:
    main()

结果：
send 1 to pipe
recv 1 from pipe
recv 2 from pipe
send 2 to pipe
send 3 to pipe
recv 3 from pipe
recv 4 from pipe
send 4 to pipe
send 5 to pipe
recv 5 from pipe
recv 6 from pipe
send 6 to pipe
recv 7 from pipe
send 7 to pipe
recv 8 from pipe
send 8 to pipe
send 9 to pipe
recv 9 from pipe

from multiprocessing import Queue
from threading import Thread
import time

"""
一个生产者和两个消费者，
采用多线程继承的方式，
一个消费偶数，一个消费奇数。
"""


class Proceducer(Thread):
    def __init__(self, queue):
        super(Proceducer, self).__init__()
        self.queue = queue

    def run(self):
        try:
            for i in xrange(1, 10):
                print ("put {0} to queue".format(i))
                self.queue.put(i)
        except Exception as e:
            print ("put data error")
            raise e


class Consumer_even(Thread):
    def __init__(self, queue):
        super(Consumer_even, self).__init__()
        self.queue = queue

    def run(self):
        try:
            while not self.queue.empty():   # 判断队列是否为空
                number = self.queue.get(block=True, timeout=3)  # 从队列中获取消息，block=True表示阻塞，设置超时未3s
                if number % 2 == 0:   # 如果获取的消息是偶数
                    print("get {0} from queue EVEN, thread name is {1}".format(number, self.getName()))
                else:
                    self.queue.put(number)  # 如果获取的消息不是偶数，就接着把它放回队列中
                time.sleep(1)
        except Exception as e:
            raise e


class Consumer_odd(Thread):
    def __init__(self, queue):
        super(Consumer_odd, self).__init__()
        self.queue = queue

    def run(self):
        try:
            while not self.queue.empty():
                number = self.queue.get(block=True, timeout=3)
                if number % 2 != 0:  # 如果获取的消息是奇数
                    print("get {0} from queue ODD, thread name is {1}".format(number, self.getName()))
                else:
                    self.queue.put(number)
                time.sleep(1)
        except Exception as e:
            raise e


def main():
    queue = Queue()
    p = Proceducer(queue=queue)
    # 开始产生消息
    print("开始产生消息")
    p.start()
    p.join()   # 等待生产消息的进程结束
    time.sleep(1) # 消息生产完成之后暂停1s
    c1 = Consumer_even(queue=queue)
    c2 = Consumer_odd(queue=queue)

    # 开始消费消息
    print("开始消费消息")
    c1.start()
    c2.start()
    c1.join()
    c2.join()
    print ("消息消费完成")


if __name__ == ‘__main__‘:
    main()

结果：
开始产生消息
put 1 to queue
put 2 to queue
put 3 to queue
put 4 to queue
put 5 to queue
put 6 to queue
put 7 to queue
put 8 to queue
put 9 to queue
开始消费消息
get 1 from queue ODD, thread name is Thread-3
get 2 from queue EVEN, thread name is Thread-2
get 3 from queue ODD, thread name is Thread-3
get 4 from queue EVEN, thread name is Thread-2
get 5 from queue ODD, thread name is Thread-3
get 6 from queue EVEN, thread name is Thread-2
get 7 from queue ODD, thread name is Thread-3
get 8 from queue EVEN, thread name is Thread-2
get 9 from queue ODD, thread name is Thread-3
消息消费完成

[root@izwz920j4zsv1q15yhii1qz scripts]# celery -A celery_test worker -l info
/usr/lib/python2.7/site-packages/celery/platforms.py:796: RuntimeWarning: You‘re running the worker with superuser privileges: this is absolutely not recommended!

Please specify a different user using the -u option.

User information: uid=0 euid=0 gid=0 egid=0

  uid=uid, euid=euid, gid=gid, egid=egid,
 
 -------------- celery@izwz920j4zsv1q15yhii1qz v4.1.1 (latentcall)
---- **** ----- 
--- * ***  * -- Linux-3.10.0-693.2.2.el7.x86_64-x86_64-with-centos-7.4.1708-Core 2018-05-25 14:28:38
-- * - **** --- 
- ** ---------- [config]
- ** ---------- .> app:         celery_test:0x25a6450
- ** ---------- .> transport:   redis://110.106.106.220:5000/5
- ** ---------- .> results:     redis://110.106.106.220:5000/6
- *** --- * --- .> concurrency: 1 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                

[tasks]
  . tasks.add

[2018-05-25 14:28:38,431: INFO/MainProcess] Connected to redis://110.106.106.220:5000/5
[2018-05-25 14:28:38,443: INFO/MainProcess] mingle: searching for neighbors
[2018-05-25 14:28:39,475: INFO/MainProcess] mingle: all alone
[2018-05-25 14:28:39,528: INFO/MainProcess] celery@izwz920j4zsv1q15yhii1qz ready.

vi trigger.py
from tasks import add
result = add.delay(4, 4) #不要直接 add(4, 4)，这里需要用 celery 提供的接口 delay 进行调用
while not result.ready():  # 是否处理
    time.sleep(1)
print ‘task done: {0}‘.format(result.get())  # 获取结果
print(result.task_id)