使用阿里云消息队列-阿里云开发者社区

使用阿里云消息队列

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

(1)生成Producer ID

点击"申请发布"

示例代码:

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

(2)生成Consumer ID

点击"申请订阅"

示例代码:

Java代码  
public class ConsumerTest {  
    public static void main(String[] args) {  
        Properties properties = new Properties();  
        properties.put(PropertyKeyConst.ConsumerId, "CID_tv_mobile");  
        properties.put(PropertyKeyConst.AccessKey, "请输入AccessKey");  
        properties.put(PropertyKeyConst.SecretKey, "请输入SecretKey");  
        Consumer consumer = ONSFactory.createConsumer(properties);  
        consumer.subscribe("com_hbjltv", "*", new MessageListener() {  
            public Action consume(Message message, ConsumeContext context) {  
                System.out.println("Receive: " + message);  
                return Action.CommitMessage;  
            }  
        });  
        consumer.start();  
        System.out.println("Consumer Started");  
    }  
}     

(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代码  
package com.service;  
  
import java.security.InvalidKeyException;  
import java.security.NoSuchAlgorithmException;  
  
import org.eclipse.paho.client.mqttv3.MqttClient;  
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;  
import org.eclipse.paho.client.mqttv3.MqttException;  
import org.eclipse.paho.client.mqttv3.internal.MemoryPersistence;  
  
import android.app.Service;  
import android.content.Context;  
import android.content.Intent;  
import android.content.SharedPreferences;  
import android.os.IBinder;  
  
import com.common.util.SystemHWUtil;  
import com.dict.Constants3;  
import com.jianli.R;  
import com.push.PushCallback;  
import com.string.widget.util.ValueWidget;  
import com.util.MacSignature;  
import com.util.ShopUtil;  
  
/** 
 * @author Dominik Obermaier 
 */  
public class MQTTService extends Service {  
  
    // public static final String BROKER_URL =  
    // "tcp://broker.mqttdashboard.com:1883";  
    // public static String BROKER_URL = "tcp://172.16.15.50:1883";  
    public static String BROKER_URL_FORMAT = "tcp://%s:%s";  
    // public static final String BROKER_URL = "tcp://test.mosquitto.org:1883";  
  
    /* 
     * In a real application, you should get an Unique Client ID of the device 
     * and use this, see 
     * http://android-developers.blogspot.de/2011/03/identifying 
     * -app-installations.html 
     */  
    public static String clientId = null;  
  
    /** 
     * 不能含有英文句点,可以包含下划线 
     */  
    public static final String TOPIC = "com_hbjltv";  
    private MqttClient mqttClient;  
//  private String ip="182.92.80.122";  
    /*** 
     * 是否连接上activeMQ 
     */  
    private boolean online = false;  
    boolean isAliyun=false;  
  
    public IBinder onBind(Intent intent) {  
        return null;  
    }  
  
    @Override  
    public void onCreate() {  
        super.onCreate();  
    }  
  
    private MqttClient createMqttClient(String serverURL, String clientId) throws MqttException{  
        return new MqttClient(serverURL, clientId,  
                new MemoryPersistence());  
    }  
    /*** 
     *  
     * @param serverURL 
     * @param clientId 
     *            : 最大长度:23 
     * @param isAllowOffline 
     * @param username 
     * @param password 
     * @throws MqttException 
     */  
    private void connectAndSubscribe(String serverURL, String clientId,  
    /* String topicFilter, */boolean isAllowOffline, String username,  
            String password) throws MqttException {  
          
        if(isAliyun){  
            if(!ShopUtil.validateClientId(getApplicationContext(), clientId)){  
                return;  
            }  
        }  
        mqttClient = createMqttClient(serverURL, clientId);  
        MqttConnectOptions options = new MqttConnectOptions();  
        options.setCleanSession(!isAllowOffline);// mqtt receive offline message  
        if (ValueWidget.isNullOrEmpty(username)) {  
            username = null;  
        }  
        String sign=null;  
        if(isAliyun){  
            try {  
                sign = MacSignature.macSignature(Constants3.CONSUMER_ID_TV, password);  
                password=sign;  
            } catch (InvalidKeyException e) {  
                e.printStackTrace();  
            } catch (NoSuchAlgorithmException e) {  
                e.printStackTrace();  
            }  
              
        }  
        if (ValueWidget.isNullOrEmpty(password)) {  
            password = null;  
        } else {  
            options.setPassword(password.toCharArray());  
        }  
        options.setUserName(username);  
        options.setConnectionTimeout(10);  
        options.setKeepAliveInterval(10);  
        if(null==mqttClient){//点击HOME键,过很长时间,再点击应用时,mqttClient为null  
            mqttClient = createMqttClient(serverURL, clientId);  
        }  
        mqttClient.setCallback(new PushCallback(this));  
        boolean isSuccess=false;  
        mqttClient.connect(options);  
        isSuccess=true;  
        // Subscribe to all subtopics of homeautomation  
        // mqttClient.subscribe(topicFilter);  
        if(null==mqttClient){//点击HOME键,过很长时间,再点击应用时,mqttClient为null  
            mqttClient = createMqttClient(serverURL, clientId);  
        }  
        if(isAliyun){  
            final String p2ptopic = TOPIC+"/p2p/";  
            //同时订阅两个topic，一个是基于标准mqtt协议的发布订阅模式，一个是扩展的点对点推送模式  
            final String[] topicFilters=new String[]{TOPIC,p2ptopic};  
            mqttClient.subscribe(topicFilters);  
        }else{  
            mqttClient.subscribe(new String[] { TOPIC, clientId });  
        }  
    }  
  
    @Override  
    public void onStart(Intent intent, int startId) {  
        final boolean isRestart=intent.getBooleanExtra("isRestart", false);  
        ShopUtil.logger2("restart MQTT service:"+isRestart);  
        // super.onStart(intent, startId);  
//      if (intent == null) {//重启服务时intent 确实为空  
          
//          Log.d(Constants.LOG_TAG, "intent is null");  
//          return;  
//      }  
        Context context = getApplicationContext();  
        clientId = ShopUtil.getIMEI(context);  
        // Bundle bundle=intent.getExtras();  
        // String ip=bundle.getString(Constants.ACTIVEMQ_IP);  
//      final String ip = context.getString(R.string.pushserver_ip);  
        SharedPreferences preferences = getApplicationContext()  
                .getSharedPreferences(Constants3.SHAREDPREFERENCES_NAME,  
                        Context.MODE_PRIVATE);  
        final String ip ="mqtt.ons.aliyun.com";// preferences.getString("pushserver_ip", context.getString(R.string.pushserver_ip));  
        final String port = preferences.getString("pushserver_port", "1883");  
        isAliyun=SystemHWUtil.parse2Boolean(preferences.getString("is_aliyun_mq_ONS", "false"));  
        // String topic=bundle.getString(Constants.ACTIVEMQ_TOPIC);  
         System.out.println("push ip:"+ip);  
        new Thread(new Runnable() {  
            /**** 
             * 尝试连接的次数,为什么要尝试连接几次那? 
             * (1)无wifi时启动,则肯定连接失败,所以尝试连接三次,只要在这个期间启动wifi就可以连接上activeMQ;<br /> 
             * (2)之前连接上,然后断开wifi,然后又启动wifi,<br /> 
             * 这时容易报 "Broker unavailable"异常,暂时不清楚原因,所以也需要继续尝试连接;<br /> 
             *  
             */  
            private int tryTime = 5;  
  
            @Override  
            public void run() {  
                System.out.println(tryTime+","+mqttClient+","+isOnline() );  
                while (tryTime > 0  
                        && (!isOnline() || mqttClient == null || (!mqttClient  
                                .isConnected())||isRestart)) {  
                    try {  
                        ShopUtil.logger2("start push service");  
                        ShopUtil.logger2("push server:"+ip);  
                        String prefix=Constants3.CONSUMER_ID_TV+"@@@";  
                        int remainingLength=23-prefix.length();  
                        String suffix=null;  
                        if(clientId.length()>remainingLength){  
                            suffix=clientId.substring(0,remainingLength);  
                        }else{  
                            suffix=clientId;  
                        }  
                        String clientId2=prefix+suffix;  
                        connectAndSubscribe(String.format(  
                                MQTTService.BROKER_URL_FORMAT, ip, port),  
                                clientId2, /* topic, */true, ""/*自己申请的access key*/, ""/*secret*/);  
                         ShopUtil.logger2("clientId:" + clientId2);  
                        ShopUtil.logger2("succeed to connect to activeMQ");  
                        setOnline(true);  
                    } catch (MqttException e) {  
                        setOnline(false);  
                        mqttClient=null;  
                         e.printStackTrace();  
                        ShopUtil.logger2("抛异常:"+e.getMessage());  
                        ShopUtil.logger2("ip:" + ip + " ,port:" + port);  
                        try {  
                            Thread.sleep(10000);  
                        } catch (InterruptedException e1) {  
                            e1.printStackTrace();  
                        }  
                    }  
                    tryTime--;  
                }  
  
            }  
        }).start();  
//      new Thread(new Runnable() {  
//          @Override  
//          public void run() {  
//              System.out.println("start:"+System.currentTimeMillis());  
//              try {  
//                  Thread.sleep(10000);  
//              } catch (InterruptedException e) {  
//                  e.printStackTrace();  
//              }  
//              while(true){  
//                  try {  
//                      Thread.sleep(10000);  
//                      if(mqttClient!=null&& !mqttClient.isConnected()){  
//                          System.out.println("disConnected:"+System.currentTimeMillis());  
//                      }  
//                  } catch (InterruptedException e) {  
//                      e.printStackTrace();  
//                  }  
//              }  
//          }  
//      }).start();  
    }  
  
    @Override  
    public void onDestroy() {  
        setOnline(false);  
           
        try {  
            ShopUtil.logger2("MQTTService destory");  
            mqttClient.disconnect(0);  
        } catch (MqttException e) {  
//          Toast.makeText(getApplicationContext(),  
//                  "Something went wrong!" + e.getMessage(), Toast.LENGTH_LONG)  
//                  .show();  
            e.printStackTrace();  
              
        }  
        mqttClient = null;  
        stopForeground(true);  
        Intent intent = new Intent("com.dbjtech.waiqin.destroy");    
        sendBroadcast(intent);   
    }  
  
      
  
    public boolean isOnline() {  
        return online;  
    }  
  
    public void setOnline(boolean online) {  
        this.online = online;  
    }  
    @Override  
    public int onStartCommand(Intent intent, int flags, int startId) {  
        flags = START_STICKY;  
        return super.onStartCommand(intent, flags, startId);  
    }  
}  