热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

Python操作RabbitMQ

RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统。他遵循MozillaPublicLicense开源协议。MQ全称为MessageQueue,消息

RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统。他遵循Mozilla Public License开源协议。

MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信,而不是通过直接调用彼此来通信,直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

RabbitMQ安装

1

2

3

4

5

6

7

8

安装配置epel源

   $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm

 

安装erlang

   $ yum -y install erlang

 

安装RabbitMQ

   $ yum -y install rabbitmq-server

注意:service rabbitmq-server start/stop

安装API

1

2

3

4

5

6

7

pip install pika

or

easy_install pika

or

源码

 

https://pypi.python.org/pypi/pika

使用API操作RabbitMQ

基于Queue实现生产者消费者模型

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import Queue
import threadingmessage = Queue.Queue(10)def producer(i):while True:message.put(i)def consumer(i):while True:msg = message.get()for i in range(12):t = threading.Thread(target=producer, args=(i,))t.start()for i in range(10):t = threading.Thread(target=consumer, args=(i,))t.start()

对于RabbitMQ来说,生产和消费不再针对内存里的一个Queue对象,而是某台服务器上的RabbitMQ Server实现的消息队列。

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

#!/usr/bin/env python

import pika

 

# ######################### 生产者 #########################

 

connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='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()


1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

#!/usr/bin/env python

import pika

 

# ########################## 消费者 ##########################

 

connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='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()

1、acknowledgment 消息不丢失

no-ack = False,如果消费者遇到情况(its channel is closed, connection is closed, or TCP connection is lost)挂掉了,那么,RabbitMQ会重新将该任务添加到队列中。

import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()channel.queue_declare(queue='hello')def callback(ch, method, properties, body):print(" [x] Received %r" % body)import timetime.sleep(10)print 'ok'ch.basic_ack(delivery_tag = method.delivery_tag)channel.basic_consume(callback,queue='hello',no_ack=False)print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

2、durable   消息不丢失

