阿里云-消息队列-HTTP接入手册-D

1、消息队列HTTP 接入(简单)消息队列/HTTP 接入(简单)消息队列/HTTP 接入(简单) PAGE 28 PAGE 28HTTP 接入(简单)MQ HTTP 接入简介HTTPMQHTTP 协议为载体，通过使用 HTTP 的常用接口来对消息队列进行增删查。HTTP 接 入 优 势HTTP接入主要有以下几大优势：HTTP解决异构网络环境下的服务相互访问屏障，对于没有提供相关操作消息队列SDK的环境中，使用HTTP方式接入更为方便；消息队列HTTP接入方式在使用上简单，上手快。HTTP 接入应用场景HTTP 接入方式应用的场景主要依托于客户的业务场景，假设客户的业务场景或者部分模块是基于 HT

2、TP 协议并且需要通信服务，就可以使用 MQPythonPHPHTTP 域名目前，HTTP 各 Region 域名如下。Region公网/内网域名公网测试公网publictest- 华北2公网beijing-rest- 华北2内网beijing-rest- 华东1公网hangzhou-rest- 华东1内网hangzhou-rest- 华南1内网shenzhen-rest- 华南1公网shenzhen-rest- 华东2内网shanghai-rest- 华东2公网shanghai-rest- MQ HTTP 协议规范本文详细介绍了 HTTP 协议相关的规范。域名字段说明URL HYPERLIN

