生活物联网平台 - 数据AMQP方式推送

<!--amqp1.0qpidclient--><dependency><groupId>org.apache.qpid</groupId><artifactId>qpid-jms-client</artifactId><version>0.47.0</version></dependency><!--utilforbase64--><dependency><groupId>commons-codec</groupId><artifactId>commons-codec</artifactId><version>1.10</version></dependency>

//EndPoint是连接节点，具体取值如下表所示中国内地：amqps://ilop.iot-amqp.cn-shanghai.aliyuncs.com:5671新加坡：amqps://ilop.iot-amqp.ap-southeast-1.aliyuncs.com:5671美国（弗吉尼亚）：amqps://ilop.iot-amqp.us-east-1.aliyuncs.com:5671德国（法兰克福）：amqps://ilop.iot-amqp.eu-central-1.aliyuncs.com:5671

importjava.net.URI;
importjava.util.Hashtable;
importjava.util.concurrent.ExecutorService;
importjava.util.concurrent.LinkedBlockingQueue;
importjava.util.concurrent.ThreadPoolExecutor;
importjava.util.concurrent.TimeUnit;
importjavax.crypto.Mac;
importjavax.crypto.spec.SecretKeySpec;
importjavax.jms.Connection;
importjavax.jms.ConnectionFactory;
importjavax.jms.Destination;
importjavax.jms.Message;
importjavax.jms.MessageConsumer;
importjavax.jms.MessageListener;
importjavax.jms.MessageProducer;
importjavax.jms.Session;
importjavax.naming.Context;
importjavax.naming.InitialContext;
importorg.apache.commons.codec.binary.Base64;
importorg.apache.qpid.jms.JmsConnection;
importorg.apache.qpid.jms.JmsConnectionListener;
importorg.apache.qpid.jms.message.JmsInboundMessageDispatch;
importorg.slf4j.Logger;
importorg.slf4j.LoggerFactory;
publicclassAmqpJavaClientDemo {
privatefinalstaticLoggerlogger=LoggerFactory.getLogger(AmqpJavaClientDemo.class);
//业务处理异步线程池，线程池参数可以根据您的业务特点调整，或者您也可以用其他异步方式处理接收到的消息。privatefinalstaticExecutorServiceexecutorService=newThreadPoolExecutor(
Runtime.getRuntime().availableProcessors(),
Runtime.getRuntime().availableProcessors() *2, 60, TimeUnit.SECONDS,
newLinkedBlockingQueue<>(50000));
publicstaticvoidmain(String[] args) throwsException {
StringappKey="${YourAppkey}";
StringappSecret="${YourAppSecret}";
StringconsumerGroupId="${YourAppkey}";
longrandom=xxxxx;
//建议使用机器UUID、MAC地址、IP等唯一标识等作为clientId。便于您区分识别不同的客户端。StringclientId="${YourClientId}";
StringuserName=clientId+"|authMode=appkey"+",signMethod="+"SHA256"+",random="+random+",appKey="+appKey+",groupId="+consumerGroupId+"|";
StringsignContent="random="+random;
Stringpassword=doSign(signContent, appSecret, "HmacSHA256");
StringconnectionUrlTemplate="failover:(${AMQPEndPointUrl}?amqp.idleTimeout=80000)"+"?failover.maxReconnectAttempts=10&failover.reconnectDelay=30";
Hashtable<String, String>hashtable=newHashtable<>();
hashtable.put("connectionfactory.SBCF",connectionUrlTemplate);
hashtable.put("queue.QUEUE", "default");
hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
Contextcontext=newInitialContext(hashtable);
ConnectionFactorycf= (ConnectionFactory)context.lookup("SBCF");
Destinationqueue= (Destination)context.lookup("QUEUE");
// 创建连接。Connectionconnection=cf.createConnection(userName, password);
        ((JmsConnection) connection).addConnectionListener(myJmsConnectionListener);
// 创建会话。// Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手动调用message.acknowledge()。// Session.AUTO_ACKNOWLEDGE: SDK自动ACK（推荐）。Sessionsession=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
connection.start();
// 创建Receiver连接。MessageConsumerconsumer=session.createConsumer(queue);
consumer.setMessageListener(messageListener);
    }
privatestaticMessageListenermessageListener=newMessageListener() {
@OverridepublicvoidonMessage(Messagemessage) {
try {
//1.收到消息之后一定要ACK。// 推荐做法：创建Session选择Session.AUTO_ACKNOWLEDGE，这里会自动ACK。// 其他做法：创建Session选择Session.CLIENT_ACKNOWLEDGE，这里一定要调message.acknowledge()来ACK。// message.acknowledge();//2.建议异步处理收到的消息，确保onMessage函数里没有耗时逻辑。// 如果业务处理耗时过程过长阻塞住线程，可能会影响SDK收到消息后的正常回调。executorService.submit(() ->processMessage(message));
            } catch (Exceptione) {
logger.error("submit task occurs exception ", e);
            }
        }
    };
/*** 在这里处理您收到消息后的具体业务逻辑。*/privatestaticvoidprocessMessage(Messagemessage) {
try {
byte[] body=message.getBody(byte[].class);
Stringcontent=newString(body);
Stringtopic=message.getStringProperty("topic");
StringmessageId=message.getStringProperty("messageId");
logger.info("receive message"+", topic = "+topic+", messageId = "+messageId+", content = "+content);
        } catch (Exceptione) {
logger.error("processMessage occurs error ", e);
        }
    }
privatestaticJmsConnectionListenermyJmsConnectionListener=newJmsConnectionListener() {
/*** 连接成功建立。*/@OverridepublicvoidonConnectionEstablished(URIremoteURI) {
logger.info("onConnectionEstablished, remoteUri:{}", remoteURI);
        }
/*** 尝试过最大重试次数之后，最终连接失败。*/@OverridepublicvoidonConnectionFailure(Throwableerror) {
logger.error("onConnectionFailure, {}", error.getMessage());
        }
/*** 连接中断。*/@OverridepublicvoidonConnectionInterrupted(URIremoteURI) {
logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI);
        }
/*** 连接中断后又自动重连上。*/@OverridepublicvoidonConnectionRestored(URIremoteURI) {
logger.info("onConnectionRestored, remoteUri:{}", remoteURI);
        }
@OverridepublicvoidonInboundMessage(JmsInboundMessageDispatchenvelope) {}
@OverridepublicvoidonSessionClosed(Sessionsession, Throwablecause) {}
@OverridepublicvoidonConsumerClosed(MessageConsumerconsumer, Throwablecause) {}
@OverridepublicvoidonProducerClosed(MessageProducerproducer, Throwablecause) {}
    };
/*** 计算签名，password组装方法，请参见AMQP客户端接入说明文档。*/privatestaticStringdoSign(StringtoSignString, Stringsecret, StringsignMethod) throwsException {
SecretKeySpecsigningKey=newSecretKeySpec(secret.getBytes(), signMethod);
Macmac=Mac.getInstance(signMethod);
mac.init(signingKey);
byte[] rawHmac=mac.doFinal(toSignString.getBytes());
returnHex.encodeHexString(rawHmac);
    }
}