金沙澳门官网登录-www.js8.com-金沙澳门官网下载app

Python开发【第十篇】:RabbitMQ队列

no_ack=False(此时为手动应答)

在这种情况下,要求consumer在处理完接收到的Basic.Deliver+Content-Header+Content-Body之后才回复Ack,而这个Ack是AMQP协议中的Basic.Ack。此Ack的回复与业务处理相关,所以具体的回复时间应该要取决于业务处理的耗时。

远程过程调用(RPC)

RPC(Remote Procedure Call Protocol)远程过程调用协议。在一个大型的公司,系统由大大小小的服务构成,不同的团队维护不同的代码,部署在不同的服务器。但是在做开发的时候往往要用到其他团队的方法,因为已经有了实现。但是这些服务部署在不同的服务器,想要调用就需要网络通信,这些代码繁琐且复杂,一不小心就会很低效。PRC协议定义了规划,其它的公司都给出了不同的实现。比如微软的wcf,以及WebApi。

在RabbitMQ中RPC的实现是很简单高效的,现在客户端、服务端都是消息发布者与消息接受者。

图片 1

首先客户端通过RPC向服务端发生请求。correlation_id:请求标识,erply_to:结果返回队列。(我这里有一些数据需要你给我处理一下,correlation_id是我请求标识,你处理完成之后把结果返回到erply_to队列)

服务端拿到请求,开始处理并返回。correlation_id:客户端请求标识。(correlation_id这是你的请求标识,还给你。这时候客户端用自己的correlation_id与服务端返回的correlation_id进行对比,相同则接收。)

rpc_server

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika,time

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

  1. channel = connection.channel()

  2.  

  3. channel.queue_declare(queue='rpc_queue')

  1. def fib(n):

  2.     if n == 0:

  3.         return 0

  4. Python开发【第十篇】:RabbitMQ队列。    elif n == 1:

  5.         return 1

  6.     else:

  7.         return fib(n-1) + fib(n-2)

  8.  

  9. def on_request(ch,method,props,body):

  1.     n = int(body)

  2.     print('[.] fib(%s)' % n)

  3.     response = fib(n)

  4. Python开发【第十篇】:RabbitMQ队列。    ch.basic_publish(exchange='',

  1.                      routing_key=props.reply_to,
  1.                      properties=pika.BasicProperties(correlation_id=props.correlation_id),
  1.                      body = str(response))

  2.     ch.basic_ack(delivery_tag=method.delivery_tag)

  1.  

  2. channel.basic_qos(prefetch_count=1)

  1. channel.basic_consume(on_request,queue='rpc_queue')
  1.  

  2. print('[x] Awaiting RPC requests')

  1. channel.start_consuming()

rpc_client

  1. __author__Python开发【第十篇】:RabbitMQ队列。 = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika,uuid

  6.  

  7. class FibonacciRpcClient(object):

  8.     def __init__(self):

  9.         self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

  1.         self.channel = self.connection.channel()

  2.         result = self.channel.queue_declare(exclusive=True)

  3.         self.callback_queue = result.method.queue

  4.         self.channel.basic_consume(self.on_response,no_ack=True,

  1.                                    queue=self.callback_queue)
  1.  

  2.     def on_response(self,ch,method,props,body):

  3.         if self.corr_id == props.correlation_id:

  4.             self.response = body

  1.  

  2.     def call(self,n):

  3.         self.response = None

  4.         self.corr_id = str(uuid.uuid4())

  5.         self.channel.basic_publish(exchange='',

  1.                                    routing_key='rpc_queue',
  1.                                    properties=pika.BasicProperties(
  1.                                        reply_to=self.callback_queue,
  1.                                        correlation_id=self.corr_id,),
  1.                                    body=str(n))
  1.         while self.response is None:

  2.             self.connection.process_data_events()

  1.         return int(self.response)

  2.  

  3. fibonacci_rpc = FibonacciRpcClient()

  1.  

  2. print('[x] Requesting fib(10)')

  1. response = fibonacci_rpc.call(10)
  1. print('[.] Got %r ' % response)

 

