RabbitMQ 消息队列

PY
threading Queue
进程Queue 父进程与子进程，或同一父进程下的多个子进程进行交互
缺点：两个不同Python文件不能通过上面两个Queue进行交互

erlong
基于这个语言创建的一种中间商
win中需要先安装erlong才能使用
rabbitmq_server start

安装 Python module

pip install pika

or

easy_install pika

or
源码

rabbit 默认端口15672
查看当前时刻的队列数
rabbitmqctl.bat list_queue

exchange
在定义的时候就是有类型的，决定到底哪些queue符合条件，可以接受消息
fanout：所有bind到此exchange的queue都可以收到消息
direct：通过routingkey和exchange决定唯一的queue可以接受消息
topic： 所有符合routingkey（此时可以是一个表达式）的routingkey所bind的queue都可以接受消息
表达式符号说明：
# 代表一个或多个字符 * 代表任何字符

RPC
remote procedure call 双向传输，指令<-------->指令执行结果
实现方法： 创建两个队列，一个队列收指令，一个队列发送执行结果

用rabbitmq实现简单的生产者消费者模型

1） rabbit_producer.py

# Author : Xuefengimport pikaconnection = pika.BlockingConnection(pika.Connection.Parameters( "localhost"))# statement a channelchannel = connection.channel()# create the queue, the name of queue is "hello"# durable=True can make the queue be exist, although the service have stopped before.channel.queue_declare(queue="hello", durable=True)# n RabbitMQ a message can never be sent directly to queue,it always need to go throughchannel.basic_publish(exchange = " ", routing_key = "hello", body = "Hello world!", properties = pika.BasicPropreties( delivery_mode=2, # make the message persistence ) )print("[x] sent 'Hello world!'")connection.close()

2） rabbit_consumer.py

# Author : Xuefengimport pikaconnection = pika.BlockingConnection(pika.Connection.Parameters( "localhost"))# statement a channelchannel = connection.channel()channel.queue_declare(queue="hello", durable=True)def callback(ch, method, properties, body): ''' Handle the recieved data :param ch: The address of the channel :param method: Information about the connection :param properties: :param body: :return: ''' print("------>", ch, method, properties ) print("[x] Recieved %r" % body) # ack by ourself ch.basic_ack(delivery_tag = method.delivery_tag)# follow is for consumer to auto change with the abilitychannel.basic_qos(profetch_count=1)# no_ack = True represent that the message cannot be transfor to next consumer,# when the current consumer is stop by accident.channel.basic_consume(callback, # If have recieved message, enable the callback() function to handle the message. queue = "hello", no_ack = True)print("[*] Waiting for messages. To Exit press CTRL+C")channel.start_consuming()

用rabbitmq中的fanout模式实现广播模式

1） fanout_rabbit_publish.py

# Author : Xuefengimport pikaimport sys# 广播模式：# 生产者发送一条消息，所有的开通链接的消费者都可以接收到消息connection = pika.BlockingConnection(pika.Connection.Parameters( "localhost"))# statement a channelchannel = 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] Send %r" % message)connection.close()

2） fanout_rabbit_consumer.py

# Author : Xuefengimport pikaimport sysconnection = pika.BlockingConnection(pika.Connection.Parameters( "localhost"))# statement a channelchannel = connection.channel()# exclusive 排他，唯一的 随机生成queueresult = channel.queue_declare(exclusive=True)queue_name = result.method.queueprint("Random queue name:", queue_name)channel.queue_bind(exchange="logs", queue=queue_name)def callback(ch, method, properties, body): ''' Handle the recieved data :param ch: The address of the channel :param method: Information about the connection :param properties: :param body: :return: ''' print("------>", ch, method, properties ) print("[x] Recieved %r" % body) # ack by ourself ch.basic_ack(delivery_tag = method.delivery_tag)# no_ack = True represent that the message cannot be transfor to next consumer,# when the current consumer is stop by accident.channel.basic_consume(callback, # If have recieved message, enable the callback() function to handle the message. queue = "hello", no_ack = True)print("[*] Waiting for messages. To Exit press CTRL+C")channel.start_consuming()

用rabbitmq中的direct模式实现消息过滤模式

1） direct_rabbit_publisher.py

# Author : Xuefengimport pikaimport sys# 消息过滤模式：# 生产者发送一条消息，通过severity优先级来确定是否可以接收到消息connection = pika.BlockingConnection(pika.Connection.Parameters( "localhost"))# statement a channelchannel = 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 "info:Hello world!"channel.basic_publish( exchange="direct_logs", routing_key=severity, body=message)print("[x] Send %r:%r" % (severity, message))connection.close()

2） direct_rabbit_consumer.py

