接入阿里額溫槍服務器端開發

5.服務端開發
服務端我們以Java為例，演示如何接收IoT平臺推送過來的設備上報數據。
5.1 業務服務器接收IoT數據
參考服務端訂閱AMQP文檔 https://help.aliyun.com/document_detail/143601.html
完整代碼如下：
package com.aliyun.iot;


import org.apache.commons.Codec.binary.Base64;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsConnectionListener;
import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
import org.slf4j.Logger;
import org.slf4j.LoggeRFactory;


import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.net.URI;
import java.util.Hashtable;


public class AMQPClient {


    private final static Logger logger = LoggerFactory.getLogger(AMQPClient.class);
    //消費組配置參數
    private static String accessKey = "阿里云賬號ak";
    private static String accessSecret = "阿里云賬號as";
    private static String consumerGroupId = "服務端訂閱消費組ID";
    private static String aliUID = "替換你的阿里云賬號UID";


    public static void main(String[] args) throws Exception {
        
        long timeStamp = System.currentTimeMillis();
        //簽名方法
        String signMethod = "hmacsha1";
        //控制臺服務端訂閱中消費組狀態頁客戶端ID一欄將顯示clientId參數。
        //建議使用機器UUID、MAC地址、IP等唯一標識等作為clientId。便于您區分識別不同的客戶端。
        String clientId = "ecs_"+System.currentTimeMillis();


        //UserName組裝
        String userName = clientId + "|authMode=aksign"
                + ",signMethod=" + signMethod
                + ",timestamp=" + timeStamp
                + ",authId=" + accessKey
                + ",consumerGroupId=" + consumerGroupId
                + "|";
        //password組裝
        String signContent = "authId=" + accessKey + "×tamp=" + timeStamp;
        String password = doSign(signContent,accessSecret, signMethod);
        //按照qpid-jms的規范，組裝連接URL。
        String connectionUrl = "failover:(amqps://"+aliUID+".iot-amqp.cn-shanghai.aliyuncs.com:5671?amqp.idleTimeout=80000)"
                + "?failover.reconnectDelay=30";


        Hashtable<String, String> hashtable = new Hashtable<>();
        hashtable.put("connectionfactory.SBCF",connectionUrl);
        hashtable.put("queue.QUEUE", "default");
        hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
        Context context = new InitialContext(hashtable);
        ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF");
        Destination queue = (Destination)context.lookup("QUEUE");
        // 創建和IoT平臺的AMQP連接
        Connection connection = cf.createConnection(userName, password);
        ((JmsConnection) connection).addConnectionListener(myJmsConnectionListener);
        // 創建 Session
        // Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手動調用message.acknowledge()
        // Session.AUTO_ACKNOWLEDGE: SDK自動ACK（推薦）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        connection.start();
        // 創建消費者
        MessageConsumer consumer = session.createConsumer(queue);
        consumer.setMessageListener(messageListener);
    }


    private static MessageListener messageListener = new MessageListener() {
        @Override
        public void onMessage(Message message) {
            try {
                byte[] body = message.getBody(byte[].class);
                String content = new String(body);
                String topic = message.getStringProperty("topic");
                String messageId = message.getStringProperty("messageId");
                logger.info("receive message"
                        + ", topic = " + topic
                        + ", messageId = " + messageId
                        + ", content = " + content);
                System.out.println();
                //如果創建Session選擇的是Session.CLIENT_ACKNOWLEDGE，這里需要手動ACK。
                //message.acknowledge();
                //如果要對收到的消息做耗時的處理，請異步處理，確保這里不要有耗時邏輯。
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    };


    private static JmsConnectionListener myJmsConnectionListener = new JmsConnectionListener() {
        /**
         * 連接成功建立。
         */
        @Override
        public void onConnectionEstablished(URI remoteURI) {
            logger.info("onConnectionEstablished, remoteUri:{}", remoteURI);
        }


        /**
         * 嘗試過最大重試次數之后，最終連接失敗。
         */
        @Override
        public void onConnectionFailure(Throwable error) {
            logger.error("onConnectionFailure, {}", error.getMessage());
        }


        /**
         * 連接中斷。
         */
        @Override
        public void onConnectionInterrupted(URI remoteURI) {
            logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI);
        }


        /**
         * 連接中斷后又自動重連上。
         */
        @Override
        public void onConnectionRestored(URI remoteURI) {
            logger.info("onConnectionRestored, remoteUri:{}", remoteURI);
        }


        @Override
        public void onInboundMessage(JmsInboundMessageDispatch envelope) {}


        @Override
        public void onSessionClosed(Session session, Throwable cause) {}


        @Override
        public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {}


        @Override
        public void onProducerClosed(MessageProducer producer, Throwable cause) {}
    };


    /**
     * password簽名計算方法，請參見上一篇文檔：AMQP客戶端接入說明。
     */
    private static String doSign(String toSignString, String secret, String signMethod) throws Exception {
        SecretKeySpec signingKey = new SecretKeySpec(secret.getBytes(), signMethod);
        Mac mac = Mac.getInstance(signMethod);
        mac.init(signingKey);
        byte[] rawHmac = mac.doFinal(toSignString.getBytes());
        return Base64.encodeBase64String(rawHmac);
    }
}
復制代碼
6.設備運行日志6.1 運行數據上報


6.2 數據流轉日志


6.3 服務端訂閱消費組情況