durable消息持久化

producer发送消息时挂掉了,consumer接收消息时挂掉了,以下方法会让RabbitMQ重新将该消息添加到队列中。

回调函数中

  1. ch.basic_ack(delivery_tag=method.delivery_tag)

basic_consume中

  1. no_ack=False

basic_publish中添加参数

  1. properties=pika.BasicProperties(delivery_mode=2)

channel.queue_declare中添加参数

  1. channel.queue_declare(queue='hello',durable=True)

send端(producer)

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

  1. channel = connection.channel()

  2.  

  3. # 声明queue

  4. channel.queue_declare(queue='hello',durable=True)

  1.  

  2. channel.basic_publish(exchange='',

  1.                       routing_key='hello',
  1.                       body='hello word',

  2.                       properties=pika.BasicProperties(delivery_mode=2))

  1. print("[x] Sent 'hello word!'")

  2. connection.close()

receive端(consumer)与acknowledgment消息持久化中receive端(consumer)相同。

Linux安装步骤

安装erlang。

  1. yum -y install erlang

安装RabbitMQ。

  1. wget
  1. rpm -ivh rabbitmq-server-3.6.10-1.el6.noarch.rpm

RabbitMQ安装失败,报错如下。

  1. warning: rabbitmq-server-3.6.10-1.el6.noarch.rpm: Header V4 RSA/SHA512 Signature, key ID 6026dfca: NOKEY

  2. error: Failed dependencies:

  3.         erlang >= R16B-03 is needed by rabbitmq-server-3.6.10-1.el6.noarch

  4.         socat is needed by rabbitmq-server-3.6.10-1.el6.noarch

原因是yum安装的erlang版本太低,这里提供的RabbitMQ是最新版3.6.10,所需的erlang版本最低为R16B-03,否则编译时将失败,也就是上述错误。

重新安装erlang。

  1. wget
  1. tar xvzf otp_src_20.0.tar.gz

  2. cd otp_src_20.0

  3. ./configure

  4. make && make install

重新安装erlang完毕。

运行erlang。

  1. erl

  2. Erlang/OTP 20 [erts-9.0] [source] [64-bit] [smp:1:1] [ds:1:1:10] [async-threads:10] [hipe] [kernel-poll:false]

  3.  

  4. Eshell V9.0 (abort with ^G)

安装socat。

  1. yum install -y socat

再次安装RabbitMQ。

  1. rpm -ivh rabbitmq-server-3.6.10-1.el6.noarch.rpm

  2. warning: rabbitmq-server-3.6.10-1.el6.noarch.rpm: Header V4 RSA/SHA512 Signature, key ID 6026dfca: NOKEY

  3. error: Failed dependencies:

  4.         erlang >= R16B-03 is needed by rabbitmq-server-3.6.10-1.el6.noarch

上述错误信息显示安装失败,因为rabbitMQ的依赖关系所导致,所以要忽略依赖,执行以下命令。

  1. rpm -ivh --nodeps rabbitmq-server-3.6.10-1.el6.noarch.rpm

安装成功。

启动、停止RabbitMQ。

  1. rabbitmq-server start     #启动
  1. rabbitmq-server stop     #停止
  1. rabbitmq-server restart    #重启

 

消息持久化

RabbitMQ使用

实现最简单的队列通信

图片 2

send端(producer)

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

  1. channel = connection.channel()

  2.  

  3. # 声明queue

  4. channel.queue_declare(queue='hello')

  1.  

  2. channel.basic_publish(exchange='',

  1.                       routing_key='hello',
  1.                       body='hello word')

  2. print("[x] Sent 'hello word!'")

  3. connection.close()

