开发者社区 云原生 云消息队列 文章 正文

使用阿里云消息队列

2016年05月11日 9617
版权
版权声明:
本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写 侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。
简介:

使用阿里云消息队列

控制台地址:http://ons.console.aliyun.com/#/home/topic

(1)生成Producer ID

点击"申请发布"

示例代码:

Java代码 收藏代码
  1. package com.alibaba.ons.demo;
  3. import java.util.Properties;
  5. import com.aliyun.openservices.ons.api.Message;
  6. import com.aliyun.openservices.ons.api.ONSFactory;
  7. import com.aliyun.openservices.ons.api.Producer;
  8. import com.aliyun.openservices.ons.api.PropertyKeyConst;
  9. import com.aliyun.openservices.ons.api.SendResult;
  11. public class ProducerClient {
  13. public static void main(String[] args) {
  14. Properties properties = new Properties();
  15. properties.put(PropertyKeyConst.ProducerId, "PID_whuang");
  16. properties.put(PropertyKeyConst.AccessKey, "请输入AccessKey");
  17. properties.put(PropertyKeyConst.SecretKey, "请输入SecretKey");
  18. Producer producer = ONSFactory.createProducer(properties);
  20. //在发送消息前,必须调用start方法来启动Producer,只需调用一次即可。
  21. producer.start();
  22. Message msg = new Message(
  23. //Message Topic
  24. "com_hbjltv",
  25. //Message Tag,
  26. //可理解为Gmail中的标签,对消息进行再归类,方便Consumer指定过滤条件在ONS服务器过滤
  27. "TagA",
  28. //Message Body
  29. //任何二进制形式的数据,ONS不做任何干预,需要Producer与Consumer协商好一致的序列化和反序列化方式
  30. "Hello ONS".getBytes()
  31. );
  33. // 设置代表消息的业务关键属性,请尽可能全局唯一。
  34. // 以方便您在无法正常收到消息情况下,可通过ONS Console查询消息并补发。
  35. // 注意:不设置也不会影响消息正常收发
  36. msg.setKey("ORDERID_100");
  38. //发送消息,只要不抛异常就是成功
  39. SendResult sendResult = producer.send(msg);
  40. System.out.println(sendResult);
  42. // 在应用退出前,销毁Producer对象
  43. // 注意:如果不销毁也没有问题
  44. producer.shutdown();
  45. }
  46. }

(2)生成Consumer ID

点击"申请订阅"

示例代码:

Java代码 收藏代码
  1. public class ConsumerTest {
  2. public static void main(String[] args) {
  3. Properties properties = new Properties();
  4. properties.put(PropertyKeyConst.ConsumerId, "CID_tv_mobile");
  5. properties.put(PropertyKeyConst.AccessKey, "请输入AccessKey");
  6. properties.put(PropertyKeyConst.SecretKey, "请输入SecretKey");
  7. Consumer consumer = ONSFactory.createConsumer(properties);
  8. consumer.subscribe("com_hbjltv", "*", new MessageListener() {
  9. public Action consume(Message message, ConsumeContext context) {
  10. System.out.println("Receive: " + message);
  11. return Action.CommitMessage;
  12. }
  13. });
  14. consumer.start();
  15. System.out.println("Consumer Started");
  16. }
  17. }

(3) clientId 的限制

阿里云消息队列对clientId的名称有严格限制:

(a)必须以申请的Consumer ID 开头,后面跟@@@,接着跟用于区分客户端的标志,

例如:

CID_tv_mobile@@@86458fd 是合法的

CID_tv_mobile@@86458fd 是非法的,因为只有两个@

(b)总长度不能超过23个字符

例如

CID_tv_mobile@@@86458_A是合法的

CID_tv_mobile@@@86458_Ab是非法的,因为超过了23个字符



(4)在手机端(客户端)增加订阅逻辑

