## RabbitMQ - windows下安裝 - 安裝Erlan 全部勾選 - 安裝RabbitMQ 全部勾選 - cd C:\Program Files\RabbitMQ Server\rabbitmq_server-3.7.14\sbin 目錄 - rabbitmq-plugins.bat enable rabbitmq\_management 開啟網頁版控制臺 - rabbitmq-server 啟動RabbitMQ服務 - 在本地瀏覽器中輸入：localhost:15672訪問RabbitMQ的后臺管理頁面(初始化用戶名和密碼都是guest) >rabbitmqctl.bat list_queues #查看隊列里的任務 >rabbitmqctl add_user admin 123456 #增加訪問用戶，默認用戶guest只能本地訪問。 >rabbitmqctl set_user_tags admin administrator 設置角色： >rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*" 設置默認vhost（“/”）訪問權限 ` [啟動rabbitmq，提示ERROR: node with name "rabbit" already running on "U57..."]` - 在任務管理器中結束進程 epmd.exe erl.exe - 在再sbin命令，rabbitmq-server start - linux安裝 - 安裝Erlang >yum install erlang - 安裝RabbitMQ >wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitmq-server-3.6.6-1.el7.noarch.rpm >yum install rabbitmq-server-3.6.6-1.el7.noarch.rpm systemctl startr abbitmq-server.service systemctl status rabbitmq-server.service ``` # 查看當前所有用戶 rabbitmqctl list_users # 查看默認guest用戶的權限 rabbitmqctl list_user_permissions guest # 由于RabbitMQ默認的賬號用戶名和密碼都是guest。為了安全起見, 先刪掉默認用戶 rabbitmqctl delete_user guest # 添加新用戶 rabbitmqctl add_user admin lssadmin # 設置用戶tag rabbitmqctl set_user_tags admin administrator # 賦予用戶默認vhost的全部操作權限 rabbitmqctl set_permissions -p / admin ".*" ".*" ".*" # 查看用戶的權限 rabbitmqctl list_user_permissions admin ``` - 端口放行 >firewall-cmd --zone=public --add-port=5672/tcp --permanent >firewall-cmd --zone=public --add-port=15672/tcp --permanent >firewall-cmd --reload ## RabbitMQ 默認不返回結果,默認不是隊列持久化,服務關閉隊列消失,多消費者,默認是公平分發 >pip install pika==0.12.0 - 消費者 receive.py ~~~ import pika # 注意,guest用戶只是被容許從localhost訪問 user_pwd = pika.PlainCredentials('guest', 'guest') # 創建連接 s_conn = pika.BlockingConnection(pika.ConnectionParameters('localhost', credentials=user_pwd)) # 在連接上創建一個頻道 chan = s_conn.channel() # 聲明一個隊列，生產者和消費者都要聲明一個相同的隊列，用來防止萬一某一方掛了，另一方能正常運行 chan.queue_declare(queue='hello') # 定義一個回調函數，用來接收生產者發送的消息 def callback(ch,method,properties,body): print("[消費者] recv %s" % body) ch.basic_ack(delivery_tag=method.delivery_tag) # 手動對消息進行確認,確定后列隊會自動刪除消息 # 只有在consumer處理并確認了上一個message后才分配新的message給他 ,否則分給另一個空閑的consumer, 避免積壓任務 chan.basic_qos(prefetch_count=1) chan.basic_consume( callback, # 調用回調函數 'hello', # 指定取消息的隊列名 no_ack=False, # 該參數默認為false, True自動確認 ) # 開始循環取消息 chan.start_consuming() ~~~ - 生產者 sender.py ~~~ import pika # 注意,guest用戶只是被容許從localhost訪問 user_pwd = pika.PlainCredentials('guest', 'guest') # 創建連接 s_conn = pika.BlockingConnection(pika.ConnectionParameters('localhost', credentials=user_pwd)) # 在連接上創建一個頻道 chan = s_conn.channel() #聲明一個隊列，生產者和消費者都要聲明一個相同的隊列，用來防止萬一某一方掛了，另一方能正常運行 chan.queue_declare(queue='hello') # 交換機, 路由鍵，寫明將消息發往哪個隊列，本例是將消息發往隊列hello,生產者要發送的消息 chan.basic_publish(exchange='', routing_key='hello', body='hello world') # 當生產者發送完消息后，可選擇關閉連接 s_conn.close() ~~~ - mq_rpc mq_rpc_server.py ~~~ import pika import time # 鏈接socket connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() # 生成rpc queue 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) # 收到消息就調用 # ch 管道內存對象地址 # method 消息發給哪個queue # props 返回給消費的返回參數 # body數據對象 def on_request(ch, method, props, body): n = int(body) print(" [.] fib(%s)" % n) # 調用斐波那契函數 傳入結果 response = fib(n) ch.basic_publish(exchange='', # 生產端隨機生成的queue routing_key=props.reply_to, # 獲取UUID唯一 字符串數值 properties=pika.BasicProperties(correlation_id = \ props.correlation_id), # 消息返回給生產端 body=str(response)) # 確保任務完成 ch.basic_ack(delivery_tag=method.delivery_tag) # rpc_queue收到消息:調用on_request回調函數 # queue='rpc_queue'從rpc內收 channel.basic_consume(on_request, queue='rpc_queue') print(" [x] Awaiting RPC requests") channel.start_consuming() ~~~ mq_rpc_client.py ~~~ import pika import uuid import time # 斐波那契數列 前兩個數相加依次排列 class FibonacciRpcClient(object): def __init__(self): # 鏈接遠程 self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) self.channel = self.connection.channel() # 生成隨機queue result = self.channel.queue_declare(exclusive=True) # 隨機取queue名字，發給消費端 self.callback_queue = result.method.queue # self.on_response 回調函數:只要收到消息就調用這個函數。 # 聲明收到消息后就 收queue=self.callback_queue內的消息 self.channel.basic_consume(self.on_response, no_ack=True, queue=self.callback_queue) # 收到消息就調用 # ch 管道內存對象地址 # method 消息發給哪個queue # body數據對象 def on_response(self, ch, method, props, body): # 判斷本機生成的ID 與 生產端發過來的ID是否相等 if self.corr_id == props.correlation_id: # 將body值 賦值給self.response self.response = body def call(self, n): # 賦值變量，一個循環值 self.response = None #　隨機一次唯一的字符串 self.corr_id = str(uuid.uuid4()) # routing_key='rpc_queue' 發一個消息到rpc_queue內 self.channel.basic_publish(exchange='', routing_key='rpc_queue', properties=pika.BasicProperties( # 執行命令之后結果返回給self.callaback_queue這個隊列中 reply_to=self.callback_queue, # 生成UUID 發送給消費端 correlation_id=self.corr_id, ), # 發的消息，必須傳入字符串，不能傳數字 body=str(n)) # 沒有數據就循環收 while self.response is None: # 非阻塞版的start_consuming() # 沒有消息不阻塞 self.connection.process_data_events() print("no msg...") time.sleep(0.5) return self.response #　實例化 fibonacci_rpc = FibonacciRpcClient() print(" [x] Requesting fib(30)") response = fibonacci_rpc.call(6) print(" [.] Got %r" % response) ~~~