z安装pika模块：pip install pika
一个最简单的生产者/消费者：
生产者，send.py:
import pika class SenderClient(object): def __init__(self, username, passwd, host="127.0.0.1", port=5672, queuename='eeg', exchange='eeg', routing_key='eeg'): self.__host = host self.__port = port self.__username = username self.__passwd = passwd self.__queue = queuename self.__exchange = exchange self.__rout_key = routing_key def connect_mq(self): """ 连接mq :return: """ try: # 创建一个连接对象 if not hasattr(self, 'connection') or self.connection.is_closed: # 添加用户名和密码 credentials = pika.PlainCredentials(self.__username, self.__passwd) # 配置连接参数 parameters = pika.ConnectionParameters(host=self.__host, port=self.__port, credentials=credentials) self.connection = pika.BlockingConnection(parameters) except Exception as e: print(e) def channel_mq(self): """ # 创建一个信道 # 声明队列和交换机和绑定 :return: """ if not hasattr(self, "connection"): self.connect_mq() if not hasattr(self, 'channel') or self.channel.is_closed: self.channel = self.connection.channel() # 声明一个队列,durable参数声明队列持久化 self.channel.queue_declare(queue=self.__queue, durable=True) self.channel.exchange_declare(exchange=self.__exchange, durable=True) # 交换机和队列绑定:生产者发布消息时无需绑定，消费者消费消息时需要绑定 self.channel.queue_bind(exchange=self.__exchange, queue=self.__queue, routing_key=self.__rout_key) def open_data(self, filename): """ # 打开一个数据文件 :return: """ with open(filename, 'r', encoding='utf-8') as f: data = f.read() return data def send_data(self, data:str): """ # 发送数据 :param channel: :return: """ self.channel_mq() # 使用默认交换机投递消息,返回TRUE或False self.channel.basic_publish(exchange=self.__exchange, routing_key=self.__rout_key, body=data, properties=pika.BasicProperties(delivery_mode=2)) def close_connect(self): """ # 关闭tcp连接 :return: """ self.connection.close() def close_channel(self, channel): """ # 关闭信道 :param channel: :return: """ if not hasattr(self, 'channel'): raise ValueError("the object of SenderClient has not attr of channel.") self.channel.close()
消费者，receive.py
from .send import SenderClient class ReceiverClient(SenderClient): """ 接收rabbitmq消息的消费者 """ def run(self,queuename=None): """ 从mq中接收消息。 :return: """ if not hasattr(self, 'channel') or self.channel.is_closed: self.channel_mq() if not queuename: queuename=self.__queue # 订阅消息 self.channel.basic_consume(self.callback, queue=queuename, no_ack=False) # 循环等待 self.channel.start_consuming() def callback(self,ch, method, properties, body): """ # 接收消息处理函数 :param ch: :param method: :param properties: :param body: :return: """ print('接收成功！') # 发送确认 ch.basic_ack(delivery_tag=method.delivery_tag)
分析方法:
创建一个连接connection:
# 添加用户名和密码 credentials = pika.PlainCredentials(self.__username, self.__passwd) # 配置连接参数 parameters = pika.ConnectionParameters(host=self.__host, credentials=credentials) # 创建一个连接对象 connection = pika.BlockingConnection(parameters)
- pika.PlainCredentials:一个凭据类
# 该类传递的参数 def __init__(self, username, # 用户名 password, # 密码 erase_on_connect=False): # 是否保存凭据在内存中，如果参数是True，那么该类会在连接完成后删除
- pika.ConnectionParameters：TCP连接的参数类
def __init__(self, host=_DEFAULT, # 默认的ip127.0.0.1 port=_DEFAULT, # 端口5672 virtual_host=_DEFAULT, # 默认的虚拟主机/ credentials=_DEFAULT, # 默认的凭据 user:guest passwd:guest # 以下的参数含义可以在配置文件中找到，一般不需要在这里配置 channel_max=_DEFAULT, frame_max=_DEFAULT, heartbeat=_DEFAULT, ssl=_DEFAULT, ssl_options=_DEFAULT, connection_attempts=_DEFAULT, retry_delay=_DEFAULT, socket_timeout=_DEFAULT, locale=_DEFAULT, backpressure_detection=_DEFAULT, blocked_connection_timeout=_DEFAULT, client_properties=_DEFAULT, tcp_options=_DEFAULT, **kwargs):
- pika.BlockingConnection：创建一个连接类
def __init__(self, parameters=None, _impl_class=None): # 传递一个参数类就可以了，_impl_class只用于测试
-
创建一个channel
# 创建一个信道 channel = connection.channel() def channel(self, channel_number=None): # 指定通道的编号，一般让rabbitmq自动去管理
-
声明一个队列queue
def queue_declare(self, queue='', # 队列的名字，默认为空，此时将自动创建一个名字， passive=False, # 检查一下队列是否存在，如果该参数为True，该方法判断队列存在否，不会声明队列；存在返回queue的状态； durable=False, # 队列持久化参数，默认不持久化 exclusive=False, # 设置独享队列，该队列只被当前的connection使用，如果该tcp关闭了，队列会被删除 auto_delete=False,# 当最后一个消费者退订后自动删除，默认不开启 arguments=None) # 一个字典，用于队列传递额外的参数
-
声明一个交换机exchange
def exchange_declare(self, exchange=None, # 交换机的名字，为空则自动创建一个名字 exchange_type='direct', # 默认交换机类型为direct passive=False, # 检查交换机是否存在，存在返回状态信息，不存在返回404错误 durable=False, # 设置是否持久化 auto_delete=False, # 最后一个队列解绑则删除 internal=False, # 是否设置为值接收从其他交换机发送过来的消息，不接收生产者的消息 arguments=None): # 一个字典，用于传递额外的参数
-
声明一个绑定：
- channel.queue_bind：队列和交换机绑定
def queue_bind(self, queue, # 队列的名字 exchange, # 交换机的名字 routing_key=None, # 路由键规则，当为None时，默认使用queue的名字作为路由键规则 arguments=None): # 一个字典，传递额外的参数 # 返回绑定的状态信息
- channel.exchange_bind:交换机之间的绑定
def exchange_bind(self, destination=None, # 目的交换机的名字 source=None, # 源交换机的名字 routing_key='', # 路由键规则，当为None时，默认使用queue的名字作为路由键规则 arguments=None): # 一个字典，传递额外的参数
- channel.queue_bind：队列和交换机绑定
-
投递一条消息
def basic_publish(self, exchange, # 交换机的名字 routing_key, # 路由键，吐过交换机是扇形交换机，可以随便写 body, # 消息主体 properties=None, # 消息的属性 mandatory=False, # 是否设置消息托管 immediate=False) # 是否消息实时同步确认，一般和confirm模式配合使用 # properties属性有一个专门的类来设置 pika.BasicProperties： def __init__(self, content_type=None, content_encoding=None, headers=None, delivery_mode=None, priority=None, correlation_id=None, reply_to=None, expiration=None, message_id=None, timestamp=None, type=None, user_id=None, app_id=None, cluster_id=None): self.content_type = content_type # 消息的类型，如text/html,json等 self.content_encoding = content_encoding # 消息的编码，如gbk，utf-8等 self.headers = headers # 消息头，可以和头交换机约定规则 self.delivery_mode = delivery_mode # 消息持久化，2表示持久化， self.priority = priority # 消息的优先权 self.correlation_id = correlation_id self.reply_to = reply_to self.expiration = expiration # 消息的有效期 self.message_id = message_id # 消息iD，自动管理 self.timestamp = timestamp # 消息的时间戳 self.type = type self.user_id = user_id self.app_id = app_id # 发布应用的ID self.cluster_id = cluster_id
-
订阅消息
-
方式一：客户端主动推送消息：channel.basic_consume+channel.start_consuming
注意：推送是异步的，也就是说一次可能推送多条消息，提高性能。
- start_consuming ：开始阻塞等待消息
阻塞等待消息是有时间限制的，超过一定时间内如果没有新的消息推送过来会强制关闭连接，因此如果需要全时段等待的话需要监听该连接；
def basic_consume(self, # 启动队列消费者，告诉服务端开启一个消费者 consumer_callback, # 消费者回调函数 queue, # 队列名称 no_ack=False, # 发送确认，默认开启消息确认模式，为True是关闭消息确认；如果回调函数中不发送消息确认，消息会一直存在队列中，等待推送给新连接的消费者 exclusive=False, # 设置独享消费者，不允许其他消费者订阅该队列 consumer_tag=None, # 消费者标签，如果不指定，系统自动生成 arguments=None): # 字典，额外的参数 consumer_callback：回调函数 def consumer_callback(channel, # 信道 method, # 一个交付的deliver对象，用来通知客户端消息 properties, # 消息的属性，就是消息在发送时定义的属性 body) # 消息的主题，二进制格式 method:spec.Basic.Deliver:交付对象的属性 def __init__(self, consumer_tag=None, delivery_tag=None, redelivered=False, exchange=None, routing_key=None): self.consumer_tag = consumer_tag # 消费者标签，用来标记是哪一个消费者 self.delivery_tag = delivery_tag # 交付标签，用来发送消息确认和标记这是推送给该消费者的第几条消息 self.redelivered = redelivered # bool类型，若果为False表示这是消息第一次被推送，如果是True，表示这是一条被重复推送的消息 self.exchange = exchange # 该消息是从哪个交换机上来的 self.routing_key = routing_key # 该消息的路由键是什么 # 函数中，可以通过method.等获取相应的属性 method.consumer_tag/method.delivery_tag/method.redelivered
-
方式二:客户端主动获取消息：channel.basic_get：同步获取消息，性能比方式一低，比较少使用
def basic_get(self, queue=None, # 队列名称 no_ack=False): # 是否需要开启确认模式 return method,properties,body # 需要主动进行消息确认，basic_ack
-
-
取消订阅
channel.basic_cancel():取消某个消费者订阅 channel.stop_consuming():取消所有的订阅 -
订阅消息确认
def basic_ack(self, delivery_tag=0, # 消息的标记，int类型，一般将回调函数consumer_callback中获取的交付标记放到这个位置 multiple=False): # Flase表示确认单个消息，为True表示确认多个消息
-
订阅消息拒绝：
-
basic_nack()
-
basic_reject：另一个方法，只能拒绝单个消息，没有multiple参数；
def basic_nack(self, delivery_tag=None, # 交付这标记，和basic_ack一样 multiple=False, # Flase表示拒绝单个消息，为True表示拒绝多个消息 requeue=True) # True表示拒绝了消息后重新放回队列，False表示丢弃消息
-
-
公平调度
- basic_qos：一般在信道声明的时候使用，确定该信道的预取数，提高性能 def basic_qos(self, prefetch_size=0, # 设置预取数据包的大小 prefetch_count=0, # 设置预取的数量，如果为0则不预取，消费者处理越快，可以将这个这设置的越高 all_channels=False) # 是否将所有的信道都设置上述参数
-
投递消息确认机制
-
AMQP协议的事务方式，不推荐使用：tx_select，tx_commit,tx_rollback，方式是十分消耗rabbitmq的性能的，一般不推荐使用
# 开启一个事务,在提交事务之前必须先执行此方法 channel.tx_select() # 提交一个事务 channel.tx_select() # 捕捉到异常就使用回滚 channel.tx_rollback()
-
import pika if __name__ == "__main__": # 配置连接参数 parameters = pika.ConnectionParameters(host=self.__host) # 创建一个连接对象 connection = pika.BlockingConnection(parameters) # 创建一个信道 channel = connection.channel() # 声明队列 channel.queue_declare(queue='test',durable=True) # 开启事务 channel.tx_select() try: channel.basic_publish(exchange='', routing_key='test', body='hello-world') result = 1/0 channel.tx_commit() except: channel.tx_rollback() # tx_select和tx_commit之间的所有的操作都是事务的一部分
-
confirm 模式：channel.confirm_delivery()
-
rabbbitmq自带confirm模式，生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者（包含消息的唯一ID）。
import pika if __name__ == "__main__": # 配置连接参数 parameters = pika.ConnectionParameters(host=HOST) # 创建一个连接对象 connection = pika.BlockingConnection(parameters) # 创建一个信道 channel = connection.channel() # 声明队列 channel.queue_declare(queue='test', durable=True) # 打开通道的确认模式 channel.confirm_delivery() for i in range(3): result = channel.basic_publish(exchange='', routing_key='test', body='hello-world') if result: break
说明:
-
当确认模式没有打开时，即使队列和交换机不存在，投递消息返回的都是True;
-
当确认模式打开时，投递失败会返回False，成功返回True，如果队列不存在，交换机会叫消息丢掉，但不会通知生产者；如果交换机不存在，会报错；
- 同一个信道，确认模式和事务模式只能存在一个，不能同时启用，否则报错；
-
-
-
- 交换机绑定：channel.exchange_bind()
import pika if __name__ == "__main__": # 添加用户名和密码 credentials = pika.PlainCredentials(USERNAME, PASSWD) # 配置连接参数 parameters = pika.ConnectionParameters(host=HOST, credentials=credentials) # 创建一个连接对象 connection = pika.BlockingConnection(parameters) # 创建一个信道 channel = connection.channel() # 声明队列 channel.queue_declare(queue='test1', durable=True) channel.queue_declare(queue='test2', durable=True) # 声明交换机 channel.exchange_declare('myname') channel.exchange_declare('youname') # 交换机绑定 channel.exchange_bind(destination='youname',source='myname',routing_key='ourheart') # 队列绑定 channel.queue_bind(queue='test1',exchange='myname',routing_key='ourheart') channel.queue_bind(queue='test2', exchange='youname',routing_key='ourheart') channel.basic_publish(exchange='myname', routing_key='ourheart', body='hello-world')
说明
-
交换机相互绑定后，如果他们之间连接的桥routing_key是相同的，向源交换机投递消息，数据可以到达相同路由键的所有的队列；向目的交换机投递消息，消息不能到达源交换机；
-
交换机设置了internal了True参数后，该交换机不能再接收到生产者发送的消息，但可以得到源交换机发送的消息；
-
-
其他方法
exchange_delete: 删除交换机 queue_delete: 删除队列 queue_purge: 清除指定队列的所有的消息 queue_unbind: 队列和交换机解除绑定 basic.cancel: 清除消费者
参考