python中RabbitMQ的使用(安装和简单教程)
1,简介
RabbitMQ是一个由erlang开发的AMQP(Advanced Message Queue )的开源实现的产品,RabbitMQ是一个消息代理,从“生产者”接收消息并传递消息至“消费者”,期间可根据规则路由、缓存、持久化消息。“生产者”也即message发送者以下简称P,相对应的“消费者”乃message接收者以下简称C,message通过queue由P到C,queue存在于RabbitMQ,可存储尽可能多的message,多个P可向同一queue发送message,多个C可从同一个queue接收message
1.1关键词说明:
Broker:消息队列服务器实体。 Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列。 Queue:消息队列载体,每个消息都会被投入到一个或多个队列。 Binding:绑定,它的作用就是把exchange和queue按照路由规则绑定起来。 Routing Key:路由关键字,exchange根据这个关键字进行消息投递。 vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离。 producer:消息生产者,就是投递消息的程序。 consumer:消息消费者,就是接受消息的程序。 channel:消息通道,在客户端的每个连接里,可建立多个channel,每个channel代表一个会话任务。
1.2消息队列运行机制:
(1)客户端连接到消息队列服务器,打开一个channel。 (2)客户端声明一个exchange,并设置相关属性。 (3)客户端声明一个queue,并设置相关属性。 (4)客户端使用routing key,在exchange和queue之间建立好绑定关系。 (5)客户端投递消息到exchange。 (6)exchange接收到消息后,就根据消息的key和已经设置的binding,将消息投递到一个或多个队列里。
注:在声明一个队列后,如果将其持久化,则下次不需要进行声明,因为该队列已经在rabbitMQ中了!!!
例如下面的例子中都为首次声明一个队列!!!
1.3exchange类型:
1.Direct交换机 特点:依据key进行投递 例如绑定时设置了routing key为”hello”,那么客户端提交的消息,只有设置了key为”hello”的才会投递到队列。
2.Topic交换机 特点:对key模式匹配后进行投递,符号”#”匹配一个或多个词,符号” * ”匹配一个词 例如”abc.#”匹配”abc.def.ghi”,”abc.*”只匹配”abc.def”。
3.Fanout交换机 特点:不需要key,采取广播模式,一个消息进来时,投递到与该交换机绑定的所有队列
1.4消息确认
当客户端从队列中取出消息之后,可能需要一段时间才能处理完成,如果在这个过程中,客户端出错了,异常退出了,而数据还没有处理完成,那么非常不幸,这段数据就丢失了,因为rabbitmq默认会把此消息标记为已完成,然后从队列中移除, 消息确认是客户端从rabbitmq中取出消息,并处理完成之后,会发送一个ack告诉rabbitmq,消息处理完成,当rabbitmq收到客户端的获取消息请求之后,或标记为处理中,当再次收到ack之后,才会标记为已完成,然后从队列中删除。当rabbitmq检测到客户端和自己断开链接之后,还没收到ack,则会重新将消息放回消息队列,交给下一个客户端处理,保证消息不丢失,也就是说,RabbitMQ给了客户端足够长的时间来做数据处理。
在客户端使用no_ack来标记是否需要发送ack,默认是False,开启状态
2.构建环境
2.1在windows环境下安装rabbitmq,教程如下:
http://jingyan.baidu.com/article/a17d5285173ce68098c8f2e5.html
2.2安装pika模块
python使用rabbitmq服务,可以使用现成的类库pika、txAMQP或者py-amqplib,这里选择了pika。
在命令行中直接使用pip命令:
pip install pika
3.示例测试
实例的内容就是从send.py发送消息到rabbitmq,receive.py从rabbitmq接收send.py发送的信息。
P表示produce,生产者的意思,也可以称为发送者,实例中表现为send.py; C表示consumer,消费者的意思,也可以称为接收者,实例中表现为receive.py; 中间红色的表示队列的意思,实例中表现为hello队列。
send.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import pika
import random
queue = 'oss.url_test'#队列名
routing_key = 'url_test'
exchange = 'oss_test'
# 新建连接,rabbitmq安装在本地则hostname为'localhost'
hostname = '192.168.8.190'
port=5672
credentials = pika.PlainCredentials(username='',password='')
parameters = pika.ConnectionParameters(host=hostname,port=port,credentials=credentials)
connection = pika.BlockingConnection(parameters=parameters)
# 创建通道
channel = self.connection.channel()
# 创建broker
channel.exchange_declare(exchange=exchange, exchange_type='direct', durable=True)
# 声明一个队列,生产者和消费者都要声明一个相同的队列,用来防止万一某一方挂了,另一方能正常运行
channel.queue_declare(queue=queue, durable=True)
# 把队列和中间人绑定
channel.queue_bind(exchange=exchange, queue=queue, routing_key=routing_key)
# 交换机; 队列名,写明将消息发往哪个队列; 消息内容
# routing_key在使用匿名交换机的时候才需要指定,表示发送到哪个队列,注意当未定义exchange时,routing_key需和queue的值保持一致
channel.basic_publish(exchange=exchange, routing_key=routing_key, body=json.dumps(data))
#关闭连接
connection.close()
receive.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import pika
hostname = '192.168.8.190'
port=5672
credentials = pika.PlainCredentials(username='',password='')
parameters = pika.ConnectionParameters(host=hostname,port=port,credentials=credentials)
connection = pika.BlockingConnection(parameters=parameters)
# 创建通道
channel = connection.channel()
channel.queue_declare(queue=queue,durable=True)
def callback(ch, method, properties, body):
print " [x] Received %r" % (body,)
ch.basic_ack(delivery_tag=method.delivery_tag) # 发送ack消息
#添加不按顺序分配消息的参数,可有可无
# channel.basic_qos(prefetch_count=1)
# 告诉rabbitmq使用callback来接收信息
channel.basic_consume(callback, queue=queue, no_ack=False)#no_ack来标记是否需要发送ack,默认是False,开启状态
# 开始接收信息,并进入阻塞状态,队列里有信息才会调用callback进行处理,按ctrl+c退出
print ' [*] Waiting for messages. To exit press CTRL+C'
channel.start_consuming()
我们先运行send.py发送消息:
再运行receive.py接收消息:
4消息持久化
消息持久化 消息确认机制使得客户端在崩溃的时候,服务端消息不丢失,但是如果rabbitmq奔溃了呢?该如何保证队列中的消息不丢失? 此就需要product在往队列中push消息的时候,告诉rabbitmq,此队列中的消息需要持久化,用到的参数:durable=True,再次强调,Producer和client都应该去创建这个queue,尽管只有一个地方的创建是真正起作用的:
channel.basic_publish(exchange='',
routing_key="test",
body=message,
properties=pika.BasicProperties(
delivery_mode = 2, # make message persistent
))
具体代码:
- product端:
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='127.0.0.1', port=5672, )) #定义连接池
channel = connection.channel() #声明队列以向其发送消息消息
channel.queue_declare(queue='test_persistent',durable=True)
for i in range(10):
channel.basic_publish(exchange='', routing_key='test_persistent', body=str(i),properties=pika.BasicProperties(delivery_mode=2))
print('send success msg[%s] to rabbitmq' %i)
connection.close() #关闭连接
- client端:
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='127.0.0.1', port=5672))
channel = connection.channel()
channel.queue_declare(queue='test_persistent',durable=True)
def callback(ch, method, properties, body):
'''回调函数,处理从rabbitmq中取出的消息'''
print(" [x] Received %r" % body)
#time.sleep(5)
ch.basic_ack(delivery_tag = method.delivery_tag) #发送ack消息
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,queue='test_persistent',no_ack=False)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() #开始监听 接受消息
注意:client端也需配置durable=True,否则将报下面错误:
pika.exceptions.ChannelClosed: (406, "PRECONDITION_FAILED - parameters for queue 'test_persistent' in vhost '/' not equivalent")
配置完之后,发现product往rabbitmq端push消息之后,重启rabbitmq,消息依然存在
文章参考: https://www.cnblogs.com/pycode/p/rabbitmq.html https://www.cnblogs.com/wt11/p/5970297.html
- Golang写的并行排序算法
- Go中调用dll示例
- python 序列化数据:pickle与json ,dumps与loads
- golang继承,和多态
- python 利用random生成验证码与MD5码加密过程
- Java内部类的继承
- Java继承类中static成员函数的重写
- Search for a range寻找上下界-Leetcode
- Basic Calculator 基本计算器-Leetcode
- python yield函数深入浅出理解
- 十分钟搞定 Tensorflow 服务
- datapump跨平台升级迁移的总结 (r8笔记第77天)
- Java中isAssignableFrom()方法与instanceof()方法用法
- 与Ajax同样重要的jQuery(1)
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- SQL-查询各工程项目使用所提供零件最多的供应商
- 浙大版《C语言程序设计(第3版)》题目集 习题8-3 数组循环右移
- SwiftUI:控制图像插值
- SQL-查询比p6零件供应数量都高的零件
- 浙大版《C语言程序设计(第3版)》题目集 习题8-5 使用函数实现字符串部分复制
- 最小生成树-Magicpig密室出逃(Kruskal+并查集)
- 浙大版《C语言程序设计(第3版)》题目集 习题8-6 删除字符
- SQL-显示供应商供应零件的汇总列表(with rollup+coalesce)
- 浙大版《C语言程序设计(第3版)》题目集 习题8-8 判断回文字符串
- 双向广搜-HDU1401 Solitaire
- TCP三次握手和四次挥手以及11种状态
- 浙大版《C语言程序设计(第3版)》题目集 题8-9 分类统计各类字符个数
- 迭代加深搜索-POJ 3134 Power Calculus
- 浙大版《C语言程序设计(第3版)》题目集 习题9-2 计算两个复数之积
- 浙大版《C语言程序设计(第3版)》题目集 习题9-6 按等级统计学生成绩