• linkedu视频
  • 平面设计
  • 电脑入门
  • 操作系统
  • 办公应用
  • 电脑硬件
  • 动画设计
  • 3D设计
  • 网页设计
  • CAD设计
  • 影音处理
  • 数据库
  • 程序设计
  • 认证考试
  • 信息管理
  • 信息安全
菜单
linkedu.com
  • 网页制作
  • 数据库
  • 程序设计
  • 操作系统
  • CMS教程
  • 游戏攻略
  • 脚本语言
  • 平面设计
  • 软件教程
  • 网络安全
  • 电脑知识
  • 服务器
  • 视频教程
  • vbs
  • DOS/BAT
  • hta/htc
  • python
  • perl
  • VBA
  • ColdFusion
  • ruby
  • PowerShell
  • Lua
  • Golang
  • linux shell
您的位置:首页 > 脚本语言 >python > Python+Pika+RabbitMQ环境部署及实现工作队列的实例教程

Python+Pika+RabbitMQ环境部署及实现工作队列的实例教程

作者: 字体:[增加 减小] 来源:互联网

通过本文主要向大家介绍了python pika,rabbitmq消息队列,rabbitmq 删除队列,rabbitmq队列,rabbitmq 清空队列等相关知识,希望对您有所帮助,也希望大家支持linkedu.com www.linkedu.com

rabbitmq中文翻译的话,主要还是mq字母上:Message Queue,即消息队列的意思。前面还有个rabbit单词,就是兔子的意思,和python语言叫python一样,老外还是蛮幽默的。rabbitmq服务类似于mysql、apache服务,只是提供的功能不一样。rabbimq是用来提供发送消息的服务,可以用在不同的应用程序之间进行通信。

安装rabbitmq
先来安装下rabbitmq,在ubuntu 12.04下可以直接通过apt-get安装:

sudo apt-get install rabbitmq-server
</div>

安装好后,rabbitmq服务就已经启动好了。接下来看下python编写Hello World!的实例。实例的内容就是从send.py发送“Hello World!”到rabbitmq,receive.py从rabbitmq接收send.py发送的信息。

2016629175148628.png (288×100)

其中P表示produce,生产者的意思,也可以称为发送者,实例中表现为send.py;C表示consumer,消费者的意思,也可以称为接收者,实例中表现为receive.py;中间红色的表示队列的意思,实例中表现为hello队列。

python使用rabbitmq服务,可以使用现成的类库pika、txAMQP或者py-amqplib,这里选择了pika。

安装pika

安装pika可以使用pip来进行安装,pip是python的软件管理包,如果没有安装,可以通过apt-get安装

sudo apt-get install python-pip
</div>

通过pip安装pika:

sudo pip install pika

</div>

send.py代码

连接到rabbitmq服务器,因为是在本地测试,所以就用localhost就可以了。

connection = pika.BlockingConnection(pika.ConnectionParameters(
        'localhost'))
channel = connection.channel()
</div>

声明消息队列,消息将在这个队列中进行传递。如果将消息发送到不存在的队列,rabbitmq将会自动清除这些消息。

channel.queue_declare(queue='hello')
</div>

发送消息到上面声明的hello队列,其中exchange表示交换器,能精确指定消息应该发送到哪个队列,routing_key设置为队列的名称,body就是发送的内容,具体发送细节暂时先不关注。

channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
</div>

关闭连接

connection.close()
</div>

完整代码

#!/usr/bin/env python
#coding=utf8
import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        'localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='hello')
 
channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
print " [x] Sent 'Hello World!'"
connection.close()
</div>

先来执行下这个程序,执行成功的话,rabbitmqctl应该成功增加了hello队列,并且队列里应该有一条信息,用rabbitmqctl命令来查看下

rabbitmqctl list_queues
</div>

在笔者的电脑上输出如下信息:

2016629175215597.png (193×71)

确实有一个hello队列,并且队列里有一条信息。接下来用receive.py来获取队列里的信息。

receive.py代码

和send.py的前面两个步骤一样,都是要先连接服务器,然后声明消息的队列,这里就不再贴同样代码了。

接收消息更为复杂一些,需要定义一个回调函数来处理,这边的回调函数就是将信息打印出来。

def callback(ch, method, properties, body):
  print "Received %r" % (body,)
</div>

告诉rabbitmq使用callback来接收信息

channel.basic_consume(callback, queue='hello', no_ack=True)
</div>

开始接收信息,并进入阻塞状态,队列里有信息才会调用callback进行处理。按ctrl+c退出。

channel.start_consuming()
</div>

完整代码

#!/usr/bin/env python
#coding=utf8
import pika
 
