|
@@ -43,47 +43,106 @@ import java.util.function.Supplier;
|
|
|
@Component
|
|
|
@Order(value = 1)
|
|
|
@EnableAsync
|
|
|
-@ConditionalOnProperty(prefix = "spring", name="location", havingValue="Cloud")
|
|
|
+@ConditionalOnProperty(prefix = "spring", name = "location", havingValue = "Cloud")
|
|
|
public class alibabaiotgateway implements ApplicationRunner {
|
|
|
private final static Logger logger = LoggerFactory.getLogger(alibabaiotgateway.class);
|
|
|
+ //业务处理异步线程池,线程池参数可以根据您的业务特点调整,或者您也可以用其他异步方式处理接收到的消息。
|
|
|
+ private final static ExecutorService executorService = new ThreadPoolExecutor(
|
|
|
+ Runtime.getRuntime().availableProcessors(),
|
|
|
+ Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
|
|
|
+ new LinkedBlockingQueue(50000));
|
|
|
+ // 指定单个进程启动的连接数
|
|
|
+ // 单个连接消费速率有限,请参考使用限制,最大64个连接
|
|
|
+ // 连接数和消费速率及rebalance相关,建议每500QPS增加一个连接
|
|
|
+ private static final int connectionCount = 4;
|
|
|
+ private static final MessageListener messageListener = new MessageListener() {
|
|
|
+ @Override
|
|
|
+ public void onMessage(final Message message) {
|
|
|
+ try {
|
|
|
+ //1.收到消息之后一定要ACK。
|
|
|
+ // 推荐做法:创建Session选择Session.AUTO_ACKNOWLEDGE,这里会自动ACK。
|
|
|
+ // 其他做法:创建Session选择Session.CLIENT_ACKNOWLEDGE,这里一定要调message.acknowledge()来ACK。
|
|
|
+ // message.acknowledge();
|
|
|
+ //2.建议异步处理收到的消息,确保onMessage函数里没有耗时逻辑。
|
|
|
+ // 如果业务处理耗时过程过长阻塞住线程,可能会影响SDK收到消息后的正常回调。
|
|
|
+ executorService.submit(new Runnable() {
|
|
|
+ @Override
|
|
|
+ public void run() {
|
|
|
+ processMessage(message);
|
|
|
+ }
|
|
|
+ });
|
|
|
+ } catch (Exception e) {
|
|
|
+ logger.error("submit task occurs exception ", e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ };
|
|
|
+ private static final 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) {
|
|
|
+ }
|
|
|
+ };
|
|
|
@Value("${alibaba.iot.accessKey}")
|
|
|
- private String accessKey ;
|
|
|
+ private String accessKey;
|
|
|
@Value("${alibaba.iot.accessSecret}")
|
|
|
- private String accessSecret ;
|
|
|
+ private String accessSecret;
|
|
|
@Value("${alibaba.iot.consumerGroupId}")
|
|
|
- private String consumerGroupId ;
|
|
|
-
|
|
|
+ private String consumerGroupId;
|
|
|
//iotInstanceId:企业版实例请填写实例ID,公共实例请填空字符串""。
|
|
|
@Value("${alibaba.iot.iotInstanceId}")
|
|
|
- private String iotInstanceId ;
|
|
|
-
|
|
|
+ private String iotInstanceId;
|
|
|
//控制台服务端订阅中消费组状态页客户端ID一栏将显示clientId参数。
|
|
|
//建议使用机器UUID、MAC地址、IP等唯一标识等作为clientId。便于您区分识别不同的客户端。
|
|
|
@Value("${alibaba.iot.clientId}")
|
|
|
- private String clientId ;
|
|
|
-
|
|
|
+ private String clientId;
|
|
|
//${YourHost}为接入域名,请参见AMQP客户端接入说明文档。
|
|
|
@Value("${alibaba.iot.host}")
|
|
|
- private String host ;
|
|
|
+ private String host;
|
|
|
@Value("${alibaba.iot.proxy}")
|
|
|
- private boolean proxy;
|
|
|
- @Value("${alibaba.iot.proxyhost}")
|
|
|
- private String proxyhost;
|
|
|
- @Value("${alibaba.iot.proxyport}")
|
|
|
- private Integer proxyport;
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- // 指定单个进程启动的连接数
|
|
|
- // 单个连接消费速率有限,请参考使用限制,最大64个连接
|
|
|
- // 连接数和消费速率及rebalance相关,建议每500QPS增加一个连接
|
|
|
- private static int connectionCount = 4;
|
|
|
-
|
|
|
- //业务处理异步线程池,线程池参数可以根据您的业务特点调整,或者您也可以用其他异步方式处理接收到的消息。
|
|
|
- private final static ExecutorService executorService = new ThreadPoolExecutor(
|
|
|
- Runtime.getRuntime().availableProcessors(),
|
|
|
- Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
|
|
|
- new LinkedBlockingQueue(50000));
|
|
|
+ private boolean proxy;
|
|
|
|
|
|
|
|
|
// public static void main(String[] args) throws Exception {
|
|
@@ -154,28 +213,10 @@ public class alibabaiotgateway implements ApplicationRunner {
|
|
|
// logger.info("failed to handle messages");
|
|
|
// }
|
|
|
// }
|
|
|
-
|
|
|
- private static MessageListener messageListener = new MessageListener() {
|
|
|
- @Override
|
|
|
- public void onMessage(final Message message) {
|
|
|
- try {
|
|
|
- //1.收到消息之后一定要ACK。
|
|
|
- // 推荐做法:创建Session选择Session.AUTO_ACKNOWLEDGE,这里会自动ACK。
|
|
|
- // 其他做法:创建Session选择Session.CLIENT_ACKNOWLEDGE,这里一定要调message.acknowledge()来ACK。
|
|
|
- // message.acknowledge();
|
|
|
- //2.建议异步处理收到的消息,确保onMessage函数里没有耗时逻辑。
|
|
|
- // 如果业务处理耗时过程过长阻塞住线程,可能会影响SDK收到消息后的正常回调。
|
|
|
- executorService.submit(new Runnable() {
|
|
|
- @Override
|
|
|
- public void run() {
|
|
|
- processMessage(message);
|
|
|
- }
|
|
|
- });
|
|
|
- } catch (Exception e) {
|
|
|
- logger.error("submit task occurs exception ", e);
|
|
|
- }
|
|
|
- }
|
|
|
- };
|
|
|
+ @Value("${alibaba.iot.proxyhost}")
|
|
|
+ private String proxyhost;
|
|
|
+ @Value("${alibaba.iot.proxyport}")
|
|
|
+ private Integer proxyport;
|
|
|
|
|
|
/**
|
|
|
* 在这里处理您收到消息后的具体业务逻辑。
|
|
@@ -195,52 +236,6 @@ public class alibabaiotgateway implements ApplicationRunner {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- 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客户端接入说明文档。
|
|
|
*/
|
|
@@ -264,7 +259,7 @@ public class alibabaiotgateway implements ApplicationRunner {
|
|
|
String signMethod = "hmacsha1";
|
|
|
|
|
|
//userName组装方法,请参见AMQP客户端接入说明文档。
|
|
|
- String userName = clientId +"-" + i + "|authMode=aksign"
|
|
|
+ String userName = clientId + "-" + i + "|authMode=aksign"
|
|
|
+ ",signMethod=" + signMethod
|
|
|
+ ",timestamp=" + timeStamp
|
|
|
+ ",authId=" + accessKey
|
|
@@ -285,7 +280,7 @@ public class alibabaiotgateway implements ApplicationRunner {
|
|
|
Context context = new InitialContext(hashtable);
|
|
|
|
|
|
Connection connection = null;
|
|
|
- if (proxy){
|
|
|
+ if (proxy) {
|
|
|
logger.info(">>>>进入代理服务处理方法");
|
|
|
// context =new InitialContext(hashtable);
|
|
|
// JMSConnectionFactory cf1 = (JMSConnectionFactory) context.lookup("SBCF");
|
|
@@ -295,7 +290,7 @@ public class alibabaiotgateway implements ApplicationRunner {
|
|
|
// cf1.setExtension(JmsConnectionExtensions.PROXY_HANDLER_SUPPLIER.toString(),(connection1,remote)->{
|
|
|
// return proxyHandlerSupplier;
|
|
|
// });
|
|
|
- Supplier<ProxyHandler> proxyHandlerSupplier =()-> new HttpProxyHandler(new InetSocketAddress(proxyhost,proxyport));
|
|
|
+ Supplier<ProxyHandler> proxyHandlerSupplier = () -> new HttpProxyHandler(new InetSocketAddress(proxyhost, proxyport));
|
|
|
JmsConnectionFactory factory = (JmsConnectionFactory) context.lookup("SBCF");
|
|
|
factory.setExtension(JmsConnectionExtensions.PROXY_HANDLER_SUPPLIER.toString(), (connection1, remote) -> {
|
|
|
// SocketAddress proxyAddress = new InetSocketAddress(proxyhost, proxyport);
|
|
@@ -306,16 +301,16 @@ public class alibabaiotgateway implements ApplicationRunner {
|
|
|
});
|
|
|
connection = factory.createConnection(userName, password);
|
|
|
logger.info(">>>>>>>创建阿里iot通过招商代理访问互联网");
|
|
|
- }else{
|
|
|
- ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF");
|
|
|
+ } else {
|
|
|
+ ConnectionFactory cf = (ConnectionFactory) context.lookup("SBCF");
|
|
|
// 创建连接。
|
|
|
- connection= cf.createConnection(userName, password);
|
|
|
+ connection = cf.createConnection(userName, password);
|
|
|
}
|
|
|
|
|
|
- Destination queue = (Destination)context.lookup("QUEUE");
|
|
|
+ Destination queue = (Destination) context.lookup("QUEUE");
|
|
|
connections.add(connection);
|
|
|
|
|
|
- ((JmsConnection)connection).addConnectionListener(myJmsConnectionListener);
|
|
|
+ ((JmsConnection) connection).addConnectionListener(myJmsConnectionListener);
|
|
|
// 创建会话。
|
|
|
// Session.CLIENT_ACKNOWLEDGE: 收到消息后,需要手动调用message.acknowledge()。
|
|
|
// Session.AUTO_ACKNOWLEDGE: SDK自动ACK(推荐)。
|
|
@@ -334,7 +329,7 @@ public class alibabaiotgateway implements ApplicationRunner {
|
|
|
Thread.sleep(60 * 1000);
|
|
|
logger.info("run shutdown");
|
|
|
|
|
|
- connections.forEach(c-> {
|
|
|
+ connections.forEach(c -> {
|
|
|
try {
|
|
|
c.close();
|
|
|
} catch (JMSException e) {
|