3、K http:/domain/message/ http:/domain/message/字段说明URL HYPERLINK http:/domain/message/ http:/domain/message/MethodPOSTHEADERAccessKey : 阿里云身份验证;Signature 签 名 : (signString=topic+n+pid+n+md5(Body)+n+time;Producer ID : 消息发布者IDParamTopic Time 时间戳（自1970-01-01, 00:00:00 GMT经历的毫秒数，如果MQ收到时间戳已经过了15s，那么会返回403）

4、Body消息体内容为UTF-8编码消息发送-ResponseStatus code说明Body（json格式）201消息添加成功msgId:0A021F7300002A9 F0000000006531D6F,sendStatus:SEND_OK400请求失败code:BODY_TOO_BIG,in fo:Message body size exceeds the upper limit of 64KB403鉴权失败408请求超时消息接收-Request字段说明URL HYPERLINK http:/domain/message/ http:/domain/message/MethodGETHE

5、ADERAccessKey : 阿里云AccessKeySignature 签 名 : (signString=topic+n+ cid+n+time) Consumer ID : 消息订阅者IDParamTopic Time 时间戳（自1970-01-01, 00:00:00 GMT经历的毫秒数，如果MQ收到时间戳已经过了15s，那么会返回403）消息接收-ResponseStatus code说明Body（json格式）200消息读取成功，会返回至多32条消息，如果没有消息会循环遍历事件直到有消息到来（poll模式），持续时间 30sbody:HelloMQ,bornTi me:14189

6、73464204,msgHandle:X1BFTkRJTkd NU0dfXyVSRVRSWSUkbG9uZ2ppJENJRF9sb25namlfdGxv bmdqaQ=, msgId:0A021F7300002A9F 000000000647076D,reconsumeTimes:1400请求失败code:TOPIC_NOT_EXIST,info:topic not exist403鉴权失败408请求超时消息删除-Request字段说明URL HYPERLINK http:/domain/message/ http:/domain/message/MethodDELETEHEADERAcces

7、sKey : 阿 里 云 AccessKey Signature 签 名 : (signString=topic+n+cid+n+msgHandl e+n+timeConsumerId : 消息订阅者IDParamTopic Time 时间戳（自1970-01-01, 00:00:00 GMT经历的毫秒数，如果MQ收到时间戳已经过了15s，那么会返回403） msgHandle 消息句柄消息删除-ResponseStatus code说明Body（json格式）204消息删除成功，无需返回内容400请求失败code:TOPIC_NOT_EXIST,info:topic not exist403

8、鉴权失败408请求超时Java 收发消息本文主要描述如何在 Java 环境下使用 HTTP 协议收发 MQ 消息。准备环境org.eclipse.jettyorg.eclipse.jettyjetty-client9.3.4.RC1com.aliyun.openservicesons-client1.1.11运行代码配置（perties）您需要设置配置文件(perties)的相关内容，具体请参考申请 MQ 资源 。propertyproperty#您在控制台创建的Topic Topic=xxx#公测url HYPERLINK / URL=#阿里云身份验证码Ak=xxx#阿里云身份验证密钥Sk=

9、xxx#MQ控制台创建的Producer ID ProducerID=xxx#MQ控制台创建的Consumer ID ConsumerID=xxx说明：URLKey，TagPOST Content-TypeKeyTpertiesHTTPpackage com.aliyun.openservice.ons.http.demo; import java.nio.charset.Charset;package com.aliyun.openservice.ons.http.demo; import java.nio.charset.Charset;import java.util.Date;impo

10、rt java.util.Properties;import org.eclipse.jetty.client.HttpClient;import org.eclipse.jetty.client.api.ContentProvider; import org.eclipse.jetty.client.api.ContentResponse; import org.eclipse.jetty.client.api.Request;import org.eclipse.jetty.client.util.StringContentProvider; import com.aliyun.opens

11、ervices.ons.api.impl.authority.AuthUtil;public class HttpProducer public static String SIGNATURE=Signature; public static String NUM=num;public static String CONSUMERID=ConsumerID; public static String PRODUCERID=ProducerID;public static String TIMEOUT=timeout; public static String TOPIC=Topic; publ

12、ic static String AK=AccessKey; public static String BODY=body;public static String MSGHANDLE=msgHandle; public static String TIME=time;public static void main(String args) throws Exception HttpClient httpClient=new HttpClient(); httpClient.setMaxConnectionsPerDestination(1); httpClient.start();Prope

13、rties properties=new Properties(); properties.load(HttpProducer.class.getClassLoader().getResourceAsStream(perties); String topic=properties.getProperty(Topic); /请在perties配置您的TopicString HYPERLINK / url=properties.getProperty(URL);/公测集群配置为/String ak=properties.getProperty(Ak);/请在perties配置您的Ak String

14、 sk=properties.getProperty(Sk);/请在perties配置您的SkString pid=properties.getProperty(ProducerID);/请在perties配置您的Producer ID String date=String.valueOf(new Date().getTime();String sign=null;String body=hello ons http; String NEWLINE=n;String signString;for (int i = 0; i 10; i+) date=String.valueOf(new Dat

15、e().getTime();Request req=httpClient.POST(url+message/?topic=+topic+&time=+date+&tag=http+&key=http); ContentProvider content=new StringContentProvider(body);req.content(content); signString=topic+NEWLINE+pid+NEWLINE+MD5.getInstance().getMD5String(body)+NEWLINE+date; System.out.println(signString);s

16、ign=AuthUtil.calSignature(signString.getBytes(Charset.forName(UTF-8), sk); req.header(SIGNATURE, sign);req.header(AK, ak); req.header(PRODUCERID, pid); ContentResponse response; response=req.send();System.out.println(send msg:+response.getStatus()+response.getContentAsString();HTTP接收消息示例代码package co

17、m.aliyun.openservice.ons.http.demo; import java.nio.charset.Charset;package com.aliyun.openservice.ons.http.demo; import java.nio.charset.Charset;import java.util.Date;import java.util.List; import java.util.Properties;import org.eclipse.jetty.client.HttpClient;import org.eclipse.jetty.client.api.Co

18、ntentProvider; import org.eclipse.jetty.client.api.ContentResponse; import org.eclipse.jetty.client.api.Request;import org.eclipse.jetty.client.util.StringContentProvider; import org.eclipse.jetty.http.HttpMethod;import com.alibaba.fastjson.JSON;import com.aliyun.openservice.ons.mqtt.demo.MqttProduc

19、er; import com.aliyun.openservices.ons.api.impl.authority.AuthUtil;public class HttpConsumer public static String SIGNATURE=Signature; public static String NUM=num;public static String CONSUMERID=ConsumerID; public static String PRODUCERID=ProducerID;public static String TIMEOUT=timeout; public stat

20、ic String TOPIC=Topic; public static String AK=AccessKey; public static String BODY=body;public static String MSGHANDLE=msgHandle; public static String TIME=time;public static void main(String args) throws Exception HttpClient httpClient=new HttpClient(); httpClient.setMaxConnectionsPerDestination(1

21、); httpClient.start();Properties properties=new Properties(); properties.load(HttpConsumer.class.getClassLoader().getResourceAsStream(perties); String topic=properties.getProperty(Topic); /请在perties配置您的topicString HYPERLINK / url=properties.getProperty(URL);/公测集群配置为/String ak=properties.getProperty(

22、Ak);/请在perties配置您的Ak String sk=properties.getProperty(Sk);/请在perties配置您的SkString cid=properties.getProperty(ConsumerID);/请在perties配置您的Consumer ID String date=String.valueOf(new Date().getTime();String sign=null; String NEWLINE=n; String signString;System.out.println(NEWLINE+NEWLINE); while (true) tr

23、y date=String.valueOf(new Date().getTime();Request req=httpClient.POST(url+message/?topic=+topic+&time=+date+&num=+32); req.method(HttpMethod.GET);ContentResponse response; signString=topic+NEWLINE+cid+NEWLINE+date;sign=AuthUtil.calSignature(signString.getBytes(Charset.forName(UTF-8), sk); req.heade

24、r(SIGNATURE, sign);req.header(AK, ak); req.header(CONSUMERID, cid);long start=System.currentTimeMillis(); response=req.send();System.out.println(get cost:+(System.currentTimeMillis()-start)/1000+ +response.getStatus()+ +response.getContentAsString(); List list = null;if (response.getContentAsString(

25、)!=null&!response.getContentAsString().isEmpty() list=JSON.parseArray(response.getContentAsString(), SimpleMessage.class);if(list=null|list.size()=0) Thread.sleep(100); continue;System.out.println(size is :+list.size();for (SimpleMessage simpleMessage : list) for (SimpleMessage simpleMessage : list)

26、 date=String.valueOf(new Date().getTime();System.out.println(receive msg:+simpleMessage.getBody()+ born time +simpleMessage.getBornTime(); req=httpClient.POST(url+message/?msgHandle=+simpleMessage.getMsgHandle()+&topic=+topic+&time=+ date);req.method(HttpMethod.DELETE); signString=topic+NEWLINE+cid+

27、NEWLINE+simpleMessage.getMsgHandle()+NEWLINE+date; sign=AuthUtil.calSignature(signString.getBytes(Charset.forName(UTF-8), sk); req.header(SIGNATURE, sign);req.header(AK, ak); req.header(CONSUMERID, cid); response=req.send();System.out.println(delete msg:+response.toString();Thread.sleep(100); catch

28、(Exception e) e.printStackTrace();HTTP示例程序工具类package com.aliyun.openservice.ons.http.demo; public class SimpleMessage package com.aliyun.openservice.ons.http.demo; public class SimpleMessage private String body;private String msgId; private String bornTime; private String msgHandle; privateintrecons

29、umeTimes; private Stringtag;public void setTag(String tag) this.tag = tag;public String getTag() return tag;public int getReconsumeTimes() return reconsumeTimes;public void setReconsumeTimes(int reconsumeTimes) this.reconsumeTimes = reconsumeTimes;public void setMsgHandle(String msgHandle) this.msgH

30、andle = msgHandle;public String getMsgHandle() return msgHandle;return msgHandle;public String getBody() return body;public void setBody(String body) this.body = body;public String getMsgId() return msgId;public void setMsgId(String msgId) this.msgId = msgId;public String getBornTime() return bornTi

31、me;public void setBornTime(String bornTime) this.bornTime = bornTime;package com.aliyun.openservice.ons.http.demo;import java.io.UnsupportedEncodingException; package com.aliyun.openservice.ons.http.demo;import java.io.UnsupportedEncodingException; import java.nio.charset.Charset;import java.securit

32、y.MessageDigest; import java.sql.SQLException; import java.util.Date;import java.util.HashMap; import java.util.Map;import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.locks.ReentrantLock;import org.slf4j.LoggerFactory; public class MD5 private static final org.slf4j.Logger lo

33、g = LoggerFactory.getLogger(MD5.class);private static char digits = 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, a, b, c, d, e, f ;privatestaticMaprDigits=newHashMap(16); staticfor (int i = 0; i digits.length; +i) rDigits.put(digitsi, i);private static MD5 me = new MD5(); private MessageDigest mHasher;private fina

34、l ReentrantLock opLock = new ReentrantLock();private MD5() try this.mHasher = MessageDigest.getInstance(md5); catch (Exception e) throw new RuntimeException(e);public static MD5 getInstance() return me;public String getMD5String(String content) return this.bytes2string(this.hash(content);public Stri

35、ng getMD5String(byte content) return this.bytes2string(this.hash(content);public byte getMD5Bytes(byte content) return this.hash(content);public byte hash(String str) this.opLock.lock();try byte bt = this.mHasher.digest(str.getBytes(utf-8); if (null = bt | bt.length != 16) throw new IllegalArgumentE

36、xception(md5 need);return bt; catch (UnsupportedEncodingException e) throw new RuntimeException(unsupported utf-8 encoding, e); finally this.opLock.unlock();public byte hash(byte data) this.opLock.lock();try byte bt = this.mHasher.digest(data); if (null = bt | bt.length != 16) throw new IllegalArgum

37、entException(md5 need);return bt; finally this.opLock.unlock();public String bytes2string(byte bt) int l = bt.length;char out = new charl 1;for (int i = 0, j = 0; i l; i+) for (int i = 0, j = 0; i 4; outj+ = digits0 x0F & bti;if (log.isDebugEnabled() log.debug(hash + new String(out);return new Strin

38、g(out);public byte string2bytes(String str) if (null = str) throw new NullPointerException(Argument is not allowed empty);if (str.length() != 32) throw new IllegalArgumentException(String length must equals 32);byte data = new byte16; char chs = str.toCharArray(); for (int i = 0; i 16; +i) int h = r

39、Digits.get(chsi * 2).intValue();int l = rDigits.get(chsi * 2 + 1).intValue(); datai = (byte) (h & 0 x0F) 4 | l & 0 x0F);return data;PHP 收发消息本文描述如何在 PHP 环境下用 HTTP 协议收发消息。运行环境准备用 HTTP 协议发送或者接收消息，请完成以下准备工作。WindowsIntelliJphpStorm HYPERLINK /phpstorm/download/index.html /phpstorm/download/index.htmlphpS

40、torm其他 IDE 开发环境安装步骤与此类似。Linux/Unix从官网上下载 phpStorm 的 Linux 版本： HYPERLINK /phpstorm/download/index.html /phpstorm/download/index.html解压下载成功的 phpStorm 安装包： tar xfz PhpStorm-2016.1.tar.gzphpStormbincd phpStorm-2016.1;./phpStorm.sh 在输入注册码页面直接单击试用版。单击确定直到出现安装完成界面。运行示例代码在Windows/Linux/Unix环境下，请按照以下步骤运行示例代码

41、。 phpStormPHP（工程名无特殊要求）。将下文具体示例程序中所提供的配置文件（config.ini）以及示例程序（httpProducer.php, httpConsumer.php, Util.php）拷贝到当前的工程中，如图：根据示例代码里的说明修改相关配置信息。右键点击创建的 PHP 文件，选择 Run 执行。注意：请先执行 httpProducer.php 代码，再执行 httpConsumer.php 代码。观察执行结果，如果执行结果有问题，请检查 config.ini 配置是否正确。具体示例程序以下是配置文件（config.ini）、发送消息（httpProducer.ph

42、p）和接收消息（httpConsumer.php）以及所 用工具方法（Util.php）的示例代码。配置文件section#您在控制台创建的Topic Topic = xxx#公测环境的URLsection#您在控制台创建的Topic Topic = xxx#公测环境的URLURL = HYPERLINK / #阿里云身份验证码Ak = xxx#阿里云身份验证密钥Sk = xxx#MQ控制台创建的Producer ID ProducerID = xxx#MQ控制台创建的Consumer ID ConsumerID = xxxURLKey，TagPOST Content-TypeKeyTag，可

43、以放在 perties 里面。发送消息示例程序（httpProducer.php）?php/包含工具类?php/包含工具类include(Util.php);/* 消息发布者者*/class HttpProducer/签名private static $signature = Signature;/在MQ控制台创建的Producer IDprivate static $producerid = ProducerID;/阿里云身份验证码private static $aks = AccessKey;/配置信息private static $configs = null;/构造函数function

44、 construct()/读取配置信息$this:$configs = parse_ini_file(config.ini);/计算md5private function md5($str)return md5($str);/发布消息流程public function process()/打印配置信息var_dump($this:$configs);/获取Topic$topic = $this:$configsTopic;/获取保存Topic的URL路径$url = $this:$configsURL;/读取阿里云访问码$ak = $this:$configsAk;/读取阿里云密钥$sk =

45、$this:$configsSk;/读取Producer ID$pid = $this:$configsProducerID;/HTTP请求体内容$body = utf8_encode(阿里巴巴);$newline = n;/构造工具对象$util = new Util();for ($i = 0; $imd5($body).$newline.$date;/计算签名$sign = $util-calSignatue($signString,$sk);/初始化网络通信模块$ch = curl_init();/构造签名标记$signFlag = $this:$signature.:.$sign;/

46、构造密钥标记$akFlag = $this:$aks.:.$ak;/标记$producerFlag = $this:$producerid.:.$pid;/构造HTTP请求头部内容类型标记$contentFlag = Content-Type:text/html;charset=UTF-8;/构造HTTP请求头部$headers = array($signFlag,$akFlag,$producerFlag,$contentFlag,);/设置HTTP头部内容curl_setopt($ch,CURLOPT_HTTPHEADER,$headers);/ 设 置 HTTP 请 求 类 型 , 此

47、处 为 POST curl_setopt($ch,CURLOPT_CUSTOMREQUEST,POST);/ 设 置 HTTP 请 求 的 URL curl_setopt($ch,CURLOPT_URL,$postUrl);/ 设 置 HTTP 请 求 的 body curl_setopt($ch,CURLOPT_POSTFIELDS,$body);/构造执行环境ob_start();/开始发送HTTP请求curl_exec($ch);/获取请求应答消息$result = ob_get_contents();/清理执行环境ob_end_clean();/打印请求应答结果var_dump($re

48、sult);/关闭连接curl_close($ch);/构造消息发布者$producer = new HttpProducer();/启动消息发布者/启动消息发布者$producer-process();?接收消息示例程序（httpConsumer.php）通过 HTTP 协议接收消息，请参考以下示例代码。calSignatue($signString,$sk);/构造签名标记$signFlag = $this:$signature.:.$sign;/构造密钥标记$akFlag = $this:$ak.:.$ak;/标记$consumerFlag = $this:$consumerid.:.$

49、cid;/构造HTTP请求发送内容类型标记$contentFlag = Content-Type:text/html;charset=UTF-8;/构造HTTP头部信息$headers = array($signFlag,$akFlag,$consumerFlag,$contentFlag,);/构造HTTP请求URL$getUrl = $url./message/?topic=.$topic.&time=.$date.&num=32;/初始化网络通信模块$ch = curl_init();/填充HTTP头部信息curl_setopt($ch,CURLOPT_HTTPHEADER,$heade

50、rs);/ 设 置 HTTP 请 求 类 型 , 此 处 为 GET curl_setopt($ch,CURLOPT_CUSTOMREQUEST,GET);/ 设 置 HTTP 请 求 URL curl_setopt($ch,CURLOPT_URL,$getUrl);/构造执行环境ob_start();/开始发送HTTP请求curl_exec($ch);/获取请求应答消息$result = ob_get_contents();/清理执行环境ob_end_clean();/打印请求应答信息var_dump($result);/关闭HTTP网络连接curl_close($ch);/解析HTTP应答

51、信息$messages = json_decode($result,true);Topic信息,则直接跳过 if (count($messages=0)continue;/依次遍历每个Topic消息foreach (array)$messages as $message)var_dump($message);/获取时间戳$date = (int)($util-microtime_float()*1000);/构造删除Topic消息URL$delUrl = $url./message/?msgHandle=.$messagemsgHandle.&topic=.$topic.&time=.$dat

52、e;/签名字符串$signString = $topic.$newline.$cid.$newline.$messagemsgHandle.$newline.$date;/计算签名$sign = $util-calSignatue($signString,$sk);/构造签名标记$signFlag = $this:$signature.:.$sign;/构造密钥标记$akFlag = $this:$ak.:.$ak;/构造消费者组标记$consumerFlag = $this:$consumerid.:.$cid;/构造HTTP请求头部信息$delheaders = array($signFl

53、ag,$akFlag,$consumerFlag,$contentFlag,);/初始化网络通信模块$ch = curl_init();/填充HTTP请求头部信息curl_setopt($ch,CURLOPT_HTTPHEADER,$delheaders);/设置HTTP请求URL信息curl_setopt($ch,CURLOPT_URL,$delUrl);/ 设 置 HTTP 请 求 类 型 , 此 处 为 DELETE curl_setopt($ch,CURLOPT_CUSTOMREQUEST,DELETE);/构造执行环境ob_start();/开始发送HTTP请求curl_exec($

54、ch);/开始发送HTTP请求curl_exec($ch);/获取请求应答消息$result = ob_get_contents();/清理执行环境ob_end_clean();/打印应答消息var_dump($result);/关闭连接curl_close($ch);catch (Exception $e)/打印异常信息echo $e-getMessage();/构造消息订阅者$consumer = new HttpConsumer();/启动消息订阅者$consumer-process();?php/* 工具类 $blockSize)$key = pack(H*,$hashfunc($ke

55、y);$key = str_pad($key,$blockSize,chr(0 x00);$ipad = str_repeat(chr(0 x36),$blockSize);$opad = str_repeat(chr(0 x5c),$blockSize);$hmac = pack( H*,$hashfunc( ($key$opad).pack(H*,$hashfunc($key$ipad).$str);$sign = base64_encode($hmac);return $sign;/计算时间戳public static function microtime_float()list($us

56、ec,$sec) = explode( ,microtime(); return (float)$usec+(float)$sec);?Python 收发消息本文主要描述如何在 Python 环境下使用 HTTP 协议收发 MQ 消息。运行环境准备用 HTTP 协议发送或者接收消息 ，请完成以下环境准备工作。WindowsPython（ HYPERLINK /downloads/windows/) /downloads/windows/Python2.7。Windows（dospythonPythonLinux/Unix您可以在 Linux 终端输入 python 指令确认是否有预装。如果有执

57、行信息，则说明本台机器已经预装。如果没有，请按以下步骤进行安装：从 Python 官网下载 Python2.7 Linux 版本安装包： HYPERLINK /download/ /download/在 Python2.7 安装包下载保存目录下，运行指令进行解压，例如：tar -xzf python-2.7.11.tgz 进入2.7.11目录，执行./configure 指令。此步骤主要是用于生成 makefile 文件。执行 make 指令进行实际编译。执行 make install 指令进行安装操作。安装完毕后执行 python 命令检查安装是否成功，如下图：运行示例代码按照以下步骤运行示

58、例代码。Windows将下文具体示例程序小节所提供的配置文件以及示例程序拷贝到本地，并保存为 .py 文件，如下图：根据下文示例代码里的说明修改 perties 文件中的相关字段。将终端窗口切换到保存 .py 的文件目录，执行保存的 python 文件，如下图:Linux/Unix步骤参见 Windows。执行结果参照下图：具体示例程序您可以参考以下示例程序测试消息收发功能。1.配置文件property#您在控制台创建的Topic Topic=xxx#公测集群URLproperty#您在控制台创建的Topic Topic=xxx#公测集群URL HYPERLINK / URL=#阿里云官网身份

59、验证访问码Ak=xxx#阿里云身份验证密钥Sk=xxx#MQ控制台创建的Producer ID ProducerID=xxx#MQ控制台创建的Consumer ID ConsumerID=xxx说明：URL中的 Key，Tag 以及 POST Content-Type 没有任何的限制，只要确保 Key 和 Tag 相同唯一即可，可以放在 perties 里面。2.发送消息示例程序通过 HTTP 协议发送消息，请参考以下示例代码。#encoding:utf-8 import ConfigParser import hashlib import httplib import timefrom ur

60、lparse import urlparsefrom Util import parseURL,calSignature消息发布者class HttpProducer(object):def init (self): 签名值self.signature = SignatureProducerID ducerid = ProducerID 消 息 主 题 self.topic = Topic访问码self.ak = AccessKey配置文件解析器self.cf = ConfigParser.ConfigParser()MD5对象self.md5 = hashlib.md5()发布消息主流程de