connection = pika.BlockingConnection(pika.ConnectionParameters(
        'localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='hello')
 
def callback(ch, method, properties, body):
  print " [x] Received %r" % (body,)
 
channel.basic_consume(callback, queue='hello', no_ack=True)
 
print ' [*] Waiting for messages. To exit press CTRL+C'
channel.start_consuming()
</div>

执行程序,就能够接收到队列hello里的消息Hello World!,然后打印在屏幕上。换一个终端,再次执行send.py,可以看到receive.py这边会再次接收到信息。

工作队列示例

1.准备工作(Preparation)

在实例程序中,用new_task.py来模拟任务分配者, worker.py来模拟工作者。

修改send.py,从命令行参数里接收信息,并发送

import sys
 
message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
           routing_key='hello',
           body=message)
print " [x] Sent %r" % (message,)
</div>

修改receive.py的回调函数。

import time
 
def callback(ch, method, properties, body):
  print " [x] Received %r" % (body,)
  time.sleep( body.count('.') )
  print " [x] Done"
</div>

这边先打开两个终端,都运行worker.py,处于监听状态,这边就相当于两个工作者。打开第三个终端,运行new_task.py

$ python new_task.py First message.
$ python new_task.py Second message..
$ python new_task.py Third message...
$ python new_task.py Fourth message....
$ python new_task.py Fifth message.....

</div> 观察worker.py接收到任务,其中一个工作者接收到3个任务 :</div>
$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'First message.'
 [x] Received 'Third message...'
 [x] Received 'Fifth message.....'
</div>

另外一个工作者接收到2个任务 :

$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'Second message..'
 [x] Received 'Fourth message....'
</div>

从上面来看,每个工作者,都会依次分配到任务。那么如果一个工作者,在处理任务的时候挂掉,这个任务就没有完成,应当交由其他工作者处理。所以应当有一种机制,当一个工作者完成任务时,会反馈消息。

2.消息确认(Message acknowledgment)

消息确认就是当工作者完成任务后,会反馈给rabbitmq。修改worker.py中的回调函数:

def callback(ch, method, properties, body):
  print " [x] Received %r" % (body,)
  time.sleep(5)
  print " [x] Done"
  ch.basic_ack(delivery_tag = method.delivery_tag)
</div>

这边停顿5秒,可以方便ctrl+c退出。

去除no_ack=True参数或者设置为False也可以。

channel.basic_consume(callback, queue='hello', no_ack=False)
</div>

用这个代码运行,即使其中一个工作者ctrl+c退出后,正在执行的任务也不会丢失,rabbitmq会将任务重新分配给其他工作者。

3.消息持久化存储(Message durability)

虽然有了消息反馈机制,但是如果rabbitmq自身挂掉的话,那么任务还是会丢失。所以需要将任务持久化存储起来。声明持久化存储:

channel.queue_declare(queue='hello', durable=True)
</div>

但是这个程序会执行错误,因为hello这个队列已经存在,并且是非持久化的,rabbitmq不允许使用不同的参数来重新定义存在的队列。重新定义一个队列:

channel.queue_declare(queue='ta
  


 

您可能想查找下面的文章:

  • Python+Pika+RabbitMQ环境部署及实现工作队列的实例教程

相关文章

  • Python设计模式编程中Adapter适配器模式的使用实例
  • 浅谈Python的Django框架中的缓存控制
  • 利用Python生成文件md5校验值函数的方法
  • python基础教程之基本内置数据类型介绍
  • python3简单实现微信爬虫
  • Python中asyncore异步模块的用法及实现httpclient的实例
  • Python爬虫代理IP池实现方法
  • 使用PDB模式调试Python程序介绍
  • Python制作简易注册登录系统
  • Python中List.count()方法的使用教程

文章分类

  • vbs
  • DOS/BAT
  • hta/htc
  • python
  • perl
  • VBA
  • ColdFusion
  • ruby
  • PowerShell
  • Lua
  • Golang
  • linux shell

最近更新的内容

    • 利用python生成一个导出数据库的bat脚本文件的方法
    • 删除目录下相同文件的python代码(逐级优化)
    • Python采用raw_input读取输入值的方法
    • python3.5 + PyQt5 +Eric6 实现的一个计算器代码
    • 详解Python中 __get__和__getattr__和__getattribute__的区别
    • Python操作列表之List.insert()方法的使用
    • Python内置函数的用法实例教程
    • python获取指定路径下所有指定后缀文件的方法
    • python回溯法实现数组全排列输出实例分析
    • python中异常报错处理方法汇总

关于我们 - 联系我们 - 免责声明 - 网站地图

©2020-2025 All Rights Reserved. linkedu.com 版权所有