Java HTTP協(xié)議收發(fā)MQ 消息代碼實例詳解
更新時間：2017年04月20日 14:31:46 作者：落葉
這篇文章主要通過實例代碼為大家詳細介紹了如何在Java 環(huán)境下使用 HTTP 協(xié)議收發(fā) MQ 消息，需要的朋友可以參考下
1. 準備環(huán)境
在工程 POM 文件添加 HTTP Java 客戶端的依賴。
<dependency> <groupId>org.eclipse.jetty</groupId> <artifactId>jetty-client</artifactId> <version>9.3.4.RC1</version> </dependency> <dependency> <groupId>com.aliyun.openservices</groupId> <artifactId>ons-client</artifactId> <version>1.1.11</version> </dependency>
2. 運行代碼配置（user.properties）
您需要設置配置文件(user.properties)的相關內(nèi)容，具體請參考申請 MQ 資源 。
#您在控制臺創(chuàng)建的Topic Topic=xxx #公測url URL=http://publictest-rest.ons.aliyun.com #阿里云身份驗證碼 Ak=xxx #阿里云身份驗證密鑰 Sk=xxx #MQ控制臺創(chuàng)建的Producer ID ProducerID=xxx #MQ控制臺創(chuàng)建的Consumer ID ConsumerID=xxx
說明：URL 中的 Key，Tag以及 POST Content-Type 沒有任何的限制，只要確保Key 和 Tag 相同唯一即可，可以放在 user.properties 里面。
3. HTTP 發(fā)送消息示例代碼
您可以按以下說明設置相應參數(shù)并測試 HTTP 消息發(fā)送功能。
package com.aliyun.openservice.ons.http.demo;
import java.nio.charset.Charset;
import java.util.Date;
import 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.openservices.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";
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);
httpClient.start();
Properties properties=new Properties();
properties.load(HttpProducer.class.getClassLoader().getResourceAsStream("user.properties"));
String topic=properties.getProperty("Topic"); //請在user.properties配置您的Topic
String url=properties.getProperty("URL");//公測集群配置為http://publictest-rest.ons.aliyun.com/
String ak=properties.getProperty("Ak");//請在user.properties配置您的Ak
String sk=properties.getProperty("Sk");//請在user.properties配置您的Sk
String pid=properties.getProperty("ProducerID");//請在user.properties配置您的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 Date().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);
sign=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());
}
}
}
4. HTTP接收消息示例代碼
請按以下說明設置相應參數(shù)并測試 HTTP 消息接收功能。
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.ContentProvider;
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.MqttProducer;
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 static 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);
httpClient.start();
Properties properties=new Properties();
properties.load(HttpConsumer.class.getClassLoader().getResourceAsStream("user.properties"));
String topic=properties.getProperty("Topic"); //請在user.properties配置您的topic
String url=properties.getProperty("URL");//公測集群配置為http://publictest-rest.ons.aliyun.com/
String ak=properties.getProperty("Ak");//請在user.properties配置您的Ak
String sk=properties.getProperty("Sk");//請在user.properties配置您的Sk
String cid=properties.getProperty("ConsumerID");//請在user.properties配置您的Consumer ID
String date=String.valueOf(new Date().getTime());
String sign=null;
String NEWLINE="\n";
String signString;
System.out.println(NEWLINE+NEWLINE);
while (true) {
try {
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.header(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<SimpleMessage> list = null;
if (response.getContentAsString()!=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) {
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+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 (Exception e) {
e.printStackTrace();
}
}
}
}
5. HTTP示例程序工具類
(1)消息封裝類: SimpleMessage.java
package com.aliyun.openservice.ons.http.demo;
public class SimpleMessage {
private String body;
private String msgId;
private String bornTime;
private String msgHandle;
private int reconsumeTimes;
private String tag;
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.msgHandle = msgHandle;
}
public String getMsgHandle() {
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 bornTime;
}
public void setBornTime(String bornTime) {
this.bornTime = bornTime;
}
}
(2)字符串簽名類: MD5.java
package com.aliyun.openservice.ons.http.demo;
import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;
import java.security.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 log = LoggerFactory.getLogger(MD5.class);
private static char[] digits = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' };
private static Map<Character, Integer> rDigits = new HashMap<Character, Integer>(16);
static {
for (int i = 0; i < digits.length; ++i) {
rDigits.put(digits[i], i);
}
}
private static MD5 me = new MD5();
private MessageDigest mHasher;
private final 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 String 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 IllegalArgumentException("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 IllegalArgumentException("md5 need");
}
return bt;
} finally {
this.opLock.unlock();
}
}
public String bytes2string(byte[] bt) {
int l = bt.length;
char[] out = new char[l << 1];
for (int i = 0, j = 0; i < l; i++) {
out[j++] = digits[(0xF0 & bt[i]) >>> 4];
out[j++] = digits[0x0F & bt[i]];
}
if (log.isDebugEnabled()) {
log.debug("[hash]" + new String(out));
}
return new String(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 byte[16];
char[] chs = str.toCharArray();
for (int i = 0; i < 16; ++i) {
int h = rDigits.get(chs[i * 2]).intValue();
int l = rDigits.get(chs[i * 2 + 1]).intValue();
data[i] = (byte) ((h & 0x0F) << 4 | l & 0x0F);
}
return data;
}
}
希望本篇文章對您有所幫助
相關文章
這篇文章主要介紹了Spring AOP入門Demo分享，涉及創(chuàng)建maven項目，編寫切面類，通過bean配置關聯(lián)等相關內(nèi)容，具有一定借鑒價值，需要的朋友可以參考下。2017-12-12
java.lang.ArrayStoreException異常的解決方案
這篇文章主要介紹了java.lang.ArrayStoreException異常的解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12
業(yè)務中經(jīng)常需要按照一個特定的模板導出特定內(nèi)容,有些單元格還要求特殊的格式,所以下面這篇文章主要給大家介紹了關于Java使用EasyExcel模版導出的相關資料,需要的朋友可以參考下2023-10-10
大家好，本篇文章主要講的是Java實現(xiàn)輸出三角形實例代碼，感興趣的同學趕快來看一看吧，對你有幫助的話記得收藏一下，方便下次瀏覽2022-01-01
這篇文章主要介紹了SpringWebMVC的常用注解及應用分層架構,SpringWebMVC是基于ServletAPI構建的原始Web框架,從?開始就包含在Spring框架中,感興趣的朋友可以參考下2024-05-05
springboot前后端分離集成CAS單點登錄(統(tǒng)一認證)
單點登錄是一種身份認證和授權技術,允許用戶在多個應用系統(tǒng)中使用同一套用戶名和密碼進行登錄,本文主要介紹了springboot前后端分離集成CAS單點登錄,具有一定的參考價值,感興趣的可以了解一下2024-09-09
大家都知道基于 SpringBoot 開發(fā)項目可以簡化 Spring 應用的搭建以及開發(fā)過程，提高程序員開發(fā)效率，這是由于其“約定大約配置”的策略及其自動裝配的特點，Starter 就是自動裝配的具體實現(xiàn)，本文詳細介紹了SpringBoot Starter作用及原理，歡迎大家來閱讀學習2023-04-04
這篇文章主要介紹了Java設計模式的代理模式，文中有非常詳細的代碼示例，對正在學習Java設計模式的小伙伴有很大的幫助，感興趣的小伙伴可以參考一下2021-08-08
本文介紹了如何通過Maven配置依賴,創(chuàng)建前端登錄和注冊頁面,并實現(xiàn)后端邏輯,詳細步驟包括配置文件、創(chuàng)建User類、配置中文過濾器及DispatcherServlet,并使用Spring?MVC和JQuery處理前端請求,需要的朋友可以參考下2024-11-11
MyBatis代碼自動生成器Mybatis-Generator的使用詳解
本文詳細介紹如何在SpringBoot項目中使用MyBatis-Generator進行代碼生成,包括配置文件的添加、POM依賴配置、運行配置等步驟,通過自動生成代碼,可以簡化MyBatis的繁瑣配置和SQL編寫,提高開發(fā)效率,注意要考慮MySQL版本兼容性,以及確保路徑配置正確2024-10-10