Java代码 收藏代码
  1. package com.service;
  3. import java.security.InvalidKeyException;
  4. import java.security.NoSuchAlgorithmException;
  6. import org.eclipse.paho.client.mqttv3.MqttClient;
  7. import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
  8. import org.eclipse.paho.client.mqttv3.MqttException;
  9. import org.eclipse.paho.client.mqttv3.internal.MemoryPersistence;
  11. import android.app.Service;
  12. import android.content.Context;
  13. import android.content.Intent;
  14. import android.content.SharedPreferences;
  15. import android.os.IBinder;
  17. import com.common.util.SystemHWUtil;
  18. import com.dict.Constants3;
  19. import com.jianli.R;
  20. import com.push.PushCallback;
  21. import com.string.widget.util.ValueWidget;
  22. import com.util.MacSignature;
  23. import com.util.ShopUtil;
  25. /**
  26. * @author Dominik Obermaier
  27. */
  28. public class MQTTService extends Service {
  30. // public static final String BROKER_URL =
  31. // "tcp://broker.mqttdashboard.com:1883";
  32. // public static String BROKER_URL = "tcp://172.16.15.50:1883";
  33. public static String BROKER_URL_FORMAT = "tcp://%s:%s";
  34. // public static final String BROKER_URL = "tcp://test.mosquitto.org:1883";
  36. /*
  37. * In a real application, you should get an Unique Client ID of the device
  38. * and use this, see
  39. * http://android-developers.blogspot.de/2011/03/identifying
  40. * -app-installations.html
  41. */
  42. public static String clientId = null;
  44. /**
  45. * 不能含有英文句点,可以包含下划线
  46. */
  47. public static final String TOPIC = "com_hbjltv";
  48. private MqttClient mqttClient;
  49. // private String ip="182.92.80.122";
  50. /***
  51. * 是否连接上activeMQ
  52. */
  53. private boolean online = false;
  54. boolean isAliyun=false;
  56. public IBinder onBind(Intent intent) {
  57. return null;
  58. }
  60. @Override
  61. public void onCreate() {
  62. super.onCreate();
  63. }
  65. private MqttClient createMqttClient(String serverURL, String clientId) throws MqttException{
  66. return new MqttClient(serverURL, clientId,
  67. new MemoryPersistence());
  68. }
  69. /***
  70. *
  71. * @param serverURL
  72. * @param clientId
  73. * : 最大长度:23
  74. * @param isAllowOffline
  75. * @param username
  76. * @param password
  77. * @throws MqttException
  78. */
  79. private void connectAndSubscribe(String serverURL, String clientId,
  80. /* String topicFilter, */boolean isAllowOffline, String username,
  81. String password) throws MqttException {
  83. if(isAliyun){
  84. if(!ShopUtil.validateClientId(getApplicationContext(), clientId)){
  85. return;
  86. }
  87. }
  88. mqttClient = createMqttClient(serverURL, clientId);
  89. MqttConnectOptions options = new MqttConnectOptions();
  90. options.setCleanSession(!isAllowOffline);// mqtt receive offline message
  91. if (ValueWidget.isNullOrEmpty(username)) {
  92. username = null;
  93. }
  94. String sign=null;
  95. if(isAliyun){
  96. try {
  97. sign = MacSignature.macSignature(Constants3.CONSUMER_ID_TV, password);
  98. password=sign;
  99. } catch (InvalidKeyException e) {
  100. e.printStackTrace();
  101. } catch (NoSuchAlgorithmException e) {
  102. e.printStackTrace();
  103. }
  105. }
  106. if (ValueWidget.isNullOrEmpty(password)) {
  107. password = null;
  108. } else {
  109. options.setPassword(password.toCharArray());
  110. }
  111. options.setUserName(username);
  112. options.setConnectionTimeout(10);
  113. options.setKeepAliveInterval(10);
  114. if(null==mqttClient){//点击HOME键,过很长时间,再点击应用时,mqttClient为null
  115. mqttClient = createMqttClient(serverURL, clientId);
  116. }
  117. mqttClient.setCallback(new PushCallback(this));
  118. boolean isSuccess=false;
  119. mqttClient.connect(options);
  120. isSuccess=true;
  121. // Subscribe to all subtopics of homeautomation
  122. // mqttClient.subscribe(topicFilter);
  123. if(null==mqttClient){//点击HOME键,过很长时间,再点击应用时,mqttClient为null
  124. mqttClient = createMqttClient(serverURL, clientId);
  125. }
  126. if(isAliyun){
  127. final String p2ptopic = TOPIC+"/p2p/";
  128. //同时订阅两个topic,一个是基于标准mqtt协议的发布订阅模式,一个是扩展的点对点推送模式
  129. final String[] topicFilters=new String[]{TOPIC,p2ptopic};
  130. mqttClient.subscribe(topicFilters);
  131. }else{
  132. mqttClient.subscribe(new String[] { TOPIC, clientId });
  133. }
  134. }
  136. @Override
  137. public void onStart(Intent intent, int startId) {
  138. final boolean isRestart=intent.getBooleanExtra("isRestart", false);
  139. ShopUtil.logger2("restart MQTT service:"+isRestart);
  140. // super.onStart(intent, startId);
  141. // if (intent == null) {//重启服务时intent 确实为空
  143. // Log.d(Constants.LOG_TAG, "intent is null");
  144. // return;
  145. // }
  146. Context context = getApplicationContext();
  147. clientId = ShopUtil.getIMEI(context);
  148. // Bundle bundle=intent.getExtras();
  149. // String ip=bundle.getString(Constants.ACTIVEMQ_IP);
  150. // final String ip = context.getString(R.string.pushserver_ip);
  151. SharedPreferences preferences = getApplicationContext()
  152. .getSharedPreferences(Constants3.SHAREDPREFERENCES_NAME,
  153. Context.MODE_PRIVATE);
  154. final String ip ="mqtt.ons.aliyun.com";// preferences.getString("pushserver_ip", context.getString(R.string.pushserver_ip));
  155. final String port = preferences.getString("pushserver_port", "1883");
  156. isAliyun=SystemHWUtil.parse2Boolean(preferences.getString("is_aliyun_mq_ONS", "false"));
  157. // String topic=bundle.getString(Constants.ACTIVEMQ_TOPIC);
  158. System.out.println("push ip:"+ip);
  159. new Thread(new Runnable() {
  160. /****
  161. * 尝试连接的次数,为什么要尝试连接几次那?
  162. * (1)无wifi时启动,则肯定连接失败,所以尝试连接三次,只要在这个期间启动wifi就可以连接上activeMQ;<br />
  163. * (2)之前连接上,然后断开wifi,然后又启动wifi,<br />
  164. * 这时容易报 "Broker unavailable"异常,暂时不清楚原因,所以也需要继续尝试连接;<br />
  165. *
  166. */
  167. private int tryTime = 5;
  169. @Override
  170. public void run() {
  171. System.out.println(tryTime+","+mqttClient+","+isOnline() );
  172. while (tryTime > 0
  173. && (!isOnline() || mqttClient == null || (!mqttClient
  174. .isConnected())||isRestart)) {
  175. try {
  176. ShopUtil.logger2("start push service");
  177. ShopUtil.logger2("push server:"+ip);
  178. String prefix=Constants3.CONSUMER_ID_TV+"@@@";
  179. int remainingLength=23-prefix.length();
  180. String suffix=null;
  181. if(clientId.length()>remainingLength){
  182. suffix=clientId.substring(0,remainingLength);
  183. }else{
  184. suffix=clientId;
  185. }
  186. String clientId2=prefix+suffix;
  187. connectAndSubscribe(String.format(
  188. MQTTService.BROKER_URL_FORMAT, ip, port),
  189. clientId2, /* topic, */true, ""/*自己申请的access key*/, ""/*secret*/);
  190. ShopUtil.logger2("clientId:" + clientId2);
  191. ShopUtil.logger2("succeed to connect to activeMQ");
  192. setOnline(true);
  193. } catch (MqttException e) {
  194. setOnline(false);
  195. mqttClient=null;
  196. e.printStackTrace();
  197. ShopUtil.logger2("抛异常:"+e.getMessage());
  198. ShopUtil.logger2("ip:" + ip + " ,port:" + port);
  199. try {
  200. Thread.sleep(10000);
  201. } catch (InterruptedException e1) {
  202. e1.printStackTrace();
  203. }
  204. }
  205. tryTime--;
  206. }
  208. }
  209. }).start();
  210. // new Thread(new Runnable() {
  211. // @Override
  212. // public void run() {
  213. // System.out.println("start:"+System.currentTimeMillis());
  214. // try {
  215. // Thread.sleep(10000);
  216. // } catch (InterruptedException e) {
  217. // e.printStackTrace();
  218. // }
  219. // while(true){
  220. // try {
  221. // Thread.sleep(10000);
  222. // if(mqttClient!=null&& !mqttClient.isConnected()){
  223. // System.out.println("disConnected:"+System.currentTimeMillis());
  224. // }
  225. // } catch (InterruptedException e) {
  226. // e.printStackTrace();
  227. // }
  228. // }
  229. // }
  230. // }).start();
  231. }
  233. @Override
  234. public void onDestroy() {
  235. setOnline(false);
  237. try {
  238. ShopUtil.logger2("MQTTService destory");
  239. mqttClient.disconnect(0);
  240. } catch (MqttException e) {
  241. // Toast.makeText(getApplicationContext(),
  242. // "Something went wrong!" + e.getMessage(), Toast.LENGTH_LONG)
  243. // .show();
  244. e.printStackTrace();
  246. }
  247. mqttClient = null;
  248. stopForeground(true);
  249. Intent intent = new Intent("com.dbjtech.waiqin.destroy");
  250. sendBroadcast(intent);
  251. }
  255. public boolean isOnline() {
  256. return online;
  257. }
  259. public void setOnline(boolean online) {
  260. this.online = online;
  261. }
  262. @Override
  263. public int onStartCommand(Intent intent, int flags, int startId) {
  264. flags = START_STICKY;
  265. return super.onStartCommand(intent, flags, startId);
  266. }
  267. }
目录
热门文章
最新文章

AltStyle によって変換されたページ (->オリジナル) /