溫馨提示×

您好，登錄后才能下訂單哦！

密碼登錄×
登錄注冊(cè)×
其他方式登錄
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》

Python中RabbitMQ如何實(shí)現(xiàn)進(jìn)程間通信

發(fā)布時(shí)間：2020-07-02 14:38:24 來(lái)源：億速云 閱讀：249 作者：清晨 欄目：開(kāi)發(fā)技術(shù)

這篇文章主要介紹Python中RabbitMQ如何實(shí)現(xiàn)進(jìn)程間通信，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們一定要看完！

RabbitMQ    消息隊(duì)列

PY
threading Queue
進(jìn)程Queue 父進(jìn)程與子進(jìn)程，或同一父進(jìn)程下的多個(gè)子進(jìn)程進(jìn)行交互
缺點(diǎn)：兩個(gè)不同Python文件不能通過(guò)上面兩個(gè)Queue進(jìn)行交互

erlong
基于這個(gè)語(yǔ)言創(chuàng)建的一種中間商
win中需要先安裝erlong才能使用
rabbitmq_server start

安裝 Python module

pip install pika

or

easy_install pika

or
源碼

rabbit      默認(rèn)端口15672
查看當(dāng)前時(shí)刻的隊(duì)列數(shù)
rabbitmqctl.bat list_queue

exchange
在定義的時(shí)候就是有類型的，決定到底哪些queue符合條件，可以接受消息
fanout：所有bind到此exchange的queue都可以收到消息
direct：通過(guò)routingkey和exchange決定唯一的queue可以接受消息
topic： 所有符合routingkey（此時(shí)可以是一個(gè)表達(dá)式）的routingkey所bind的queue都可以接受消息
      表達(dá)式符號(hào)說(shuō)明：
      # 代表一個(gè)或多個(gè)字符     * 代表任何字符

RPC
remote procedure call           雙向傳輸，指令<-------->指令執(zhí)行結(jié)果
實(shí)現(xiàn)方法：                        創(chuàng)建兩個(gè)隊(duì)列，一個(gè)隊(duì)列收指令，一個(gè)隊(duì)列發(fā)送執(zhí)行結(jié)果

用rabbitmq實(shí)現(xiàn)簡(jiǎn)單的生產(chǎn)者消費(fèi)者模型

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模式實(shí)現(xiàn)廣播模式

1） fanout_rabbit_publish.py

# Author : Xuefeng

import pika
import sys

# 廣播模式：
# 生產(chǎn)者發(fā)送一條消息，所有的開(kāi)通鏈接的消費(fèi)者都可以接收到消息

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 排他，唯一的 隨機(jī)生成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模式實(shí)現(xiàn)消息過(guò)濾模式

1） direct_rabbit_publisher.py

# Author : Xuefeng
import pika
import sys

# 消息過(guò)濾模式：
# 生產(chǎn)者發(fā)送一條消息，通過(guò)severity優(yōu)先級(jí)來(lái)確定是否可以接收到消息

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 排他，唯一的 隨機(jī)生成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模式實(shí)現(xiàn)細(xì)致消息過(guò)濾模式

1） topic_rabbit_publisher.py

# Author : Xuefeng

import pika
import sys

# 消息細(xì)致過(guò)濾模式：
# 生產(chǎn)者發(fā)送一條消息，通過(guò)運(yùn)行腳本 *.info 等確定接收消息類型進(jìn)行對(duì)應(yīng)接收
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 排他，唯一的 隨機(jī)生成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實(shí)現(xiàn)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    # 隨機(jī)的生成一個(gè)接收命令執(zhí)行結(jié)果的隊(duì)列
    self.channel.basic_consume(self.on_response,  # 只要收到消息就調(diào)用
                  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     # 通過(guò)隨機(jī)生成的ID來(lái)驗(yàn)證指令執(zhí)行結(jié)果與指令的匹配性
      ),
      body=str(n)
    )
    while self.response is None:
      self.connection.process_data_events()  # 非阻塞版的start_consume，有沒(méi)有消息都繼續(xù)
      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 排他，唯一的 隨機(jī)生成queue
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
print("Random queue name:", queue_name)

severities = sys.argv[1:]


以上是Python中RabbitMQ如何實(shí)現(xiàn)進(jìn)程間通信的所有內(nèi)容，感謝各位的閱讀！希望分享的內(nèi)容對(duì)大家有幫助，更多相關(guān)知識(shí)，歡迎關(guān)注億速云行業(yè)資訊頻道！

向AI問(wèn)一下細(xì)節(jié)

免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI
助
手