消息隊列rabbitmq
爲何用消息隊列
舉例html
好比在一個企業裏，技術老大接到boss的任務，技術老大把這個任務拆分紅多個小任務，完成全部的小任務就算搞定整個任務了。 那麼在執行這些小任務的時候，可能有一個環節很費時間，而且優先級很低，推遲完成也不影響整個任務運轉，那麼技術老大就會將這個很費時間，且不重要的任務，丟給他的小弟去解決，本身繼續完成其餘任務。
轉化爲計算機思想前端
那個技術老大就是一個 程序系統，那個小弟就是消息隊列。 當程序系統發現某些任務耗費時間且優先級較低，遲點完成也不影響整個任務，就把這個任務丟給消息隊列。
場景java
在程序系統中，例如外賣系統，訂單系統，庫存系統，優先級較高 發紅包，發郵件，發短信，app消息推送等任務優先級很低，很適合交給消息隊列去處理，以便於程序系統更快的處理其餘請求。
消息隊列工做流程python
消息隊列通常有三個角色： 隊列服務端 隊列生產者 隊列消費者
消息隊列工做流程就如同一個流水線，有產品加工，一個輸送帶，一個打包產品 輸送帶就是 不停運轉的消息隊列服務端 加工產品的就是 隊列生產者 在傳輸帶結尾打包產品的 就是隊列消費者
隊列產品nginx
RabbitMQ Erlang編寫的消息隊列產品，企業級消息隊列軟件，支持消息負載均衡，數據持久化等。 ZeroMQ saltstack軟件使用此消息，速度最快。 Redis key-value的系統，也支持隊列數據結構，輕量級消息隊列 Kafka 由Scala編寫，目標是爲處理實時數據提供一個統1、高通量、低等待的平臺
一個app系統消息隊列工做流程web
消費者，一個後臺進程，不斷的去檢測消息隊列中是否有消息，有消息就取走，開啓新線程去處理業務，若是沒有一會再來
消息隊列做用
1）程序解耦正則表達式
容許你獨立的擴展或修改兩邊的處理過程，只要確保它們遵照一樣的接口約束。centos
2）冗餘：緩存
消息隊列把數據進行持久化直到它們已經被徹底處理，經過這一方式規避了數據丟失風險。安全
許多消息隊列所採用的"插入-獲取-刪除"範式中，在把一個消息從隊列中刪除以前，須要你的處理系統明確的指出該消息已經被處理完畢，從而確保你的數據被安全的保存直到你使用完畢。
3）峯值處理能力：
(大白話，就是原本公司業務只須要5臺機器，可是臨時的秒殺活動，5臺機器確定受不了這個壓力，咱們又不可能將總體服務器架構提高到10臺，那在秒殺活動後，機器不就浪費了嗎？所以引入消息隊列)
在訪問量劇增的狀況下，應用仍然須要繼續發揮做用，可是這樣的突發流量並不常見。
若是爲以能處理這類峯值訪問爲標準來投入資源隨時待命無疑是巨大的浪費。
使用消息隊列可以使關鍵組件頂住突發的訪問壓力，而不會由於突發的超負荷的請求而徹底崩潰。
4）可恢復性：
系統的一部分組件失效時，不會影響到整個系統。
消息隊列下降了進程間的耦合度，因此即便一個處理消息的進程掛掉，加入隊列中的消息仍然能夠在系統恢復後被處理。
5）順序保證：
在大多使用場景下，數據處理的順序都很重要。
大部分消息隊列原本就是排序的，而且能保證數據會按照特定的順序來處理。（Kafka保證一個Partition內的消息的有序性）
6）緩衝：
有助於控制和優化數據流通過系統的速度，解決生產消息和消費消息的處理速度不一致的狀況。
7）異步通訊：
不少時候，用戶不想也不須要當即處理消息。好比發紅包，發短信等流程。
消息隊列提供了異步處理機制，容許用戶把一個消息放入隊列，但並不當即處理它。想向隊列中放入多少消息就放多少，而後在須要的時候再去處理它們。
1. 你瞭解的消息隊列
生活裏的消息隊列，如同郵局的郵箱，
若是沒郵箱的話，
郵件必須找到郵件那我的，遞給他，才玩完成，那這個任務會處理的很麻煩，很慢，效率很低
可是若是有了郵箱，
郵件直接丟給郵箱，用戶只須要去郵箱裏面去找，有沒有郵件，有就拿走，沒有就下次再來，這樣能夠極大的提高郵件收發效率！
rabbitmq是一個消息代理，它接收和轉發消息，能夠理解爲是生活的郵局。
你能夠將郵件放在郵箱裏，你能夠肯定有郵遞員會發送郵件給收件人。
歸納：
rabbitmq是接收，存儲，轉發數據的。
官方教程：http://www.rabbitmq.com/tutorials/tutorial-one-python.html
消息（Message）是指在應用間傳送的數據。消息能夠很是簡單，好比只包含文本字符串，也能夠更復雜，可能包含嵌入對象。
消息隊列（Message Queue）是一種應用間的通訊方式，消息發送後能夠當即返回，由消息系統來確保消息的可靠傳遞。消息發佈者只管把消息發佈到 MQ 中而不用管誰來取，消息使用者只管從 MQ 中取消息而無論是誰發佈的。這樣發佈者和使用者都不用知道對方的存在。
2. 公司在什麼狀況下會用消息隊列?
1.電商訂單
想必同窗們都點過外賣，點擊下單後的業務邏輯可能包括：檢查庫存、生成單據、發紅包、短信通知等，若是這些業務同步執行，完成下單率會很是低，如發紅包，短信通知等沒必要要的流程，異步執行便可。
此時使用MQ，能夠在核心流程（扣減庫存、生成訂單記錄）等完成後發送消息到MQ，快速結束本次流程。消費者拉取MQ消息時，發現紅包、短信等消息時，再進行處理。
場景：雙11是購物狂節,用戶下單後,訂單系統須要通知庫存系統,傳統的作法就是訂單系統調用庫存系統的接口
這種作法有一個缺點:
-
當庫存系統出現故障時,訂單就會失敗。(這樣馬雲將少賺好多好多錢錢。。。。)
-
訂單系統和庫存系統高耦合.
引入消息隊列
-
訂單系統:用戶下單後,訂單系統完成持久化處理,將消息寫入消息隊列,返回用戶訂單下單成功。
-
庫存系統:訂閱下單的消息,獲取下單消息,進行庫操做。 就算庫存系統出現故障,消息隊列也能保證消息的可靠投遞,不會致使消息丟失(馬雲這下高興了，鈔票快快的來呀~~).
2.秒殺活動
流量削峯通常在秒殺活動中應用普遍 場景:秒殺活動，通常會由於流量過大，致使應用掛掉,爲了解決這個問題，通常在應用前端加入消息隊列。 做用: 1.能夠控制活動人數，超過此必定閥值的訂單直接丟棄(怪不得我一次秒殺都沒搶到過。。。。。wtf？？？)
2.能夠緩解短期的高流量壓垮應用(應用程序按本身的最大處理能力獲取訂單)
3.用戶的請求，服務器接收到以後，寫入消息隊列，超過定義的閾值就直接丟棄請求，或者跳轉錯誤頁面。
4.業務系統取出隊列中的消息，再作後續處理。
3. rabbitMQ安裝
rabbitmq-server服務端 1.下載centos源 wget -O /etc/yum.repos.d/CentOS-Base.repo http://mirrors.cloud.tencent.com/repo/centos7_base.repo 2.下載epel源 wget -O /etc/yum.repos.d/epel.repo http://mirrors.cloud.tencent.com/repo/epel-7.repo 3.清空yum緩存而且生成新的yum緩存 yum clean all yum makecache 4.安裝erlang $ yum -y install erlang 5.安裝RabbitMQ $ yum -y install rabbitmq-server 6.啓動(無用戶名密碼): systemctl start/stop/restart/status rabbitmq-server 設置rabbitmq帳號密碼，以及角色權限設置 # 設置新用戶yugo 密碼123 sudo rabbitmqctl add_user yugo 123 # 設置用戶爲administrator角色 sudo rabbitmqctl set_user_tags yugo administrator # 設置權限，容許對全部的隊列都有權限
對何種資源具備配置、寫、讀的權限經過正則表達式來匹配，具體命令以下：
set_permissions [-p <vhostpath>] <user> <conf> <write> <read>
sudo rabbitmqctl set_permissions -p "/" yugo ".*" ".*" ".*" #重啓服務生效設置 service rabbitmq-server start/stop/restart rabbitmq相關命令 // 新建用戶 rabbitmqctl add_user {用戶名} {密碼} // 設置權限 rabbitmqctl set_user_tags {用戶名} {權限} // 查看用戶列表 rabbitmqctl list_users // 爲用戶受權 添加 Virtual Hosts ： rabbitmqctl add_vhost <vhost> // 刪除用戶 rabbitmqctl delete_user Username // 修改用戶的密碼 rabbitmqctl change_password Username Newpassword // 刪除 Virtual Hosts ： rabbitmqctl delete_vhost <vhost> // 添加 Users ： rabbitmqctl add_user <username> <password> rabbitmqctl set_user_tags <username> <tag> ... rabbitmqctl set_permissions [-p <vhost>] <user> <conf> <write> <read> // 刪除 Users ： delete_user <username> // 使用戶user1具備vhost1這個virtual host中全部資源的配置、寫、讀權限以便管理其中的資源 rabbitmqctl set_permissions -p vhost1 user1 '.*' '.*' '.*' // 查看權限 rabbitmqctl list_user_permissions user1 rabbitmqctl list_permissions -p vhost1 // 清除權限 rabbitmqctl clear_permissions [-p VHostPath] User //清空隊列步驟 rabbitmqctl reset 須要提早關閉應用rabbitmqctl stop_app ， 而後再清空隊列，啓動應用 rabbitmqctl start_app 此時查看隊列rabbitmqctl list_queues 查看全部的exchange： rabbitmqctl list_exchanges 查看全部的queue： rabbitmqctl list_queues 查看全部的用戶： rabbitmqctl list_users 查看全部的綁定（exchange和queue的綁定信息）： rabbitmqctl list_bindings 查看消息確認信息： rabbitmqctl list_queues name messages_ready messages_unacknowledged 查看RabbitMQ狀態，包括版本號等信息：rabbitmqctl status
#開啓web界面rabbitmq
rabbitmq-plugins enable rabbitmq_management
#訪問web界面
http://server-name:15672/
RabbitMQ組件解釋
AMQP
AMQP協議是一個高級抽象層消息通訊協議，RabbitMQ是AMQP協議的實現。它主要包括如下組件：
1.Server(broker): 接受客戶端鏈接，實現AMQP消息隊列和路由功能的進程。
2.Virtual Host:實際上是一個虛擬概念，相似於權限控制組，一個Virtual Host裏面能夠有若干個Exchange和Queue，可是權限控制的最小粒度是Virtual Host
3.Exchange:接受生產者發送的消息，並根據Binding規則將消息路由給服務器中的隊列。ExchangeType決定了Exchange路由消息的行爲，例如，在RabbitMQ中，ExchangeType有direct、Fanout和Topic三種，不一樣類型的Exchange路由的行爲是不同的。
4.Message Queue：消息隊列，用於存儲還未被消費者消費的消息。
5.Message: 由Header和Body組成，Header是由生產者添加的各類屬性的集合，包括Message是否被持久化、由哪一個Message Queue接受、優先級是多少等。而Body是真正須要傳輸的APP數據。
6.Binding:Binding聯繫了Exchange與Message Queue。Exchange在與多個Message Queue發生Binding後會生成一張路由表，路由表中存儲着Message Queue所需消息的限制條件即Binding Key。當Exchange收到Message時會解析其Header獲得Routing Key，Exchange根據Routing Key與Exchange Type將Message路由到Message Queue。Binding Key由Consumer在Binding Exchange與Message Queue時指定，而Routing Key由Producer發送Message時指定，二者的匹配方式由Exchange Type決定。
7.Connection:鏈接，對於RabbitMQ而言，其實就是一個位於客戶端和Broker之間的TCP鏈接。
8.Channel:信道，僅僅建立了客戶端到Broker之間的鏈接後，客戶端仍是不能發送消息的。須要爲每個Connection建立Channel，AMQP協議規定只有經過Channel才能執行AMQP的命令。一個Connection能夠包含多個Channel。之因此須要Channel，是由於TCP鏈接的創建和釋放都是十分昂貴的，若是一個客戶端每個線程都須要與Broker交互，若是每個線程都創建一個TCP鏈接，暫且不考慮TCP鏈接是否浪費，就算操做系統也沒法承受每秒創建如此多的TCP鏈接。RabbitMQ建議客戶端線程之間不要共用Channel，至少要保證共用Channel的線程發送消息必須是串行的，可是建議儘可能共用Connection。
9.Command:AMQP的命令，客戶端經過Command完成與AMQP服務器的交互來實現自身的邏輯。例如在RabbitMQ中，客戶端能夠經過publish命令發送消息，txSelect開啓一個事務，txCommit提交一個事務。
python客戶端
// rabbitmq官方推薦的python客戶端pika模塊 pip3 install pika
應用場景1：單發送單接收
生產-消費者模型
P 是生產者
C 是消費者
中間hello是消息隊列
能夠有多個P、多個C
P發送消息給hello隊列，C消費者從隊列中獲取消息，默認輪詢方式
生產者send.py
咱們的第一個程序send.py將向隊列發送一條消息。咱們須要作的第一件事是創建與RabbitMQ服務器的鏈接。
#!/usr/bin/env python import pika # 建立憑證，使用rabbitmq用戶密碼登陸 # 去郵局取郵件，必須得驗證身份 credentials = pika.PlainCredentials("s14","123") # 新建鏈接，這裏localhost能夠更換爲服務器ip # 找到這個郵局，等於鏈接上服務器 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.119.10',credentials=credentials)) # 建立頻道 # 建造一個大郵箱，隸屬於這家郵局的郵箱，就是個鏈接 channel = connection.channel() # 聲明一個隊列，用於接收消息，隊列名字叫「水許傳」 channel.queue_declare(queue='水許傳') # 注意在rabbitmq中，消息想要發送給隊列，必須通過交換(exchange)，初學可使用空字符串交換(exchange='')，它容許咱們精確的指定發送給哪一個隊列(routing_key=''),參數body值發送的數據 channel.basic_publish(exchange='', routing_key='水許傳', body='武松又去打老虎啦2') print("已經發送了消息") # 程序退出前，確保刷新網絡緩衝以及消息發送給rabbitmq，須要關閉本次鏈接 connection.close()
能夠同時存在多個接受者，等待接收隊列的消息，默認是輪訓方式分配消息
接受者receive.py，能夠運行屢次，運行多個消費者
import pika # 創建與rabbitmq的鏈接 credentials = pika.PlainCredentials("s14","123") connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.119.10',credentials=credentials)) channel = connection.channel() channel.queue_declare(queue="水許傳") def callbak(ch,method,properties,body): print("消費者接收到了任務：%r"%body.decode("utf8")) # 有消息來臨，當即執行callbak，沒有消息則夯住，等待消息 # 老百姓開始去郵箱取郵件啦，隊列名字是水許傳 channel.basic_consume(callbak,queue="水許傳",no_ack=True) # 開始消費，接收消息 channel.start_consuming()
練習：
分別啓動生產者、兩個消費者，往隊列發送數據，查看消費者的結果
應用場景2：單發送多接收
使用場景：一個發送端，多個接收端，如分佈式的任務派發。爲了保證消息發送的可靠性，不丟失消息，使消息持久化了。同時爲了防止接收端在處理消息時down掉，只有在消息處理完成後才發送ack消息。
rabbitmq消息確認之ack
官網資料：http://www.rabbitmq.com/tutorials/tutorial-two-python.html
默認狀況下，生產者發送數據給隊列，消費者取出消息後，數據將被清除。
特殊狀況，若是消費者處理過程當中，出現錯誤，數據處理沒有完成，那麼這段數據將從隊列丟失
no-ack機制
不確認機制也就是說每次消費者接收到數據後，無論是否處理完畢，rabbitmq-server都會把這個消息標記完成，從隊列中刪除
ACK機制
ACK機制用於保證消費者若是拿了隊列的消息，客戶端處理時出錯了，那麼隊列中仍然還存在這個消息，提供下一位消費者繼續取
流程
1.生產者無須變更，發送消息 2.消費者若是no_ack=True啊，數據消費後若是出錯就會丟失 反之no_ack=False，數據消費若是出錯，數據也不會丟失 3.ack機制在消費者代碼中演示
生產者.py 只負責發送數據便可，無須變更
#!/usr/bin/env python import pika # 建立憑證，使用rabbitmq用戶密碼登陸 # 去郵局取郵件，必須得驗證身份 credentials = pika.PlainCredentials("s14","123") # 新建鏈接，這裏localhost能夠更換爲服務器ip # 找到這個郵局，等於鏈接上服務器 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.119.10',credentials=credentials)) # 建立頻道 # 建造一個大郵箱，隸屬於這家郵局的郵箱，就是個鏈接 channel = connection.channel() # 新建一個hello隊列，用於接收消息 # 這個郵箱能夠收發各個班級的郵件，經過 channel.queue_declare(queue='金品沒') # 注意在rabbitmq中，消息想要發送給隊列，必須通過交換(exchange)，初學可使用空字符串交換(exchange='')，它容許咱們精確的指定發送給哪一個隊列(routing_key=''),參數body值發送的數據 channel.basic_publish(exchange='', routing_key='金品沒', body='潘金蓮又出去。。。') print("已經發送了消息") # 程序退出前，確保刷新網絡緩衝以及消息發送給rabbitmq，須要關閉本次鏈接 connection.close()
消費者.py給與ack回覆
拿到消息必須給rabbitmq服務端回覆ack信息，不然消息不會被刪除，防止客戶端出錯，數據丟失
import pika credentials = pika.PlainCredentials("s14","123") connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.119.10',credentials=credentials)) channel = connection.channel() # 聲明一個隊列(建立一個隊列) channel.queue_declare(queue='金品沒') def callback(ch, method, properties, body): print("消費者接受到了任務: %r" % body.decode("utf-8")) # int('asdfasdf') # 我告訴rabbitmq服務端，我已經取走了消息 # 回覆方式在這 ch.basic_ack(delivery_tag=method.delivery_tag) # 關閉no_ack，表明給與服務端ack回覆，確認給與回覆 channel.basic_consume(callback,queue='金品沒',no_ack=False) channel.start_consuming()
消息持久化
演示
1.執行生產者，向隊列寫入數據，產生一個新隊列queue 2.重啓服務端，隊列丟失
3.開啓生產者數據持久化後，重啓rabbitmq，隊列不丟失
4.依舊能夠讀取數據
消息的可靠性是RabbitMQ的一大特點，那麼RabbitMQ是如何保證消息可靠性的呢——消息持久化。 爲了保證RabbitMQ在退出或者crash等異常狀況下數據沒有丟失，須要將queue，exchange和Message都持久化。
生產者.py
import pika # 無密碼 # connection = pika.BlockingConnection(pika.ConnectionParameters('123.206.16.61')) # 有密碼 credentials = pika.PlainCredentials("s14","123") connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.119.10',credentials=credentials)) channel = connection.channel() # 聲明一個隊列(建立一個隊列) # 默認此隊列不支持持久化，若是服務掛掉，數據丟失 # durable=True 開啓持久化，必須新開啓一個隊列，本來的隊列已經不支持持久化了 ''' 實現rabbitmq持久化條件 delivery_mode=2 使用durable=True聲明queue是持久化 ''' channel.queue_declare(queue='LOL',durable=True) channel.basic_publish(exchange='', routing_key='LOL', # 消息隊列名稱 body='德瑪西亞萬歲', # 支持數據持久化 properties=pika.BasicProperties( delivery_mode=2,#表明消息是持久的 2 ) ) connection.close()
消費者.py
import pika credentials = pika.PlainCredentials("s14","123") connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.119.10',credentials=credentials)) channel = connection.channel() # 確保隊列持久化 channel.queue_declare(queue='LOL',durable=True) ''' 必須確保給與服務端消息回覆，表明我已經消費了數據，不然數據一直持久化，不會消失 ''' def callback(ch, method, properties, body): print("消費者接受到了任務: %r" % body.decode("utf-8")) # 模擬代碼報錯 # int('asdfasdf') # 此處報錯，沒有給予回覆，保證客戶端掛掉，數據不丟失 # 告訴服務端，我已經取走了數據，不然數據一直存在 ch.basic_ack(delivery_tag=method.delivery_tag) # 關閉no_ack，表明給與回覆確認 channel.basic_consume(callback,queue='LOL',no_ack=False) channel.start_consuming()
Exchange模型
rabbitmq發送消息首先是發給exchange，而後再經過exchange發送消息給隊列（queue）
exchange有四種模式
fanout
exchange將消息發送給和該exchange鏈接的全部queue；也就是所謂的廣播模式；此模式下忽略routing_key；
direct
路由模式，經過routing_key將消息發送給對應的queue; 以下面這句便可設置exchange爲direct模式，只有routing_key爲「black」時纔將其發送到隊列queue_name； channel.queue_bind(exchange=exchange_name,queue=queue_name,routing_key='black')
在上圖中，Q1和Q2能夠綁定同一個key，如綁定routing_key=‘KeySame’，那麼收到routing_key爲KeySame的消息時將會同時發送給Q1和Q2，退化爲廣播模式；
top
topic模式相似於direct模式，只是其中的routing_key變成了一個有「.」分隔的字符串，「.」將字符串分割成幾個單詞，每一個單詞表明一個條件；
headers
headers類型的Exchange不依賴於routing key與binding key的匹配規則來路由消息，而是根據發送的消息內容中的headers屬性進行匹配。
官方教程：http://www.rabbitmq.com/tutorials/tutorial-three-python.html
發佈訂閱和簡單的消息隊列區別在於，發佈訂閱會將消息發送給全部的訂閱者，而消息隊列中的數據被消費一次便消失。因此，RabbitMQ實現發佈和訂閱時，會爲每個訂閱者建立一個隊列，而發佈者發佈消息時，會將消息放置在全部相關隊列中。
# fanout全部的隊列放一份/給某些隊列發
# 傳送消息的模式
# 與exchange有關的模式都發
exchange_type = fanout
消費者_訂閱.py
能夠運行屢次，運行多個消費者，等待消息
import pika
credentials = pika.PlainCredentials("root","123")
connection = pika.BlockingConnection(pika.ConnectionParameters('123.206.16.61',credentials=credentials))
channel = connection.channel()
# exchange='m1',exchange(祕書)的名稱
# exchange_type='fanout' , 祕書工做方式將消息發送給全部的隊列
channel.exchange_declare(exchange='m1',exchange_type='fanout')
# 隨機生成一個隊列
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
# 讓exchange和queque進行綁定.
channel.queue_bind(exchange='m1',queue=queue_name)
def callback(ch, method, properties, body):
print("消費者接受到了任務: %r" % body)
channel.basic_consume(callback,queue=queue_name,no_ack=True)
channel.start_consuming()
生產者_發佈者.py
# -*- coding: utf-8 -*-
# __author__ = "yugo"
import pika
credentials = pika.PlainCredentials("root","123")
connection = pika.BlockingConnection(pika.ConnectionParameters('123.206.16.61',credentials=credentials))
channel = connection.channel()
# 指定exchange
channel.exchange_declare(exchange='m1',exchange_type='fanout')
channel.basic_publish(exchange='m1',
routing_key='',# 這裏再也不指定隊列，由exchange分配,若是是fanout模式，每個隊列放一份
body='haohaio')
connection.close()
實例
1.能夠運行多個消費者，至關於有多個滴滴司機，等待着Exchange同一個電臺發消息 2.運行發佈者，發送消息給Exchange，查看是否給全部的隊列(滴滴司機)發送了消息
關鍵字發佈Exchange
以前事例，發送消息時明確指定某個隊列並向其中發送消息，RabbitMQ還支持根據關鍵字發送，即：隊列綁定關鍵字，發送者將數據根據關鍵字發送到消息exchange，exchange根據 關鍵字 斷定應該將數據發送至指定隊列。
消費者1.py
路由關鍵字是sb,alex
# -*- coding: utf-8 -*-
# __author__ = "maple"
import pika
credentials = pika.PlainCredentials("root","123")
connection = pika.BlockingConnection(pika.ConnectionParameters('123.206.16.61',credentials=credentials))
channel = connection.channel()
# exchange='m1',exchange(祕書)的名稱
# exchange_type='fanout' , 祕書工做方式將消息發送給全部的隊列
channel.exchange_declare(exchange='m2',exchange_type='direct')
# 隨機生成一個隊列,隊列退出時，刪除這個隊列
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
# 讓exchange和queque進行綁定，只要
channel.queue_bind(exchange='m2',queue=queue_name,routing_key='alex')
channel.queue_bind(exchange='m2',queue=queue_name,routing_key='sb')
def callback(ch, method, properties, body):
print("消費者接受到了任務: %r" % body)
channel.basic_consume(callback,queue=queue_name,no_ack=True)
channel.start_consuming()
消費者2.py
路由關鍵字sb
# -*- coding: utf-8 -*-
# __author__ = "maple"
import pika
credentials = pika.PlainCredentials("root","123")
connection = pika.BlockingConnection(pika.ConnectionParameters('123.206.16.61',credentials=credentials))
channel = connection.channel()
# exchange='m1',exchange(祕書)的名稱
# exchange_type='fanout' , 祕書工做方式將消息發送給全部的隊列
channel.exchange_declare(exchange='m2',exchange_type='direct')
# 隨機生成一個隊列
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
# 讓exchange和queque進行綁定.
channel.queue_bind(exchange='m2',queue=queue_name,routing_key='sb')
def callback(ch, method, properties, body):
print("消費者接受到了任務: %r" % body)
channel.basic_consume(callback,queue=queue_name,no_ack=True)
channel.start_consuming()
生產者.py
發送消息給匹配的路由，sb或者alex
# -*- coding: utf-8 -*-
# __author__ = "yugo"
import pika
credentials = pika.PlainCredentials("root","123")
connection = pika.BlockingConnection(pika.ConnectionParameters('123.206.16.61',credentials=credentials))
channel = connection.channel()
# 路由模式的交換機會發送給綁定的key和routing_key匹配的隊列
channel.exchange_declare(exchange='m2',exchange_type='direct')
# 發送消息，給有關sb的路由關鍵字
channel.basic_publish(exchange='m2',
routing_key='sb',
body='aaaalexlaolelaodi')
connection.close()
RPC之遠程過程調用
將一個函數運行在遠程計算機上而且等待獲取那裏的結果，這個稱做遠程過程調用（Remote Procedure Call）或者 RPC。
RPC是一個計算機通訊協議。
比喻
將計算機服務運行理解爲廚師作飯，廚師想作一個小蔥拌豆腐，廚師須要洗小蔥、切豆腐、調汁、涼拌。他一我的完成全部的事，如同古老的集中式應用，一臺計算機作全部的事。
製做小蔥拌豆腐{
廚師>洗小蔥>切豆腐>涼拌
}
rpc應用場景
而現在，飯店作大了，有錢了，專職分工來幹活，再也不是廚師單打獨鬥，備菜師傅準備小蔥、豆腐，切菜師傅切小蔥、豆腐，廚師只負責調味，完成食品。
製做小蔥拌豆腐{
備菜師>洗菜
切菜師>切菜
廚師>調味
}
此時一件事好多人在作，廚師就得和其餘人溝通，通知備菜、洗菜師傅的這個動做就是遠程過程調用（RPC）。
這個過程在計算機系統中，一個電商的下單過程，涉及物流、支付、庫存、紅包等多個系統，多個系統又在多個服務器上，由不一樣的技術團隊負責，整個下單過程，須要全部團隊進行遠程調用。
下單{
庫存>減小庫存
支付>扣款
紅包>減免紅包
物流>生成訂單
}
到底什麼是rpc
rpc指的是在計算機A上的進程，調用另一臺計算機B的進程，A上的進程被掛起，B上的被調用進程開始執行後，產生返回值給A，A繼續執行。
調用方能夠經過參數將信息傳遞給被調用方，然後經過返回結果獲得信息，這個過程對於開發人員來講是透明的。
如同廚師同樣，服務員把菜單給後廚，廚師告訴洗菜人，備菜人，開始工做，完成工做後，整個過程對於服務員是透明的，他徹底不用管後廚是怎麼把菜作好的。
因爲服務在不一樣的機器上，遠程調用必經網絡通訊，調用服務必須寫一坨網絡通訊代碼，很容易出錯且很複雜，所以就出現了RPC框架。
阿里巴巴的 Dubbo java
新浪的 Motan java
谷歌的 gRPC 多語言
Apache thrift 多語言
rpc封裝了數據的序列化，反序列化，以及傳輸協議
python實現RPC
利用RabbitMQ構建一個RPC系統，包含了客戶端和RPC服務器，依舊使用pika模塊
Callback queue 回調隊列
一個客戶端向服務器發送請求，服務器端處理請求後，將其處理結果保存在一個存儲體中。而客戶端爲了得到處理結果，那麼客戶在向服務器發送請求時，同時發送一個回調隊列地址reply_to。
Correlation id 關聯標識
一個客戶端可能會發送多個請求給服務器，當服務器處理完後，客戶端沒法辨別在回調隊列中的響應具體和那個請求時對應的。爲了處理這種狀況，客戶端在發送每一個請求時，同時會附帶一個獨有correlation_id屬性，這樣客戶端在回調隊列中根據correlation_id字段的值就能夠分辨此響應屬於哪一個請求。
客戶端發送請求：某個應用將請求信息交給客戶端，而後客戶端發送RPC請求，在發送RPC請求到RPC請求隊列時，客戶端至少發送帶有reply_to以及correlation_id兩個屬性的信息
服務器端工做流： 等待接受客戶端發來RPC請求，當請求出現的時候，服務器從RPC請求隊列中取出請求，而後處理後，將響應發送到reply_to指定的回調隊列中
客戶端接受處理結果： 客戶端等待回調隊列中出現響應，當響應出現時，它會根據響應中correlation_id字段的值，將其返回給對應的應用
過程
1.啓動rpc客戶端，等待接收數據到來，來了以後就進行處理，再將結果丟進隊列 2.啓動rpc服務端，發起請求
rpc_server.py
import pika
import uuid
class FibonacciRpcClient(object):
def __init__(self):
# 客戶端啓動時，建立回調隊列，會開啓會話用於發送RPC請求以及接受響應
# 創建鏈接，指定服務器的ip地址
self.connection = pika.BlockingConnection(pika.ConnectionParameters(
host='192.168.119.10'))
# 創建一個會話，每一個channel表明一個會話任務
self.channel = self.connection.channel()
# 聲明回調隊列，再次聲明的緣由是，服務器和客戶端可能前後開啓，該聲明是冪等的，屢次聲明，但只生效一次
#exclusive=True 參數是指只對首次聲明它的鏈接可見
#exclusive=True 會在鏈接斷開的時候，自動刪除
result = self.channel.queue_declare(exclusive=True)
# 將次隊列指定爲當前客戶端的回調隊列
self.callback_queue = result.method.queue
# 客戶端訂閱回調隊列，當回調隊列中有響應時，調用`on_response`方法對響應進行處理;
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
# 發出RPC請求
# 例如這裏服務端就是一個切菜師傅，菜切好了，須要傳遞給洗菜師傅，這個過程是發送rpc請求
def call(self, n):
# 初始化 response
self.response = None
# 生成correlation_id 關聯標識，經過python的uuid庫，生成全局惟一標識ID，保證時間空間惟一性
self.corr_id = str(uuid.uuid4())
# 發送RPC請求內容到RPC請求隊列`s14rpc`，同時發送的還有`reply_to`和`correlation_id`
self.channel.basic_publish(exchange='',
routing_key='s14rpc',
properties=pika.BasicProperties(
reply_to=self.callback_queue,
correlation_id=self.corr_id,
),
body=str(n))
while self.response is None:
self.connection.process_data_events()
return int(self.response)
# 創建客戶端
fibonacci_rpc = FibonacciRpcClient()
# 發送RPC請求，丟進rpc隊列，等待客戶端處理完畢，給與響應
print("發送了請求sum(99)")
response = fibonacci_rpc.call(99)
print("獲得遠程結果響應%r" % response)
rpc_client.py
import pika# 創建鏈接，服務器地址爲localhost，可指定ip地址connection = pika.BlockingConnection(pika.ConnectionParameters( host='192.168.119.10'))# 創建會話channel = connection.channel()# 聲明RPC請求隊列channel.queue_declare(queue='s14rpc')# 模擬一個進程，例如切菜師傅，等着洗菜師傅傳遞數據def sum(n): n+=100 return n# 對RPC請求隊列中的請求進行處理def on_request(ch, method, props, body): print(body,type(body)) n = int(body) print(" 正在處理sum(%s)" % n) # 調用數據處理方法 response = sum(n) # 將處理結果(響應)發送到回調隊列 ch.basic_publish(exchange='', # reply_to表明回覆目標 routing_key=props.reply_to, # correlation_id（關聯標識）：用來將RPC的響應和請求關聯起來。 properties=pika.BasicProperties(correlation_id= \ props.correlation_id), body=str(response)) ch.basic_ack(delivery_tag=method.delivery_tag)# 負載均衡，同一時刻發送給該服務器的請求不超過一個channel.basic_qos(prefetch_count=1)channel.basic_consume(on_request, queue='s14rpc')print("等待接收rpc請求")#開始消費channel.start_consuming()
- 1. RabbitMq消息隊列
- 2. 消息隊列----RabbitMQ
- 3. RabbitMQ---消息隊列
- 4. RabbitMQ消息隊列
- 5. Rabbitmq 消息隊列
- 6. 消息隊列-RabbitMQ
- 7. 消息隊列 RabbitMQ
- 8. 消息隊列——RabbitMQ
- 9. rabbitMQ 消息隊列
- 10. 消息隊列rabbitmq
- 更多相關文章...
- • C# 隊列（Queue） - C#教程
- • HTTP 消息結構 - HTTP 教程
- • 再有人問你分佈式事務，把這篇扔給他
- • 算法總結-廣度優先算法
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. RabbitMq消息隊列
- 2. 消息隊列----RabbitMQ
- 3. RabbitMQ---消息隊列
- 4. RabbitMQ消息隊列
- 5. Rabbitmq 消息隊列
- 6. 消息隊列-RabbitMQ
- 7. 消息隊列 RabbitMQ
- 8. 消息隊列——RabbitMQ
- 9. rabbitMQ 消息隊列
- 10. 消息隊列rabbitmq