继续在web开发的路上前行
之前学的通信，异步，现在回过头来看，已经没有那么难，希望以后能排上用场。这个星期学的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(“ 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(“ %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 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 if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',
routing_key=severity,
body=message)
print(" 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
if not severities:
sys.stderr.write("Usage: %s \n" % sys.argv)
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(" %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 if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv) or 'Hello World!'
channel.basic_publish(exchange='topic_logs',
routing_key=routing_key,
body=message)
print(" 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
if not binding_keys:
sys.stderr.write("Usage: %s ...\n" % sys.argv)
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(" %r:%r" % (method.routing_key, body))
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()
页:
[1]