IotGateway/src/main/java/com/telpo/iotgateway/server/IotSubscribe.java

package com.telpo.iotgateway.server;

import com.telpo.iotgateway.listener.IotJmsConnectionListener;
import com.telpo.iotgateway.listener.IotMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsConnectionListener;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.stereotype.Component;
import org.apache.commons.codec.binary.Base64;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.jms.*;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.util.Hashtable;

/**
 * @program: DipperPositionServer
 * @description: 北斗定位
 * @author: king
 * @create: 2021-01-13 14:01
 */
@Slf4j
@Component
public class IotSubscribe {

    //参数说明，请参见AMQP客户端接入说明文档。
    private String accessKey;
    private String accessSecret;
    private String consumerGroupId;
    //iotInstanceId：企业版实例请填写实例ID，公共实例请填空字符串""。
    private String iotInstanceId;
    private long timeStamp;
    //签名方法：支持hmacmd5、hmacsha1和hmacsha256。
    private String signMethod;
    private String clientId;
    private String iotHost;
    private String iotPort;

    // 执行单元线程池
    private final static ExecutorService executorService = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(50000));
    // Jms连接侦听
    private static JmsConnectionListener myJmsConnectionListener = new IotJmsConnectionListener();
    // 消息侦听
    private static MessageListener messageListener = new IotMessageListener();

    public IotSubscribe(ConfigurableEnvironment environment) {
        //参数说明，请参见AMQP客户端接入说明文档。
        this.accessKey = environment.getProperty("iot.accessKey");
        this.accessSecret = environment.getProperty("iot.accessSecret");
        this.consumerGroupId = environment.getProperty("iot.consumerGroupId");
        //iotInstanceId：企业版实例请填写实例ID，公共实例请填空字符串""。
        this.iotInstanceId = environment.getProperty("iot.iotInstanceId");
        //签名方法：支持hmacmd5、hmacsha1和hmacsha256。
        this.signMethod = environment.getProperty("iot.signMethod");
        this.clientId = environment.getProperty("iot.clientId");
        this.iotHost = environment.getProperty("iot.iotHost");
        this.iotPort = environment.getProperty("iot.iotPort");

        timeStamp = System.currentTimeMillis();
    }

    /*
     * 同步进程线程
     */
    public void start() {

        try {
            //计算签名，password组装方法，请参见AMQP客户端接入说明文档。
            String userInfo = getUserInfo();
            String signContent = "authId=" + accessKey + "&timestamp=" + timeStamp;;
            String password = doSign(signContent,accessSecret, signMethod);

            //接入域名，请参见AMQP客户端接入说明文档。
            String connectionUrl = "failover:(amqps://" + iotHost + ":" + iotPort
                                + "?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");

            // 创建连接。
            log.warn("Befor cf.createConnection");
            Connection connection = cf.createConnection(userInfo, password);
            log.warn("Befor connection.addConnectionListener");
            ((JmsConnection) connection).addConnectionListener(myJmsConnectionListener);
            // 创建会话。
            // Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手动调用message.acknowledge()。
            // Session.AUTO_ACKNOWLEDGE: SDK自动ACK（推荐）。
            log.warn("Befor connection.createSession");
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            log.warn("Befor connection.start");
            connection.start();

            // 创建Receiver连接。消费消息
            log.warn("Befor session.createConsumer");
            MessageConsumer consumer = session.createConsumer(queue);

            // 设置消息侦听
            ((IotMessageListener)messageListener).setExecutorService(this.executorService);
            consumer.setMessageListener(messageListener);

        } catch (IllegalArgumentException e) {
            log.error("IllegalArgumentException:{}", e.getMessage());
        } catch (NamingException e) {
            log.error("NamingException:{}", e.getMessage());
        } catch (JMSException e) {
            log.error("JMSException:{}", e.getMessage());
        } catch (Exception e) {
            log.error("Exception:{}", e.getMessage());
        }
    }

    /*
     * userInfo组装
     */
    private String getUserInfo() {
        //userInfo组装
        String userInfo = clientId + "|authMode=aksign"
                + ",signMethod=" + signMethod
                + ",timestamp=" + timeStamp
                + ",authId=" + accessKey
                + ",iotInstanceId=" + iotInstanceId
                + ",consumerGroupId=" + consumerGroupId
                + "|";
        //计算签名，password组装方法，请参见AMQP客户端接入说明文档。
        return userInfo;
    }

    /**
     * 计算签名，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);
    }

}