receive端(consumer)

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika,time

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

  1. channel = connection.channel()

  2.  

  3. channel.queue_declare(queue='hello')

  1.  

  2. def callback(ch,method,properties,body):

  3.     print('-->',ch,method,properties)

  1.     print("[x] Received %s" % body)

  2.  

  3. channel.basic_consume(callback,

  1.                       queue='hello',
  1.                       no_ack=True
  1.                       )

  2.  

  3. print('[*] waiting for messages.To exit press CTRL+C')

  1. channel.start_consuming()

简介

RabbitMQ是流行的开源消息队列系统,用erlang语言开发。RabbitMQ是AMQP(高级消息队列协议)的标准实现。

消息发布和订阅(publishsubscribe)

发布和订阅与简单的消息队列区别在于,发布和订阅会将消息发送给所有的订阅者,而消息队列中的数据被消费一次便消失。所以,RabbitMQ实现发布和订阅时,会为每一个订阅者创建一个队列,而发布者发布消息时,会将消息放置在所有相关队列中。类似广播的效果,这时候就要用到exchange。Exchange在定义的时候是有类型的,以决定到底是哪些Queue符合条件,可以接收消息。

fanout:所有bind到此exchange的queue都可以接收消息。

direct:通过routingKey和exchange决定的哪个唯一的queue可以接收消息。

topic:所有符合routingKey(可以是一个表达式)的routingKey所bind的queue可以接收消息。

表达式符号说明

#:一个或多个字符

*:任何字符

例如:#.a会匹配a.a,aa.a,aaa.a等。

*.a会匹配a.a,b.a,c.a等。

注意:使用RoutingKey为#,Exchange Type为topic的时候相对于使用fanout。

heaers:通过headers来决定把消息发给哪些queue。

图片 3

publisher

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika,sys

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

  1. channel = connection.channel()

  2.  

  3. channel.exchange_declare(exchange='logs',type='fanout')

  1.  

  2. message = ''.join(sys.argv[1:]) or 'info:Hello World!'

  3. channel.basic_publish(exchange='logs',

  1.                       routing_key='',
  1.                       body=message)
  1.  

  2. print('[x] Send %r' % message)

  1. connection.close()

subscriber

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

  1. channel = connection.channel()

  2. channel.exchange_declare(exchange='logs',type='fanout')

  1. # 不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
  1. result = channel.queue_declare(exclusive=True)

  2. queue_name = result.method.queue

  1. channel.queue_bind(exchange='logs',queue=queue_name)
  1. print('[*]Waiting for logs.To exit press CTRL+C')

  2. def callback(ch,method,properties,body):

  3.     print('[*] %s'%body)

  4.  

  5. channel.basic_consume(callback,

  1.                       queue=queue_name,
  1.                       no_ack=True)
  1.  

  2. channel.start_consuming()

Windows安装步骤

第一步运行

图片 4

第二步

图片 5

第三步

图片 6

第四步

图片 7

第五步

图片 8

Erlang安装完成。

然后安装RabbitMQ,首先下载RabbitMQ的Windows版本。

官网:

Windows版下载地址:

打开安装程序,按照下面步骤安装。

图片 9

图片 10

图片 11

图片 12

图片 13

RabbitMQ安装完成。

开始菜单中进入管理工具。

图片 14

图片 15

运行命令

  1. rabbitmq-plugins enable rabbitmq_management

图片 16

查看RabbitMQ服务是否启动。

图片 17

图片 18

至此全部安装完成。

消息分发

默认消息队列里的数据是按照顺序分发到各个消费者,但是大部分情况下,消息队列后端的消费者服务器的处理能力是不相同的,这就会出现有的服务器闲置时间较长,资源浪费的情况。那么,我们就需要改变默认的消息队列获取顺序。可以在各个消费者端配置prefetch_count=1,意思就是告诉RabbitMQ在这个消费者当前消息还没有处理完的时候就不要再发新消息了。

图片 19

