256      阿裏雲  技術社區[雲棲]

Python 收發消息__HTTP 接入(簡單)_消息隊列 MQ-阿裏雲

本文主要描述如何在 Python 環境下使用 HTTP 協議收發 MQ 消息。

運行環境準備

用 HTTP 協議發送或者接收消息 ，請完成以下環境準備工作。

Windows

1. 從 Python 官網（https://www.python.org/downloads/windows/) 下載並安裝 Python 2.7。

2. 打開 Windows 終端窗口（dos界麵)，輸入 python 命令，檢查 Python 是否安裝成功， 如下圖所示。

   

Linux/Unix

您可以在 Linux 終端輸入 python 指令確認是否有預裝。如果有執行信息，則說明本台機器已經預裝。如果沒有，請按以下步驟進行安裝：

1. 從 Python 官網下載 Python2.7 Linux 版本安裝包：https://www.python.org/download/

2. 在 Python2.7 安裝包下載保存目錄下，運行指令進行解壓，例如：tar –xzf python-2.7.11.tgz

3. 進入2.7.11目錄，執行./configure 指令。此步驟主要是用於生成 makefile 文件。

4. 執行 make 指令進行實際編譯。

5. 執行 make install 指令進行安裝操作。

6. 安裝完畢後執行 python 命令檢查安裝是否成功，如下圖：

   

運行示例代碼

按照以下步驟運行示例代碼。

Windows

1. 將下文具體示例程序小節所提供的配置文件以及示例程序拷貝到本地，並保存為 .py 文件，如下圖：

   

2. 根據下文示例代碼裏的說明修改 user.properties 文件中的相關字段。

3. 將終端窗口切換到保存 .py 的文件目錄，執行保存的 python 文件，如下圖:

Linux/Unix

步驟參見 Windows。執行結果參照下圖：

具體示例程序

您可以參考以下示例程序測試消息收發功能。

1.配置文件

您需要設置配置文件（user.properties）的相關內容，具體請參考申請 MQ 資源 。

[property]
#您在控製台創建的Topic
Topic=xxx
#公測集群URL
URL=https://publictest-rest.ons.aliyun.com
#阿裏雲官網身份驗證訪問碼
Ak=xxx
#阿裏雲身份驗證密鑰
Sk=xxx
#MQ控製台創建的Producer ID
ProducerID=xxx
#MQ控製台創建的Consumer ID
ConsumerID=xxx

說明：URL中的 Key，Tag 以及 POST Content-Type 沒有任何的限製，隻要確保 Key 和 Tag 相同唯一即可，可以放在 user.properties 裏麵。

2.發送消息示例程序

通過 HTTP 協議發送消息，請參考以下示例代碼。

#encoding:utf-8
import ConfigParser
import hashlib
import httplib
import time
from urlparse import urlparse
from Util import parseURL,calSignature
"""
消息發布者
"""
class HttpProducer(object):
    def __init__(self):
        """簽名值"""
        self.signature = "Signature"
        """ProducerID"""
        self.producerid = "ProducerID"
        """消息主題"""
        self.topic = "Topic"
        """訪問碼"""
        self.ak = "AccessKey"
        """配置文件解析器"""
        self.cf = ConfigParser.ConfigParser()
        """MD5對象"""
        self.md5 = hashlib.md5()
    """
    發布消息主流程
    """
    def process(self):
        """讀取配置文件"""
        self.cf.read("user.properties")
        """讀取消息主題"""
        topic = self.cf.get("property","Topic")
        """存儲消息URL路徑"""
        url = self.cf.get("property","URL")
        """訪問碼"""
        ak = self.cf.get("property","Ak")
        """密鑰"""
        sk = self.cf.get("property","Sk")
        """Producer ID"""
        pid = self.cf.get("property","ProducerID")
        """HTTP請求主體內容"""
        content = U"中文".encode('utf-8')
        """分隔符"""
        newline = "n"
        """獲取URL域名地址"""
        urlname = urlparse(url).hostname
        """根據HTPP主體內容計算MD5值"""
        self.md5.update(content)
        """建立HTTP連接對象"""
        conn = httplib.HTTPConnection(parseURL(urlname))
        try:
            for index in range(0,100):
                """時間戳"""
                date = repr(int(time.time() * 1000))[0:13]
                """構造簽名字符串"""
                signString = str(topic + newline + pid + newline + self.md5.hexdigest() + newline + date)
                """計算簽名"""
                sign = calSignature(signString,sk)
                """內容類型"""
                contentFlag ="Content-type"
                """HTTP請求頭部對象"""
                headers = {
                    self.signature : sign,
                    self.ak : ak,
                    self.producerid : pid,
                    contentFlag : "text/html;charset=UTF-8"
            }
                """開始發送HTTP請求消息"""
                conn.request(method="POST",url="/message/?topic="+topic+"&time="+date+"&tag=http&key=http",
                             body=content,
                             headers=headers)
                """獲取HTTP應答消息"""
                response = conn.getresponse()
                """讀取HTTP應答內容"""
                msg = response.read()
                print "response:"+msg
        except Exception,e:
            print e
        finally:
            conn.close()
