专门做衣服特卖的网站,wordpress 媒体库缩略图生成,网站开发流程相关知识,wordpress自动注册插件一、分发到多Consumer(fanout) 二、Routing路由(Direct) 三、主题路由(Topic)一、分发到多Consumer(fanout)将同一个Message deliver到多个Consumer中。这个模式也被称为publish/subscribe 创建一个日志系统#xff0c;包含两部分#xff1a;第一部分发出log(Pro… 一、分发到多Consumer(fanout) 二、Routing路由(Direct) 三、主题路由(Topic)一、分发到多Consumer(fanout)将同一个Message deliver到多个Consumer中。这个模式也被称为publish/subscribe 创建一个日志系统包含两部分第一部分发出log(Producer),第二部分接收到并打印(Consumer)。两个Consumer第一个将log写到物理磁盘上第二个将log输出的屏幕。1.发送消息流程 1.Producer发送的Message实际上是发到了Exchange中。 2.Exchanges从Producer接收message投递到queue中 3.Prducer发送的消息只是到达了Exchange中Exchange具有不同的类型实现不同的分发方式 Exchnges的类型direct、topic和fanout fanout就是广播模式会将所有的Message都放到它所知道的queue中 channel.exchange_declare(exchangelogs, typefanout) //创建一个名字为logs类型为fanout的Exchange 1 2 3 4 5 6 7 8 9 10 11 [rootnode112 ~]# rabbitmqctl list_exchanges //查看所有的Exchanges Listing exchanges ... logs fanout amq.direct direct amq.fanout fanout amq.headers headers amq.match headers amq.rabbitmq.log topic amq.rabbitmq.trace topic amq.topic topic ...done. 注意amq.* exchanges 和the default (unnamed)exchange是RabbitMQ默认创建的。 通过exchange而不是routing_key来publish Message channel.basic_publish(exchangelogs, routing_key, bodymessage) 2.临时队列 截至现在我们用的queue都是有名字的第一个是hello第二个是task_queue。使用有名字的queue使得在Producer和Consumer之前共享queue成为可能。 但是对于我们将要构建的日志系统并不需要有名字的queue。我们希望得到所有的log而不是它们中间的一部分。而且我们只对当前的log感兴趣。为了实现这个目标我们需要两件事情 1每当Consumer连接时我们需要一个新的空的queue。因为我们不对老的log感兴趣。幸运的是如果在声明queue时不指定名字那么RabbitMQ会随机为我们选择这个名字。方法 result channel.queue_declare() 通过result.method.queue 可以取得queue的名字。基本上都是这个样子amq.gen-JzTY20BRgKO-HjmUJj0wLg。 2当Consumer关闭连接时这个queue要被deleted。可以加个exclusive的参数。方法 result channel.queue_declare(exclusiveTrue) //每次获取的都是新的单独使用的 3.Bindings绑定 创建好fanout类型的Exchange和没有名字的queue后实际上是RabbitMQ帮我们取的名字Exchange通过bindings把它的Message发送到目标queue channel.queue_bind(exchangelogs, queueresult.method.queue) 使用命令rabbitmqctl list_bindings 查看bindings 4.最终代码拓扑图 Producer在这里就是产生log的program基本上和前几个都差不多。最主要的区别就是publish通过了exchange而不是routing_key。 emit_log.py script: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 #!/usr/bin/env python import pika import sys connection pika.BlockingConnection(pika.ConnectionParameters( hostlocalhost)) channel connection.channel() channel.exchange_declare(exchangelogs, typefanout) message .join(sys.argv[1:]) or info: Hello World! channel.basic_publish(exchangelogs, routing_key, bodymessage) print [x] Sent %r % (message,) connection.close() 还有一点要注意的是我们声明了exchange。publish到一个不存在的exchange是被禁止的。如果没有queue bindings exchange的话log是被丢弃的。 Consumerreceive_logs.py: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 #!/usr/bin/env python import pika connection pika.BlockingConnection(pika.ConnectionParameters( hostlocalhost)) channel connection.channel() channel.exchange_declare(exchangelogs, typefanout) result channel.queue_declare(exclusiveTrue) queue_name result.method.queue channel.queue_bind(exchangelogs, queuequeue_name) print [*] Waiting for logs. To exit press CTRLC def callback(ch, method, properties, body): print [x] %r % (body,) channel.basic_consume(callback, queuequeue_name, no_ackTrue) channel.start_consuming() 试运行 Consumer1:$ python receive_logs.py logs_from_rabbit.log //追加到文件 Consumer2:python receive_logs.py //输出到屏幕 Producer:python emit_log.py 也可通过修改callback自己写文件 输出结果如图二、Routing路由(Direct)对于上一个日志系统改进。能够使用不同的severity来监听不同等级的log。比如我们希望只有error的log才保存到磁盘上。1.Bindings绑定 之前的绑定 channel.queue_bind(exchangeexchange_name, queuequeue_name) 绑定其实就是关联了exchange和queue。或者这么说queue对exchagne的内容感兴趣exchange要把它的Message deliver到queue中。 实际上绑定可以带routing_key 这个参数。其实这个参数的名称和basic_publish 的参数名是相同了。为了避免混淆我们把它成为binding key。 使用一个key来创建binding : channel.queue_bind(exchangeexchange_name, queuequeue_name, routing_keyblack) 对于fanout的exchange来说这个参数是被忽略的。2.Direct Exchange通过Bindings key完全匹配 图Direct路由模型 exchange X和两个queue绑定在一起。Q1的binding key是orange。Q2的binding key是black和green。 当P publish key是orange时exchange会把它放到Q1。如果是black或者green那么就会到Q2。其余的Message都会被丢弃。3.多重绑定(Multiple Bindings)多个queue绑定同一个key是可以的。对于下图的例子Q1和Q2都绑定了black。也就是说对于routing key是black的Message会被deliver到Q1和Q2。其余的Message都会被丢弃。 图muliti-bindings 4.生产者和消费者生产者 1 2 3 4 5 6 7 8 channel.exchange_declare(exchangedirect_logs, typedirect) //创建一个direct的exchange。使用log的severity作为routing key这样Consumer可以针对不同severity的log进行不同的处理。 publish channel.basic_publish(exchangedirect_logs, routing_keyseverity, bodymessage) //涉及三种severityinfo, warning, error. 消费者 1 2 3 4 5 6 7 result channel.queue_declare(exclusiveTrue) queue_name result.method.queue for severity in severities: channel.queue_bind(exchangedirect_logs, queuequeue_name, routing_keyseverity) //queue需要绑定severity 5.最终版本图direct_2 emit_log_direct.py 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 #!/usr/bin/env python import pika import sys connection pika.BlockingConnection(pika.ConnectionParameters( hostlocalhost)) channel connection.channel() channel.exchange_declare(exchangedirect_logs, typedirect) severity sys.argv[1] if len(sys.argv) 1 else info message .join(sys.argv[2:]) or Hello World! channel.basic_publish(exchangedirect_logs, routing_keyseverity, bodymessage) print [x] Sent %r:%r % (severity, message) connection.close() receive_logs_direct.py: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 #!/usr/bin/env python import pika import sys connection pika.BlockingConnection(pika.ConnectionParameters( hostlocalhost)) channel connection.channel() channel.exchange_declare(exchangedirect_logs, typedirect) result channel.queue_declare(exclusiveTrue) queue_name result.method.queue severities sys.argv[1:] if not severities: print sys.stderr, Usage: %s [info] [warning] [error] % \ (sys.argv[0],) sys.exit(1) for severity in severities: channel.queue_bind(exchangedirect_logs, queuequeue_name, routing_keyseverity) print [*] Waiting for logs. To exit press CTRLC def callback(ch, method, properties, body): print [x] %r:%r % (method.routing_key, body,) channel.basic_consume(callback, queuequeue_name, no_ackTrue) channel.start_consuming() 试运行 $ python receive_logs_direct.py warning error logs_from_rabbit.log //把warning和error的log记录到一个文件中 $ python receive_logs_direct.py info warning error //打印所有log到屏幕 三、主题路由(Topic)1.Topic exchangeMessage的routing_key使用限制不能使任意的。格式是以点号“.分割的字符表。 比如stock.usd.nyse, nyse.vmw, quick.orange.rabbit。你可以放任意的key在routing_key中当然最长不能超过255 bytes。 对于routing_key有两个特殊字符在正则表达式里叫元字符 * (星号) 代表任意 一个单词 # (hash) 0个或者多个单词 示例: Producer发送消息时需要设置routing_keyrouting_key包含三个单词和两个点号。 第一个key是描述了celerity灵巧敏捷第二个是colour色彩第三个是species物种celerity.colour.species。 在这里我们创建了两个绑定 Q1 的binding key 是*.orange.* Q2 是 *.*.rabbit 和 lazy.# Q1 感兴趣所有orange颜色的动物 Q2 感兴趣所有的rabbits和所有的lazy的 比如routing_key是 quick.orange.rabbit将会发送到Q1和Q2中。消息lazy.orange.elephant 也会发送到Q1和Q2。但是quick.orange.fox 会发送到Q1lazy.brown.fox会发送到Q2。lazy.pink.rabbit 也会发送到Q2但是尽管两个routing_key都匹配它也只是发送一次。quick.brown.fox 会被丢弃。 如果发送的单词不是3个呢 答案要看情况因为#是可以匹配0个或任意个单词。比如orange or quick.orange.male.rabbit它们会被丢弃。如果是lazy那么就会进入Q2。类似的还有 lazy.orange.male.rabbit尽管它包含四个单词。 Topic exchange和其他exchange 由于有* (star) and # (hash) Topic exchange 非常强大并且可以转化为其他的exchange 如果binding_key 是 # - 它会接收所有的Message不管routing_key是什么就像是fanout exchange。 如果 * (star) and # (hash) 没有被使用那么topic exchange就变成了direct exchange。2.代码实现The code for emit_log_topic.py: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 #!/usr/bin/env python import pika import sys connection pika.BlockingConnection(pika.ConnectionParameters( hostlocalhost)) channel connection.channel() channel.exchange_declare(exchangetopic_logs, typetopic) routing_key sys.argv[1] if len(sys.argv) 1 else anonymous.info message .join(sys.argv[2:]) or Hello World! channel.basic_publish(exchangetopic_logs, routing_keyrouting_key, bodymessage) print [x] Sent %r:%r % (routing_key, message) connection.close() The code for receive_logs_topic.py: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 #!/usr/bin/env python import pika import sys connection pika.BlockingConnection(pika.ConnectionParameters( hostlocalhost)) channel connection.channel() channel.exchange_declare(exchangetopic_logs, typetopic) result channel.queue_declare(exclusiveTrue) queue_name result.method.queue binding_keys sys.argv[1:] if not binding_keys: print sys.stderr, Usage: %s [binding_key]... % (sys.argv[0],) sys.exit(1) for binding_key in binding_keys: channel.queue_bind(exchangetopic_logs, queuequeue_name, routing_keybinding_key) print [*] Waiting for logs. To exit press CTRLC def callback(ch, method, properties, body): print [x] %r:%r % (method.routing_key, body,) channel.basic_consume(callback, queuequeue_name, no_ackTrue) channel.start_consuming() 3.运行和结果 python receive_logs_topic.py # //接收所有的log python receive_logs_topic.py kern.* //接收所有kern facility的log python receive_logs_topic.py *.critical //仅仅接收critical的log python receive_logs_topic.py kern.* *.critical //可以创建多个绑定 python emit_log_topic.py kern.critical A critical kernel error //Producer产生一个logkern.critical type: 参考 http://www.rabbitmq.com/tutorials/tutorial-three-python.html 本文转自MT_IT51CTO博客原文链接http://blog.51cto.com/hmtk520/2051247如需转载请自行联系原作者
