欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

python實(shí)現(xiàn)RabbitMQ的消息隊(duì)列的示例代碼

 更新時(shí)間：2018年11月08日 10:11:41   作者：wyzane  
這篇文章主要介紹了python實(shí)現(xiàn)RabbitMQ的消息隊(duì)列的示例代碼，總結(jié)了RabbitMQ中三種exchange模式的實(shí)現(xiàn)，分別是fanout, direct和topic。感興趣的小伙伴們可以參考一下

最近在研究redis做消息隊(duì)列時(shí)，順便看了一下RabbitMQ做消息隊(duì)列的實(shí)現(xiàn)。以下是總結(jié)的RabbitMQ中三種exchange模式的實(shí)現(xiàn)，分別是fanout, direct和topic。

base.py：

import pika
# 獲取認(rèn)證對(duì)象，參數(shù)是用戶名、密碼。遠(yuǎn)程連接時(shí)需要認(rèn)證
credentials = pika.PlainCredentials("admin", "admin")

# BlockingConnection(): 實(shí)例化連接對(duì)象
# ConnectionParameters(): 實(shí)例化鏈接參數(shù)對(duì)象
connection = pika.BlockingConnection(pika.ConnectionParameters(
  "192.168.0.102", 5672, "/", credentials))

# 創(chuàng)建新的channel(通道)
channel = connection.channel()

fanout模式：向綁定到指定exchange的queue中發(fā)送消息，消費(fèi)者從queue中取出數(shù)據(jù)，類似于廣播模式、發(fā)布訂閱模式。
綁定方式: 在接收端channel.queue_bind(exchange="logs", queue=queue_name)

代碼：

publisher.py:

from base import channel, connection
# 聲明exchange, 不聲明queue
channel.exchange_declare(exchange="logs", exchange_type="fanout") # 廣播
message = "hello fanout"
channel.basic_publish(
  exchange="logs",
  routing_key="",
  body=message
)
connection.close()

consumer.py：

from base import channel, connection
    
# 聲明exchange
channel.exchange_declare(exchange="logs", exchange_type="fanout")

# 不指定queue名字, rabbitmq會(huì)隨機(jī)分配一個(gè)名字, 消息處理完成后queue會(huì)自動(dòng)刪除
result = channel.queue_declare(exclusive=True) 

# 獲取queue名字
queue_name = result.method.queue

# 綁定exchange和queue
channel.queue_bind(exchange="logs", queue=queue_name)


def callback(ch, method, properties, body):
  print("body:%s" % body)


channel.basic_consume(
  callback,
  queue=queue_name
)


channel.start_consuming()

direct模式：發(fā)送端綁定一個(gè)routing_key1, queue中綁定若干個(gè)routing_key2, 若key1與key2相等，或者key1在key2中，則消息就會(huì)發(fā)送到這個(gè)queue中，再由相應(yīng)的消費(fèi)者去queue中取數(shù)據(jù)。

publisher.py：

from base import channel, connection
channel.exchange_declare(exchange="direct_test", exchange_type="direct")

message = "hello"

channel.basic_publish(
  exchange="direct_test",
  routing_key="info", # 綁定key
  body=message
)
connection.close()

consumer01.py：

from base import channel, connection
      
      
channel.exchange_declare(exchange="direct_test", exchange_type="direct")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue


channel.queue_bind(
  exchange="direct_test",
  queue=queue_name,
  # 綁定的key,與publisher中的相同
  routing_key="info" 
)


def callback(ch, method, properties, body):
  print("body:%s" % body)


channel.basic_consume(
  callback,
  queue=queue_name
)


channel.start_consuming()

consumer02.py：

from base import channel, connection


channel.exchange_declare(exchange="direct_test", exchange_type="direct")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue


channel.queue_bind(
  exchange="direct_test",
  queue=queue_name,
  # 綁定的key
  routing_key="error"  
)


def callback(ch, method, properties, bosy):
  print("body:%s" % body)


channel.basic_consume(
  callback,
  queue=queue_name
)