"""流程入口"""
if __name__ == '__main__':
    """創建消息發布者"""
    producer = HttpProducer()
    """開啟消息發布者"""
    producer.process()

3. 接收消息示例程序

#encoding:utf-8
import ConfigParser
import httplib
import json
import time
from urlparse import urlparse
from Util import parseURL,calSignature
"""
消息訂閱者
"""
class HttpConsumer(object):
        def __init__(self):
        """簽名字段"""
        self.signatue = "Signature"
        """Consumer ID"""
        self.consumerid = "ConsumerID"
        """消費主題"""
        self.topic = "Topic"
        """訪問碼"""
        self.ak = "AccessKey"
        """配置文件解析器"""
        self.cf = ConfigParser.ConfigParser()
    """
    訂閱消息流程
    """
    def process(self):
        """開始讀取配置文件"""
        self.cf.read("user.properties")
        """讀取主題"""
        topic = self.cf.get("property", "Topic")
        """存儲消息的URL路徑"""
        url = self.cf.get("property", "URL")
        """訪問碼"""
        ak = self.cf.get("property", "Ak")
        """密鑰"""
        sk = self.cf.get("property", "Sk")
        """Consumer ID"""
        cid = self.cf.get("property", "ConsumerID")
        newline = "n"
        """獲取URL主機域名地址"""
        urlname = urlparse(url).hostname
        """連接存儲消息的服務器"""
        conn = httplib.HTTPConnection(parseURL(urlname))
        while True:
            try:
                """時間戳"""
                date = repr(int(time.time() * 1000))[0:13]
                """構造簽名字符串"""
                signString = topic + newline + cid + newline + date
                """計算簽名值"""
                sign = calSignature(signString,sk)
                """請求消息HTTP頭部"""
                headers = {
                    self.signatue : sign,
                    self.ak : ak,
                    self.consumerid : cid
                    }
                """開始發送獲取消息的HTTP請求"""
            conn.request(method="GET",url=url+"/message/?topic="+topic+"&time="+date+"&num=32",headers=headers)
                """獲取HTTP應答消息"""
                response = conn.getresponse()
                """驗證應答消息狀態值"""
                if response.status != 200:
                    continue
                """從應答消息中讀取實際的消息內容"""
                msg = response.read()
                """將實際的消費消息進行解碼"""
                messages = json.loads(msg)
                if len(messages) == 0:
                    time.sleep(2)
                    continue
                """依次獲取每條消費消息"""
                for message in messages:
                    """計算時間戳"""
                    date = repr(int(time.time() * 1000))[0:13]
                    """構建刪除消費消息URL路徑"""
                    delUrl = url + "/message/?msgHandle="+message['msgHandle'] + "&topic="+topic+"&time="+date
                    """構造簽名字符串"""
                    signString = topic + newline + cid + newline + message['msgHandle'] + newline + date
                    """進行簽名"""
                    sign = calSignature(signString,sk)
                    """構造刪除消費消息HTTP頭部"""
                       delheaders = {
                           self.signatue : sign,
                        self.ak : ak,
                        self.consumerid : cid,
                }
                    """發送刪除消息請求"""
                    conn.request(method="DELETE", url=delUrl, headers=delheaders)
                    """獲取請求應答"""
                    response = conn.getresponse()
                    """讀取應答內容"""
                    msg = response.read()
                    print "delete msg:"+msg
            except Exception,e:
                print e
        conn.close()
"""啟動入口"""
if __name__ == '__main__':
    """構造消息訂閱者"""
    consumer = HttpConsumer()
    """開始啟動消息訂閱者"""
    consumer.process()

#encoding:utf-8
import socket
import hmac
from hashlib import sha1
"""
解析URL
"""
def parseURL(url):
    iplist = socket.gethostbyname_ex(url)
    if len(iplist) == 0:
        return None
    ips = iplist[2]
    if len(ips) == 0:
        return None
    return ips[0]
"""
認證簽名
"""
def calSignature(signString, sk):
    mac = hmac.new(sk, signString, sha1)
    return mac.digest().encode('base64').rstrip()