消费者端

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4. __author__ = 'Golden'

  5. #!/usr/bin/env python3

  6. # -*- coding:utf-8 -*-

  7.  

  8. import pika,time

  9.  

  10. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

  1. channel = connection.channel()

  2.  

  3. channel.queue_declare(queue='hello2',durable=True)

  1.  

  2. def callback(ch,method,properties,body):

  3.     print('-->',ch,method,properties)

  1.     print("[x] Received %s" % body)

  2.     time.sleep(30)

  3.     ch.basic_ack(delivery_tag=method.delivery_tag)

  1.  

  2. channel.basic_qos(prefetch_count=1)

  1. channel.basic_consume(callback,
  1.                       queue='hello2',
  1.                       no_ack=False
  1.                       )

  2.  

  3. print('[*] waiting for messages.To exit press CTRL+C')

  1. channel.start_consuming()

生产者端不变。

安装

首先安装erlang环境。

官网:

Windows版下载地址:

Linux版:yum安装

总结

Basic.Ack发给RabbitMQ以告知,可以将相应message从RabbitMQ的消息从缓存中移除。

Basic.Ack未被consumer发给RabbitMQ前出现了异常,RabbitMQ发现与该consumer对应的连接被断开,将该该message以轮询方式发送给其他consumer(需要存在多个consumer订阅同一个queue)。

在no_ack=true的情况下,RabbitMQ认为message一旦被deliver出去后就已被确认了,所以会立即将缓存中的message删除,因此在consumer异常时会导致消息丢失。

来自consumer的Basic.Ack与发送给Producer的Basic.Ack没有直接关系。

模糊匹配(exchange type=topic)

在topic类型下,可以让队列绑定几个模糊的关键字,发送者将数据发送到exchange,exchange将传入"路由值"和"关键字"进行匹配,匹配成功则将数据发送到指定队列。

图片 20

*:匹配任意一个字符

#:匹配任意个字符

publisher

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika,sys

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

  1. channel = connection.channel()

  2.  

  3. channel.exchange_declare(exchange='topic_logs',

  1.                          type='topic')
  1.  

  2. routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'

  3. message = ''.join(sys.argv[2:]) or 'Hello World!'

  4. channel.basic_publish(exchange='topic_logs',

  1.                       routing_key=routing_key,
  1.                       body=message)
  1.  

  2. print('[x] Sent %r:%r' % (routing_key,message))

  3. connection.close()

subscriber

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika,sys

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

  1. channel = connection.channel()

  2.  

  3. channel.exchange_declare(exchange='topic_logs',

  1.                          type='topic')
  1.  

  2. result = channel.queue_declare(exclusive=True)

  3. queue_name = result.method.queue

  1.  

  2. binding_keys = sys.argv[1:]

  3. if not binding_keys:

  4.     sys.stderr.write('Usage: %s [binding_key]...n' % sys.argv[0])

  5.     sys.exit(1)

  6.  

  7. for binding_key in binding_keys:

  1.     channel.queue_bind(exchange='topic_logs',
  1.                        queue=queue_name,
  1.                        routing_key=binding_key)
  1.  

  2. print('[*] Waiting for logs.To exit press CTRL+C')

  3.  

  4. def callback(ch,method,properties,body):

  5.     print('[x] %r:%r' % (method.routing_key,body))

  6.  

  7. channel.basic_consume(callback,

  1.                       queue=queue_name,
  1.                       no_ack=True)
  1.  

  2. channel.start_consuming()

测试

图片 21

no_ack分析

no_ack属性是在调用Basic.Consume方法时可以设置的一个重要参数。no_ack的用途是确保message被consumer成功处理了。这里成功的意识是,在设置了no_ack=false的情况下,只要consumer手动应答了Basic.Ack,就算其成功处理了。

no_ack=true(此时为自动应答)

