Python RabbitMQ实现简单的进程间通信示例

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 : Xuefeng

import pika

connection = pika.BlockingConnection(pika.Connection.Parameters(
 "localhost"
))
# statement a channel
channel = 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 through
channel.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 : Xuefeng

import pika

connection = pika.BlockingConnection(pika.Connection.Parameters(
 "localhost"
))
# statement a channel
channel = 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 ability
channel.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 : Xuefeng

import pika
import sys

# 广播模式:
# 生产者发送一条消息,所有的开通链接的消费者都可以接收到消息

connection = pika.BlockingConnection(pika.Connection.Parameters(
 "localhost"
))
# statement a channel
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] Send %r" % message)

connection.close()

2) fanout_rabbit_consumer.py

# Author : Xuefeng

import pika
import sys

connection = pika.BlockingConnection(pika.Connection.Parameters(
 "localhost"
))
# statement a channel
channel = connection.channel()
# exclusive 排他,唯一的 随机生成queue
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
print("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 : Xuefeng
import pika
import sys

# 消息过滤模式:
# 生产者发送一条消息,通过severity优先级来确定是否可以接收到消息

connection = pika.BlockingConnection(pika.Connection.Parameters(
 "localhost"
))
# statement a channel
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 "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 : Xuefeng

import pika
import sys

connection = pika.BlockingConnection(pika.Connection.Parameters(
 "localhost"
))
# statement a channel
channel = connection.channel()

channel.exchange_declare(exchange="direct_logs",
       type="direct")

# exclusive 排他,唯一的 随机生成queue
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
print("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 : Xuefeng

import pika
import sys

# 消息细致过滤模式:
# 生产者发送一条消息,通过运行脚本 *.info 等确定接收消息类型进行对应接收
connection = pika.BlockingConnection(pika.Connection.Parameters(
 "localhost"
))
# statement a channel
channel = 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 : Xuefeng

import pika
import sys

connection = pika.BlockingConnection(pika.Connection.Parameters(
 "localhost"
))
# statement a channel
channel = connection.channel()

channel.exchange_declare(exchange="topic_logs",
       type="topic")

# exclusive 排他,唯一的 随机生成queue
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
print("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 : Xuefeng

import pika
import time
import uuid

class 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 : Xuefeng

import pika
import sys

connection = pika.BlockingConnection(pika.Connection.Parameters(
 "localhost"
))
# statement a channel
channel = 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 排他,唯一的 随机生成queue
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
print("Random queue name:", queue_name)

severities = sys.argv[1:]

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

(0)

相关推荐

  • python 进程 进程池 进程间通信实现解析

    1.python 中创建进程的两种方式: from multiprocessing import Process import time def test_(): print '-----test-----' if __name__ == '__main__': p = Process(target=test_) p.start() while True: print '--main--' '''1.通过process 类创建一个进程对象,然后start即可开启进程, test test_函数是

  • Python进程间通信Queue消息队列用法分析

    本文实例讲述了Python进程间通信Queue消息队列用法.分享给大家供大家参考,具体如下: 进程间通信-Queue Process之间有时需要通信,操作系统提供了很多机制来实现进程间的通信. 1. Queue的使用 可以使用multiprocessing模块的Queue实现多进程之间的数据传递,Queue本身是一个消息列队程序,首先用一个小实例来演示下Queue的工作原理: 代码如下: #coding=utf-8 from multiprocessing import Queue #初始化一个

  • python多进程实现进程间通信实例

    python中的多线程其实并不是真正的多线程,如果想要充分地使用多核CPU的资源,在python中大部分情况需要使用多进程.Python提供了非常好用的多进程包multiprocessing,只需要定义一个函数,Python会完成其他所有事情.借助这个包,可以轻松完成从单进程到并发执行的转换.multiprocessing支持子进程.通信和共享数据.执行不同形式的同步,提供了Process.Queue.Pipe.Lock等组件. multiprocessing.Queue() 以Queue为例,

  • Python队列、进程间通信、线程案例

    进程互斥锁 多进程同时抢购余票 # 并发运行,效率高,但竞争写同一文件,数据写入错乱 # data.json文件内容为 {"ticket_num": 1} import json import time from multiprocessing import Process def search(user): with open('data.json', 'r', encoding='utf-8') as f: dic = json.load(f) print(f'用户{user}查看

  • python实现进程间通信简单实例

    本文实例讲解了python实现两个程序之间通信的方法,具体方法如下: 该实例采用socket实现,与socket网络编程不一样的是socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)的第一个参数是socket.AF_UNIX 而不是 socket.AF_INET 例中两个python程序 s.py/c.py 要先运行s.py 基于fedora13/python2.6测试,成功实现! s.py代码如下: #!/usr/bin/env python im

  • Python进程间通信之共享内存详解

    前一篇博客说了怎样通过命名管道实现进程间通信,但是要在windows是使用命名管道,需要使用python调研windows api,太麻烦,于是想到是不是可以通过共享内存的方式来实现.查了一下,Python中可以使用mmap模块来实现这一功能. Python中的mmap模块是通过映射同一个普通文件实现共享内存的.文件被映射到进程地址空间后,进程可以像访问内存一样对文件进行访问. 不过,mmap在linux和windows上的API有些许的不一样,具体细节可以查看mmap的文档. 下面看一个例子:

  • 详解Python进程间通信之命名管道

    管道是一种简单的FIFO通信信道,它是单向通信的. 通常启动进程创建一个管道,然后这个进程创建一个或者多个进程子进程接受管道信息,由于管道是单向通信,所以经常需要创建两个管道来实现双向通信. 命名管道是对传统管道的扩展,默认的管道是匿名管道,只在程序运行时存在:而命名管道是持久化的,当不需要时需要删除它. 命名管道使用文件系统,由mkfifo()方法创建.一旦创建了,两个独立的进程都可以访问它,一个读,另外一个写. 命名管道支持阻塞读和阻塞写操作: 如果一个进程打开文件读,它会阻塞直到另外一个进

  • Python RabbitMQ实现简单的进程间通信示例

    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

  • python实现socket简单通信的示例代码

    首先先来简单介绍下socket: (具体更详细介绍的可以在网上找找,都讲得非常详细),这里主要是我自己的一些理解. socket是在应用层与传输层之间的一个抽象层,它的本质是编程接口,通过socket,才能实现TCP/IP协议. 它就是一个底层套件,用来处理最底层消息的接受和发送. socket翻译为套接字,可以把TCP/IP复杂的操作抽象为简单的几个接口来供应用层调用来实现进程在网络中的通信.socket起源于Unix,而Unix的基本要素之一就是"一切都为文件",即可以通过打开--

  • Python实现ATM简单功能的示例详解

    目录 一.局部变量.全局变量 二.global关键字 三.编写ATM程序 要求 详细步骤 存在问题 改进 完整代码 一.局部变量.全局变量 1.什么是局部变量 作用范围在函数内部,在函数外部无法使用 2.什么是全局变量 在函数内部和外部均可使用 3.如何将函数内定义的变量声明为全局变量 使用global关键字,global变量 二.global关键字 思考: num=100 def testsA() print(num) def testB() num=200 print(num) testA(

  • Python人工智能构建简单聊天机器人示例详解

    目录 引言 什么是聊天机器人? 准备工作 创建聊天机器人 导入必要的库 定义响应集合 创建聊天机器人 运行聊天机器人 完整代码 结论 展望 引言 人工智能是计算机科学中一个非常热门的领域,近年来得到了越来越多的关注.它通过模拟人类思考过程和智能行为来实现对复杂任务的自主处理和学习,已经被广泛应用于许多领域,包括语音识别.自然语言处理.机器人技术.图像识别和推荐系统等. 本文将介绍如何使用Python构建一个简单的聊天机器人,以展示人工智能的基本原理和应用.我们将使用Python语言和自然语言处理

  • Python有序字典简单实现方法示例

    本文实例讲述了Python有序字典简单实现方法.分享给大家供大家参考,具体如下: 代码: # -*- coding: UTF-8 -*- import collections print 'Regular dictionary:' d = {} d['a'] = 'A' d['b'] = 'B' d['c'] = 'C' for k, v in d.items(): print k, v print '\nOrderedDict:' d = collections.OrderedDict() d

  • Python Socket实现简单TCP Server/client功能示例

    本文实例讲述了Python Socket实现简单TCP Server/client功能.分享给大家供大家参考,具体如下: 网络上关于socket的介绍文章数不胜数.自己记录下学习的点点滴滴.以供将来复习学习使用. socket中文的翻译是套接字,总感觉词不达意.简单的理解就是ip+port形成的一个管理单元.也是程序中应用程序调用的接口. 在这里我们先介绍如何启动tcp 的server. tcp连接中server部分,启动一个ip和port口,在这个port口监听,当收到client发来的请求,

  • C#调用Python脚本的简单示例

    IronPython是一种在 .NET及 Mono上的 Python实现,由微软的 Jim Hugunin所发起,是一个开源的项目,基于微软的 DLR引擎.IronPython的在CodePlex上的主页:http://ironpython.codeplex.com/ 使用场景: 如果你的小伙伴会写Python脚本,而且已经实现大部分项目的功能不需要再用C# 实现.现在缺少窗体,此时Python+C#的组合就可以完美的结局问题啦! 示例: 借由IronPython,就可以利用.NET执行存储在P

  • Python编程实现的简单神经网络算法示例

    本文实例讲述了Python编程实现的简单神经网络算法.分享给大家供大家参考,具体如下: python实现二层神经网络 包括输入层和输出层 # -*- coding:utf-8 -*- #! python2 import numpy as np #sigmoid function def nonlin(x, deriv = False): if(deriv == True): return x*(1-x) return 1/(1+np.exp(-x)) #input dataset x = np.

  • Python使用matplotlib简单绘图示例

    本文实例讲述了Python使用matplotlib简单绘图.分享给大家供大家参考,具体如下: # -*- coding:utf-8 -*- #! python2 """ Created on Mon Apr 24 12:48:40 2017 @author: x-power """ import matplotlib.pyplot as plt #首先载入 matplotlib的绘图模块pyplot,并且重命名为plt. import numpy

  • Python实现一个简单的毕业生信息管理系统的示例代码

    写在前面: 从昨晚的梦里回忆起数据管理的作业: 实现一个自己的选题---- 毕业生信息管理系统,实现学生个人信息基本的增删改查, 我想了想前段时间刚学习的列表,这个简单啊 ,设计一个学生信息列表,然后列表里面再存每个学生详细信息的列表,然后来实现一个基本的增删查改,这个不难啊!直接开始撸代码! 上代码! def Menu():##菜单主界面 print('*'*22) print("* 查看毕业生列表输入: 1 *") print("* 添加毕业生信息输入: 2 *"

随机推荐