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()