# Author : Xuefengimport pikaimport sysconnection = pika.BlockingConnection(pika.Connection.Parameters( "localhost"))# statement a channelchannel = connection.channel()channel.exchange_declare(exchange="direct_logs", type="direct")# exclusive 排他，唯一的 随机生成queueresult = channel.queue_declare(exclusive=True)queue_name = result.method.queueprint("Random queue name:", queue_name)severities = 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) def callback(ch, method, properties, body): ''' Handle the recieved data :param ch: The address of the channel :param method: Information about the connection :param properties: :param body: :return: ''' print("------>", ch, method, properties ) print("[x] Recieved %r" % body) # ack by ourself ch.basic_ack(delivery_tag = method.delivery_tag)# no_ack = True represent that the message cannot be transfor to next consumer,# when the current consumer is stop by accident.channel.basic_consume(callback, # If have recieved message, enable the callback() function to handle the message. queue = "hello", no_ack = True)print("[*] Waiting for messages. To Exit press CTRL+C")channel.start_consuming()

用rabbitmq中的topic模式实现细致消息过滤模式

1） topic_rabbit_publisher.py

# Author : Xuefengimport pikaimport sys# 消息细致过滤模式：# 生产者发送一条消息，通过运行脚本 *.info 等确定接收消息类型进行对应接收connection = pika.BlockingConnection(pika.Connection.Parameters( "localhost"))# statement a channelchannel = connection.channel()channel.exchange_declare(exchange="topic_logs", type="topic")binding_key = sys.argv[1] if len(sys.argv) > 1 else "info"message = ' '.join(sys.argv[2:]) or "info:Hello world!"channel.basic_publish( exchange="topic_logs", routing_key=binding_key, body=message)print("[x] Send %r:%r" % (binding_key, message))connection.close()

2） topic_rabbit_consumer.py

# Author : Xuefengimport pikaimport sysconnection = pika.BlockingConnection(pika.Connection.Parameters( "localhost"))# statement a channelchannel = connection.channel()channel.exchange_declare(exchange="topic_logs", type="topic")# exclusive 排他，唯一的 随机生成queueresult = channel.queue_declare(exclusive=True)queue_name = result.method.queueprint("Random queue name:", queue_name)binding_keys = sys.argv[1:]if not binding_keys: sys.stderr.write("Usage:%s [info] [warning] [error]\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)def callback(ch, method, properties, body): ''' Handle the recieved data :param ch: The address of the channel :param method: Information about the connection :param properties: :param body: :return: ''' print("------>", ch, method, properties) print("[x] Recieved %r" % body) # ack by ourself ch.basic_ack(delivery_tag=method.delivery_tag)# no_ack = True represent that the message cannot be transfor to next consumer,# when the current consumer is stop by accident.channel.basic_consume(callback, # If have recieved message, enable the callback() function to handle the message. queue="hello", no_ack=True)print("[*] Waiting for messages. To Exit press CTRL+C")channel.start_consuming()

用rabbitmq实现rpc操作

1） Rpc_rabbit_client.py

# Author : Xuefengimport pikaimport timeimport uuidclass FibonacciRpcClient(object): def __init__(self): self.connection = pika.BlockingConnection(pika.Connection.Parameters( "localhost")) self.channel = self.connection.channel() result = self.channel.queue_declare(exclusive=True) self.callback_queue = result.method.queue # 随机的生成一个接收命令执行结果的队列 self.channel.basic_consume(self.on_response, # 只要收到消息就调用 no_ack=True, queue=self.callback_queue) def on_response(self, ch, method, props, body): if self.corr_id == props.correlation_id: self.response = body def call(self,n): self.response = None self.corr_id = str(uuid.uuid4()) self.channel.basic_publish( exchange="", routing_key="rpc_queue", properties=pika.BasicPropreties( rely_to=self.callback_queue, correlation_id=self.corr_id # 通过随机生成的ID来验证指令执行结果与指令的匹配性 ), body=str(n) ) while self.response is None: self.connection.process_data_events() # 非阻塞版的start_consume，有没有消息都继续 print("no message...") time.sleep(0.5) return int(self.response)fibonacci_rcp = FibonacciRpcClient()print("[x] Requesting fib(30)")response = fibonacci_rcp.call(30)print("[x] Rec %r" % response)

2） Rpc_rabbit_server.py

# Author : Xuefengimport pikaimport sysconnection = pika.BlockingConnection(pika.Connection.Parameters( "localhost"))# statement a channelchannel = connection.channel()channel.queue_declare(queue="rpc_queue")def fib(n): if n == 0: return 0 elif n == 1: return 1 else: return fib(n-1)+fib(n-2)def on_request(ch, method, props, body): n = int(body) print("[.] fib(%s)" % n) response = fib(n) ch.basic_publish( exchange="", routing_key=props.rely_to, properties=pika.BasicPropreties(correlation_id=\ props.correlation), body = str(body) ) ch.basic_ack(delivery_tag=method.delivery_tag)channel.basic_qos(prefetch_count=1)channel.basic_consume(on_request, queue="rpc_queue")print("[x] Awaiting RPC requests")channel.start_consumeing()channel.exchange_declare(exchange="direct_logs", type="direct")# exclusive 排他，唯一的 随机生成queueresult = channel.queue_declare(exclusive=True)queue_name = result.method.queueprint("Random queue name:", queue_name)severities = sys.argv[1:]

到此这篇关于Python RabbitMQ实现简单的进程间通信示例的文章就介绍到这了,更多相关Python RabbitMQ进程间通信内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！