channel.start_consuming()

consumer03.py：

from base import channel, connection
      
      
channel.exchange_declare(exchange="direct_test", exchange_type="direct")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue


key_list = ["info", "warning"]
for key in key_list:
  channel.queue_bind(
    exchange="direct_test",
    queue=queue_name,
    # 一個(gè)queue同時(shí)綁定多個(gè)key，有一個(gè)key滿足條件時(shí)就可以收到數(shù)據(jù)
    routing_key=key 
  )


def callback(ch, method, properties, body):
  print("body:%s" % body)


channel.basic_consume(
  callback,
  queue=queue_name
)


channel.start_consuming()

執(zhí)行:

python producer.py
python consumer01.py
python consumer02.py
python consumer03.py

結(jié)果:

consumer01.py: body:b'hello'
consumer02.py沒(méi)收到結(jié)果
consumer03.py: body:b'hello'

topic模式不是太好理解，我的理解如下:

對(duì)于發(fā)送端綁定的routing_key1，queue綁定若干個(gè)routing_key2；若routing_key1滿足任意一個(gè)routing_key2，則該消息就會(huì)通過(guò)exchange發(fā)送到這個(gè)queue中，然后由接收端從queue中取出其實(shí)就是direct模式的擴(kuò)展。

綁定方式:

發(fā)送端綁定:

  channel.basic_publish(
    exchange="topic_logs",
    routing_key=routing_key,
    body=message
  )

接收端綁定:

  channel.queue_bind(
    exchange="topic_logs",
    queue=queue_name,
    routing_key=binding_key
  )

publisher.py：

import sys
from base import channel, connection


# 聲明exchange
channel.exchange_declare(exchange="topic_test", exchange_type="topic")

# 待發(fā)送消息
message = " ".join(sys.argv[1:]) or "hello topic"

# 發(fā)布消息
channel.basic_publish(
  exchange="topic_test",
  routing_key="mysql.error",  # 綁定的routing_key
  body=message
)
connection.close()

consumer01.py:

from base import channel, connection
      
      
channel.exchange_declare(exchange="topic_test", exchange_type="topic")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue


channel.queue_bind(
  exchange="topic_test",
  queue=queue_name,
  routing_key="*.error"  # 綁定的routing_key
)


def callback(ch, method, properties, body):
  print("body:%s" % body)


channel.basic_consume(
  callback,
  queue=queue_name,
  no_ack=True
)


channel.start_consuming()

consumer02.py:

from base import channel, connection
      
      
channel.exchange_declare(exchange="topic_test", exchange_type="topic")
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue


channel.queue_bind(
  exchange="topic_test",
  queue=queue_name,
  routing_key="mysql.*"  # 綁定的routing_key
)


def callback(ch, method, properties, body):
  print("body:%s" % body)


channel.basic_consume(
  callback,
  queue=queue_name,
  no_ack=True
)


channel.start_consuming()

執(zhí)行:

python publisher02.py "this is a topic test"
python consumer01.py
python consumer02.py

結(jié)果:

consumer01.py的結(jié)果: body:b'this is a topic test'
consumer02.py的結(jié)果: body:b'this is a topic test'

說(shuō)明通過(guò)綁定相應(yīng)的routing_key，兩個(gè)消費(fèi)者都收到了消息

將publisher.py的routing_key改成"mysql.info"

再此執(zhí)行:

python publisher02.py "this is a topic test"
python consumer01.py
python consumer02.py

結(jié)果:

consumer01.py沒(méi)收到結(jié)果
consumer02.py的結(jié)果: body:b'this is a topic test'

通過(guò)這個(gè)例子我們就能明白topic的運(yùn)行方式了。

參考自： http://www.dbjr.com.cn/article/150386.htm

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