#!/usr/bin/env python
import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()# make message persistent
channel.queue_declare(queue='hello', durable=True)channel.basic_publish(exchange='',routing_key='hello',body='Hello World!',properties=pika.BasicProperties(delivery_mode=2, # make message persistent))
print(" [x] Sent 'Hello World!'")
connection.close()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()# make message persistent
channel.queue_declare(queue='hello', durable=True)def callback(ch, method, properties, body):print(" [x] Received %r" % body)import timetime.sleep(10)print 'ok'ch.basic_ack(delivery_tag = method.delivery_tag)channel.basic_consume(callback,queue='hello',no_ack=False)print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

3、消息获取顺序

默认消息队列里的数据是按照顺序被消费者拿走,例如:消费者1 去队列中获取 奇数 序列的任务,消费者1去队列中获取 偶数 序列的任务。

channel.basic_qos(prefetch_count=1) 表示谁来谁取,不再按照奇偶数排列

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
channel = connection.channel()# make message persistent
channel.queue_declare(queue='hello')def callback(ch, method, properties, body):print(" [x] Received %r" % body)import timetime.sleep(10)print 'ok'ch.basic_ack(delivery_tag = method.delivery_tag)channel.basic_qos(prefetch_count=1)channel.basic_consume(callback,queue='hello',no_ack=False)print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

4、发布订阅

发布订阅和简单的消息队列区别在于,发布订阅会将消息发送给所有的订阅者,而消息队列中的数据被消费一次便消失。所以,RabbitMQ实现发布和订阅时,会为每一个订阅者创建一个队列,而发布者发布消息时,会将消息放置在所有相关队列中。

 exchange type = fanout

#!/usr/bin/env python
import pika
import sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()channel.exchange_declare(exchange='logs',type='fanout')message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs',routing_key='',body=message)
print(" [x] Sent %r" % message)
connection.close()

#!/usr/bin/env python
import pikaconnection = 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.queuechannel.queue_bind(exchange='logs',queue=queue_name)print(' [*] Waiting for logs. To exit press CTRL+C')def callback(ch, method, properties, body):print(" [x] %r" % body)channel.basic_consume(callback,queue=queue_name,no_ack=True)channel.start_consuming()

5、关键字发送

 exchange type = direct

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

#!/usr/bin/env python
import pika
import sysconnection = 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.queueseverities = sys.argv[1:]
if not severities:sys.stderr.write("Usage: %s [info] [warning] [error]\n" % 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()

#!/usr/bin/env python
import pika
import sysconnection = 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()

6、模糊匹配

 exchange type = topic

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

  • # 表示可以匹配 0 个 或 多个 单词
  • *  表示只能匹配 一个 单词

1

2

3

发送者路由值              队列中

old.boy.python          old.*  -- 不匹配

old.boy.python          old.#  -- 匹配

#!/usr/bin/env python
import pika
import sysconnection = 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.queuebinding_keys = sys.argv[1:]
if not binding_keys:sys.stderr.write("Usage: %s [binding_key]...\n" % 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()

#!/usr/bin/env python
import pika
import sysconnection = pika.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()

注意:

sudo rabbitmqctl add_user wupeiqi 123
# 设置用户为administrator角色
sudo rabbitmqctl set_user_tags wupeiqi administrator
# 设置权限
sudo rabbitmqctl set_permissions -p "/" root ".*" ".*" ".*"# 然后重启rabbiMQ服务
sudo /etc/init.d/rabbitmq-server restart# 然后可以使用刚才的用户远程连接rabbitmq server了。------------------------------
credentials = pika.PlainCredentials("wupeiqi","123")connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.14.47',credentials=credentials))

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pika
from pika.adapters.blocking_connection import BlockingChannelcredentials = pika.PlainCredentials("root", "123")conn = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.20', credentials=credentials))
# 超时时间
conn.add_timeout(5, lambda: channel.stop_consuming())channel = conn.channel()channel.queue_declare(queue='hello')def callback(ch, method, properties, body):print(" [x] Received %r" % body)channel.stop_consuming()channel.basic_consume(callback,queue='hello',no_ack=True)print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

https://www.cnblogs.com/zhangxinqi/p/8460768.html


推荐阅读
  • 大类|电阻器_使用Requests、Etree、BeautifulSoup、Pandas和Path库进行数据抓取与处理 | 将指定区域内容保存为HTML和Excel格式
    大类|电阻器_使用Requests、Etree、BeautifulSoup、Pandas和Path库进行数据抓取与处理 | 将指定区域内容保存为HTML和Excel格式 ... [详细]
  • 本文详细探讨了使用Python3编写爬虫时如何应对网站的反爬虫机制,通过实例讲解了如何模拟浏览器访问,帮助读者更好地理解和应用相关技术。 ... [详细]
  • python模块之正则
    re模块可以读懂你写的正则表达式根据你写的表达式去执行任务用re去操作正则正则表达式使用一些规则来检测一些字符串是否符合个人要求,从一段字符串中找到符合要求的内容。在 ... [详细]
  • RocketMQ在秒杀时的应用
    目录一、RocketMQ是什么二、broker和nameserver2.1Broker2.2NameServer三、MQ在秒杀场景下的应用3.1利用MQ进行异步操作3. ... [详细]
  • 使用ArcGIS for Java和Flex浏览自定义ArcGIS Server 9.3地图
    本文介绍了如何在Flex应用程序中实现浏览自定义ArcGIS Server 9.3发布的地图。这是一个基本的入门示例,适用于初学者。 ... [详细]
  • 在Windows系统中安装TensorFlow GPU版的详细指南与常见问题解决
    在Windows系统中安装TensorFlow GPU版是许多深度学习初学者面临的挑战。本文详细介绍了安装过程中的每一个步骤,并针对常见的问题提供了有效的解决方案。通过本文的指导,读者可以顺利地完成安装并避免常见的陷阱。 ... [详细]
  • Python 程序转换为 EXE 文件:详细解析 .py 脚本打包成独立可执行文件的方法与技巧
    在开发了几个简单的爬虫 Python 程序后,我决定将其封装成独立的可执行文件以便于分发和使用。为了实现这一目标,首先需要解决的是如何将 Python 脚本转换为 EXE 文件。在这个过程中,我选择了 Qt 作为 GUI 框架,因为之前对此并不熟悉,希望通过这个项目进一步学习和掌握 Qt 的基本用法。本文将详细介绍从 .py 脚本到 EXE 文件的整个过程,包括所需工具、具体步骤以及常见问题的解决方案。 ... [详细]
  • 每日前端实战:148# 视频教程展示纯 CSS 实现按钮两侧滑入装饰元素的悬停效果
    通过点击页面右侧的“预览”按钮,您可以直接在当前页面查看效果,或点击链接进入全屏预览模式。该视频教程展示了如何使用纯 CSS 实现按钮两侧滑入装饰元素的悬停效果。视频内容具有互动性,观众可以实时调整代码并观察变化。访问以下链接体验完整效果:https://codepen.io/comehope/pen/yRyOZr。 ... [详细]
  • 本文探讨了利用Python实现高效语音识别技术的方法。通过使用先进的语音处理库和算法,本文详细介绍了如何构建一个准确且高效的语音识别系统。提供的代码示例和实验结果展示了该方法在实际应用中的优越性能。相关文件可从以下链接下载:链接:https://pan.baidu.com/s/1RWNVHuXMQleOrEi5vig_bQ,提取码:p57s。 ... [详细]
  • 在重新安装Ubuntu并配置Django和PyCharm后,忘记测试MySQL连接,导致在后续配置过程中遇到错误:ERROR 2003 (HY000) - 无法连接到本地服务器 ‘127.0.0.1’ (111)。本文将详细介绍该错误的原因及解决步骤,帮助用户快速恢复MySQL服务的正常运行。我们将从检查网络配置、验证MySQL服务状态、配置防火墙规则等方面入手,提供全面的故障排除指南。 ... [详细]
  • 通过使用七牛云存储服务,本文详细介绍了如何将本地图片高效上传至云端,并实现了内容的便捷管理。借助七牛云的 Python SDK,文章提供了从认证到文件上传的具体代码示例,包括导入必要的库、生成上传凭证以及处理文件路径等关键步骤。此外,还探讨了如何利用七牛云的 URL 安全编码功能,确保数据传输的安全性和可靠性。 ... [详细]
  • HTML 页面中调用 JavaScript 函数生成随机数值并自动展示
    在HTML页面中,通过调用JavaScript函数生成随机数值,并将其自动展示在页面上。具体实现包括构建HTML页面结构,定义JavaScript函数以生成随机数,以及在页面加载时自动调用该函数并将结果呈现给用户。 ... [详细]
  • 在使用 `requests` 库进行 HTTP 请求时,如果遇到 `requests.exceptions.SSLError: HTTPSConnectionPool` 错误,通常是因为 SSL 证书验证失败。解决这一问题的方法包括:检查目标网站的 SSL 证书是否有效、更新本地的 CA 证书库、禁用 SSL 验证(不推荐用于生产环境)或使用自定义的 SSL 上下文。此外,确保 `requests` 库和相关依赖项已更新到最新版本,以避免潜在的安全漏洞。 ... [详细]
  • 基于 Bottle 框架构建的幽默应用 —— Python 实践 ... [详细]
  • 利用树莓派畅享落网电台音乐体验
    最近重新拾起了闲置已久的树莓派,这台小巧的开发板已经沉寂了半年多。上个月闲暇时间较多,我决定将其重新启用。恰逢落网电台进行了改版,回忆起之前在树莓派论坛上看到有人用它来播放豆瓣音乐,便萌生了同样的想法。通过一番调试,终于实现了在树莓派上流畅播放落网电台音乐的功能,带来了全新的音乐享受体验。 ... [详细]
author-avatar
淋殁
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有