在这种情况下,consumer会在接收到Basic.Deliver+Content-Header+Content-Body之后,立即回复Ack,而这个Ack是TCP协议中的Ack。此Ack的回复不关心consumer是否对接收到的数据进行了处理,当然也不关心处理数据所需要的耗时。

关键字发送(echange type=direct)

发送消息时明确指定某个队列并向其中发送消息,RabbitMQ还支持根据关键字发送,即队列绑定关键字,发送者将数据根据关键字发送到消息exchange,exchange根据关键字判定应该将数据发送至哪个队列。

图片 22

publisher

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika,sys

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

  1. channel = connection.channel()

  2.  

  3. channel.exchange_declare(exchange='direct_logs',

  1.                          type='direct')
  1.  

  2. # severity = 'error'

  3. severity = sys.argv[1] if len(sys.argv) > 1 else 'info'

  4. # message = 'Hello World!'

  5. message = ''.join(sys.argv[2:]) or 'Hello World!'

  6.  

  7. channel.basic_publish(exchange='direct_logs',

  1.                       routing_key=severity,
  1.                       body=message)
  1. print('[x] Send %r:%r' % (severity,message))

  2. connection.close()

subscriber

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika,sys

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

  1. channel = connection.channel()

  2.  

  3. channel.exchange_declare(exchange='direct_logs',

  1.                          type='direct')
  1.  

  2. result = channel.queue_declare(exclusive=True)

  3. queue_name = result.method.queue

  1.  

  2. severities = sys.argv[1:]

  3. if not severities:

  4.     sys.stderr.write('Usage: %s [info] [warning] [error]n' % sys.argv[0])

  5.     sys.exit(1)

  6.  

  7. for severity in severities:

  8.     channel.queue_bind(exchange='direct_logs',

  1.                        queue=queue_name,
  1.                        routing_key=severity)
  1.  

  2. print('[*] Waiting for logs.To exit press CTRL+C')

  3.  

  4. def callback(ch,method,properties,body):

  5.     print('[*] %r:%r' % (method.routing_key,body))

  6.  

  7. channel.basic_consume(callback,

  1.                       queue=queue_name,
  1.                       no_ack=True)
  1.  

  2. channel.start_consuming()

启动subscriber1

  1. python3 direct_subscriber.py warning

启动subscriber2

  1. python3 direct_subscriber.py error

启动publisher1

  1. python3 direct_publisher.py info

启动publisher2

  1. python3 direct_publisher.py warning

启动publisher3

  1. python3 direct_publisher.py error

结果

图片 23

acknowledgment消息持久化

no-ack=False,如果consumer挂掉了,那么RabbitMQ会重新将该任务添加到队列中。

回调函数中

  1. ch.basic_ack(delivery_tag=method.delivery_tag)

basic_consume中

  1. no_ack=False

receive端(consumer)

  1. __author__ = 'Golden'

  2. #!/usr/bin/env python3

  3. # -*- coding:utf-8 -*-

  4.  

  5. import pika,time

  6.  

  7. connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

  1. channel = connection.channel()

  2.  

  3. channel.queue_declare(queue='hello')

  1.  

  2. # 定义回调函数

  3. def callback(ch,method,properties,body):

  4.     print('-->',ch,method,properties)

  1.     print("[x] Received %s" % body)

  2.     ch.basic_ack(delivery_tag=method.delivery_tag)

  1.  

  2. # no_ack=False表示消费完以后不主动把状态通知RabbitMQ

  3. channel.basic_consume(callback,

  1.                       queue='hello',
  1.                       no_ack=False
  1.                       )

  2.  

  3. print('[*] waiting for messages.To exit press CTRL+C')

  1. channel.start_consuming()

本文由金沙澳门官网登录发布于金沙澳门官网下载app,转载请注明出处:Python开发【第十篇】:RabbitMQ队列

TAG标签:
Ctrl+D 将本页面保存为书签,全面了解最新资讯,方便快捷。