相關(guān)文章

  • python遍歷文件夾,指定遍歷深度與忽略目錄的方法

    python遍歷文件夾,指定遍歷深度與忽略目錄的方法

    今天小編就為大家分享一篇python遍歷文件夾,指定遍歷深度與忽略目錄的方法，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧
    2018-07-07
  • django與小程序?qū)崿F(xiàn)登錄驗(yàn)證功能的示例代碼

    django與小程序?qū)崿F(xiàn)登錄驗(yàn)證功能的示例代碼

    這篇文章主要介紹了django與小程序?qū)崿F(xiàn)登錄驗(yàn)證功能的示例代碼，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧
    2019-02-02
  • 使用tensorflow實(shí)現(xiàn)AlexNet

    使用tensorflow實(shí)現(xiàn)AlexNet

    這篇文章主要為大家詳細(xì)介紹了使用tensorflow實(shí)現(xiàn)AlexNet，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下
    2017-11-11
  • 如何用Python中19行代碼把照片寫入到Excel中

    如何用Python中19行代碼把照片寫入到Excel中

    這篇文章主要介紹了如何利用Python3中的19行代碼把照片寫入到Excel中，文章內(nèi)容實(shí)現(xiàn)的不是截取一張圖片，粘貼到excel，而是通過(guò)像素寫入到excel中,需要的朋友可以參考一下
    2022-02-02
  • 7個(gè)流行的Python強(qiáng)化學(xué)習(xí)算法及代碼實(shí)現(xiàn)詳解

    7個(gè)流行的Python強(qiáng)化學(xué)習(xí)算法及代碼實(shí)現(xiàn)詳解

    目前流行的強(qiáng)化學(xué)習(xí)算法包括?Q-learning、SARSA、DDPG、A2C、PPO、DQN?和?TRPO。這些算法已被用于在游戲、機(jī)器人和決策制定等各種應(yīng)用中，本文我們將對(duì)其做一個(gè)簡(jiǎn)單的介紹，感興趣的可以學(xué)習(xí)一下
    2023-01-01
  • 一篇文章帶你詳細(xì)了解python中一些好用的庫(kù)

    一篇文章帶你詳細(xì)了解python中一些好用的庫(kù)

    這篇文章主要推薦了一些好用的Python庫(kù)，都有這不錯(cuò)的群眾基礎(chǔ)，非常好用，也都很實(shí)用，推薦給大家，希望能夠給你帶來(lái)幫助
    2021-09-09
  • python 非線性規(guī)劃方式(scipy.optimize.minimize)

    python 非線性規(guī)劃方式(scipy.optimize.minimize)

    今天小編就為大家分享一篇python 非線性規(guī)劃方式(scipy.optimize.minimize)，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧
    2020-02-02
  • python詞云庫(kù)wordCloud使用方法詳解(解決中文亂碼)

    python詞云庫(kù)wordCloud使用方法詳解(解決中文亂碼)

    這篇文章主要介紹了python詞云庫(kù)wordCloud使用方法詳解(解決中文亂碼),需要的朋友可以參考下
    2020-02-02
  • 使用Python編排Dockerfile的代碼詳解

    使用Python編排Dockerfile的代碼詳解

    docker compose 是 docker 的容器編排工具,它是基于 YAML 配置,YAML 是一種配置文件格式,支持傳遞環(huán)境變量,但是對(duì)于復(fù)雜的容器編排顯得力不從心,于是我便開(kāi)發(fā)這個(gè)程序,可以像寫程序一樣編排 docker,感興趣的的朋友可以參考下
    2024-08-08
  • 使用Jest?在?Visual?Studio?Code?中進(jìn)行單元測(cè)試的流程分析

    使用Jest?在?Visual?Studio?Code?中進(jìn)行單元測(cè)試的流程分析

    Jest是一個(gè)流行的JavaScript測(cè)試框架，它提供了簡(jiǎn)潔、靈活和強(qiáng)大的工具來(lái)編寫和運(yùn)行單元測(cè)試，今天通過(guò)本文給大家介紹使用Jest在Visual Studio Code中進(jìn)行單元測(cè)試的流程分析，感興趣的朋友跟隨小編一起看看吧
    2023-07-07

最新評(píng)論