之前学的通信，异步，现在回过头来看，已经没有那么难，希望以后能排上用场。这个星期学的rabbitmq，redis都是挺好玩的，不过我还没有玩出个花样来。为什么读中学大学时候要抑制女生玩游戏呢，其实女生也是很喜欢玩游戏的，也挺喜欢编程的。
Python 中rabbitmq的主要模块有pika, celery, haigha
就是相当于把队列放在一台远程服务器上，然后连上这个远程服务器，取数据，发数据。
- 1. 基本版
#######生产者#########
import pika
#申明链接
connection = pika. BlockingConnection(pika.ConnectionParameters(“localhost”))
#申明管道
channel = connection.channel()
#申明queue
channel.queue_declare(queue=’hello’)
#发消息
channel.basic_publish(exchange = “”,
Routing_key = “hello”, #queue 的名字
Body = “hello word” #所发的消息内容
)
connection.close()
#######消费者#########
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(‘localhost’))
channel = connection.channel()
#因为不确定是消费者还是生产者谁先运行，所以需要声明。防止生产者端后启动报错
channel.queue_declare(queue=’hello’)
#定义回调函数，一旦消息来了，就先执行这个函数来处理消息
def callback(ch,method,properties,body): #ch是管道内寸对象地址
print(“have received the message”,body)
ch.basic_ack(delivery_tag = method.delivery_tag) # 需要手动确认，所以no_ack = False, 那么消费者端需要手动和生产者端确认。
channel.basic_consume(callback,
queue= “hello”,
no_ack = True #不管消息处理完没有，都不会给消费者一个回复，默认不加。no_ack=False, 这样的话服务器端不会将消息删除，知道消息被处理，收到消费者的确认之后，才会删除消息。
)
channel.start_consuming() #一启动就会一直收消息，要是没有消息，那就等待消息来
如果一个生产者对应多个消费者，那么消费者之间会进行负载均衡，消息发送轮询。
队列和消息持久化：
|
消费者死机 |
服务器死机 |
|
那生产者应该默认这条消息没有被处理完，所以不能把这条消息从消息队列中删除。 消费者： basic_consume 中no_ack =False, callback 函数中: ch.basic_ack(delivery_tag = method.delivery_tag) |
在申明队列的时候, 声明队列是持久化，但是队列中的消息会没有。消费者端和生产者端都需要写。 队列持久化 生产者、消费者： Channel.queue_decalre(queue=’hello’, durable=True) 消息持久化 生产者： poperties = pika.BasicProperties( dlivery_mode = 2) |
- 2. 广播模式
公平消息分发：根据处理器的处理能力来分配消息的量
消费者：
Channel.basic_qos(perfetch_count =1) #消费者要是还有一条消息的时候，生产者就不给他发送
要实现一个生产者发送消息，所有消费者都收到，那么需要广播模式。Exchange的类型，决定了消息如何被处理。
fanout: 所有绑定在这个exchange的queue都可以接收消息
|
生产者 |
消费者 |
|
import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters (host = ‘localhost’)) channel = connection.channel() channel.exchange_declare( exchange = ‘logs’, type = ‘fanout’) message = ‘info:hello world’ channel.basix_publish(exchange = ‘logs’, routing_key = ‘’, body = message) print(“[x] sent %r”%message) connection.close() |
import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host = “localhost”)) channel = connection.channel() channel.exchange_declare(exchange=’logs’, type = ‘fanout’) result = channel.queue_declare(exclusive = True) queue_name = result.method.queue #不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除 def callback(ch, method, properties, body): print(“[x] %r” %body) channel.basic_consume(callback, queue=queue_name, no_ack = True ) channel.start_consuming() |
- 3. 选择模式
接收者只接收自己需要的信息
direct:通过routinekey和exchange决定那个唯一的queue可以接收消息
channel.exchange_declare(exchange=”direct_logs”, type = “direct”)
severity = sys.argv[1] if len(sys.argv) > 1 else “info” #选择接收，有不同级别的信息，info的信息发到一个队列，error的信息发送到另外一个队列。
|
生产者 |
消费者 |
|
import pika import sys
connection= pika.BlockingConnection (pika.ConnectionParameters(host='localhost')) channel = connection.channel()
channel.exchange_declare(exchange='direct_logs', type='direct')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info' message = ' '.join(sys.argv[2:]) or 'Hello World!' channel.basic_publish(exchange='direct_logs', routing_key=severity, body=message) print(" [x] Sent %r:%r" % (severity, message)) connection.close()
|
import pika import sys
connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs', type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = sys.argv[1:]
if not severities:
sys.stderr.write("Usage: %s [info] [warning] [error] " % sys.argv[0]) sys.exit(1)
for severity in severities: channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key=severity)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body): print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming() |
- 4. 更细致的消息过滤模式
topic：所有含有routingkey和routingkey所bind的queue都可以收消息
|
生产者 |
消费者 |
|
import pika import sys
connectionpika.BlockingConnection (pika.ConnectionParameters(host='localhost')) channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',type='topic')
routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info' message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='topic_logs', routing_key=routing_key, body=message)
print(" [x] Sent %r:%r" % (routing_key, message)) connection.close() |
import pika import sys
connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel()
channel.exchange_declare(exchange='topic_logs',type='topic')
result = channel.queue_declare(exclusive=True) queue_name = result.method.queue
binding_keys = sys.argv[1:]
if not binding_keys: sys.stderr.write("Usage: %s [binding_key]... " % sys.argv[0]) sys.exit(1)
for binding_key in binding_keys: channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key)
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming() |