python-RabbitMQ

简单队列方式

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import Queue
import threading


#建立队列长度为10
message = Queue.Queue(10)

#生产者
def producer(i):
        #往队列中扔消息
        message.put(i)
        print('queue_size=%s'%message.queue)
#消费者
def consumer(i):
        #取队列消息，先进先出
        msg = message.get()


#生产线程
for i in range(10):
    t = threading.Thread(target=producer, args=(i,))
    t.start()
#消费线程
for i in range(5):
    t = threading.Thread(target=consumer, args=(i,))
    t.start()
    
#打印剩余的数据
print('queue_size=%s'%message.queue)

结果：

生产队列增长过程过程

producer_queue=deque([0])
producer_queue=deque([0, 1])
producer_queue=deque([0, 1, 2])
producer_queue=deque([0, 1, 2, 3])
producer_queue=deque([0, 1, 2, 3, 4])
producer_queue=deque([0, 1, 2, 3, 4, 5])
producer_queue=deque([0, 1, 2, 3, 4, 5, 6])
producer_queue=deque([0, 1, 2, 3, 4, 5, 6, 7])
producer_queue=deque([0, 1, 2, 3, 4, 5, 6, 7, 8])
producer_queue=deque([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])

剩余数据
queue=deque([4, 5, 6, 7, 8, 9])

 

 

 

 

 

以下是RabbitMQ 主要模式

1简单队列

2exchange{

    1、fanout

    2、direct

    3 、topic

}

 

安装组件

pip install  pika

 

对于RabbitMQ来说，生产和消费不再针对内存里的一个Queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。

1、简单队列模型、消息不丢失持久化参数durable 

 生产者

 1 # -*- coding:utf-8 -*-
 2 import pika
 3 #认证的用户密码
 4 credentials = pika.PlainCredentials('openstack', 'RABBIT_PASS')
 5 #远程主机配置
 6 parameters = pika.ConnectionParameters('172.16.21.5',5673,'/',credentials)
 7 
 8 #封装socket逻辑部分，拿到操作句柄
 9 connection = pika.BlockingConnection(parameters)
10 channel = connection.channel()
11 
12 
13 #创建队列名字为hello,这个队列的名字一般是唯一的.消费者连接的信道名称也是这个
14 channel.queue_declare(queue='hello')
15 
16 # 第一种工作状态
17 #basic_publish设置队列
18 # exchange='' 就是简单队列的意思，这里exchange不工作
19 # routing_key='hello' ，exchange不工作了那么这里如何处理，就靠routing_key来找对应队列
20 #bodybody='Hello World!' 这就是传递的数据
21 channel.basic_publish(exchange='',
22                       routing_key='hello',
23                       body='Hello World!')
24 channel.basic_qos(prefetch_count=1)
25 print(" [x] Sent 'Hello World!'")
26 connection.close()

View Code