|
|
@@ -0,0 +1,307 @@
|
|
|
+package com.tuoren.forward.config.amqp;
|
|
|
+
|
|
|
+import java.net.URI;
|
|
|
+import java.util.Date;
|
|
|
+import java.util.Hashtable;
|
|
|
+import java.util.concurrent.ExecutorService;
|
|
|
+import java.util.concurrent.LinkedBlockingQueue;
|
|
|
+import java.util.concurrent.ThreadPoolExecutor;
|
|
|
+import java.util.concurrent.TimeUnit;
|
|
|
+
|
|
|
+import javax.crypto.Mac;
|
|
|
+import javax.crypto.spec.SecretKeySpec;
|
|
|
+import javax.jms.Connection;
|
|
|
+import javax.jms.ConnectionFactory;
|
|
|
+import javax.jms.Destination;
|
|
|
+import javax.jms.Message;
|
|
|
+import javax.jms.MessageConsumer;
|
|
|
+import javax.jms.MessageListener;
|
|
|
+import javax.jms.MessageProducer;
|
|
|
+import javax.jms.Session;
|
|
|
+import javax.naming.Context;
|
|
|
+import javax.naming.InitialContext;
|
|
|
+
|
|
|
+import org.apache.commons.codec.binary.Base64;
|
|
|
+import org.apache.commons.lang3.StringUtils;
|
|
|
+import org.apache.qpid.jms.JmsConnection;
|
|
|
+import org.apache.qpid.jms.JmsConnectionListener;
|
|
|
+import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
|
|
|
+import org.springframework.beans.factory.annotation.Autowired;
|
|
|
+import org.springframework.boot.context.properties.ConfigurationProperties;
|
|
|
+import org.springframework.data.mongodb.core.MongoTemplate;
|
|
|
+import org.springframework.stereotype.Component;
|
|
|
+
|
|
|
+import com.alibaba.fastjson2.JSONObject;
|
|
|
+import com.tuoren.forward.config.mqtt.MqttUtil;
|
|
|
+import com.tuoren.forward.constant.CommonConstant;
|
|
|
+import com.tuoren.forward.entity.Device;
|
|
|
+import com.tuoren.forward.entity.User;
|
|
|
+import com.tuoren.forward.mapper.DeviceMapper;
|
|
|
+import com.tuoren.forward.mapper.UserMapper;
|
|
|
+import com.tuoren.forward.util.RemoteUtil;
|
|
|
+import com.tuoren.forward.util.UUIDUtil;
|
|
|
+
|
|
|
+import lombok.Setter;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+
|
|
|
+@Slf4j
|
|
|
+@Component
|
|
|
+@ConfigurationProperties(prefix = "aliyun1")
|
|
|
+@Setter
|
|
|
+public class AmqpClient1 {
|
|
|
+
|
|
|
+ private String accessKey;
|
|
|
+ private String accessSecret;
|
|
|
+ private String consumerGroupId;
|
|
|
+
|
|
|
+ //iotInstanceId:实例ID。若是2021年07月30日之前(不含当日)开通的公共实例,请填空字符串。
|
|
|
+ private String iotInstanceId;
|
|
|
+
|
|
|
+ //控制台服务端订阅中消费组状态页客户端ID一栏将显示clientId参数。
|
|
|
+ private String clientId;
|
|
|
+
|
|
|
+ //${YourHost}为接入域名,请参见AMQP客户端接入说明文档。
|
|
|
+ private String amqpHost;
|
|
|
+
|
|
|
+// @Autowired
|
|
|
+// @Qualifier("nbJdbcTemplate")
|
|
|
+// private JdbcTemplate nbJdbcTemplate;
|
|
|
+
|
|
|
+ @Autowired(required = false)
|
|
|
+ private MqttUtil mqttUtil;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ RemoteUtil remoteUtil;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ DeviceMapper deviceMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ UserMapper userMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ MongoTemplate mongoTemplate;
|
|
|
+
|
|
|
+ //业务处理异步线程池,线程池参数可以根据您的业务特点调整,或者您也可以用其他异步方式处理接收到的消息。
|
|
|
+ private final static ExecutorService executorService = new ThreadPoolExecutor(
|
|
|
+ Runtime.getRuntime().availableProcessors(),
|
|
|
+ Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.SECONDS,
|
|
|
+ new LinkedBlockingQueue<Runnable>(50000));
|
|
|
+
|
|
|
+ public void connection() throws Exception{
|
|
|
+
|
|
|
+
|
|
|
+ long timeStamp = System.currentTimeMillis();
|
|
|
+ String signMethod = "hmacsha1";
|
|
|
+ String userName = clientId + "-" + 0 + "|authMode=aksign"
|
|
|
+ + ",signMethod=" + signMethod
|
|
|
+ + ",timestamp=" + timeStamp
|
|
|
+ + ",authId=" + accessKey
|
|
|
+ + ",iotInstanceId=" + iotInstanceId
|
|
|
+ + ",consumerGroupId=" + consumerGroupId
|
|
|
+ + "|";
|
|
|
+ String signContent = "authId=" + accessKey + "×tamp=" + timeStamp;
|
|
|
+ String password = doSign(signContent, accessSecret, signMethod);
|
|
|
+ String connectionUrl = "failover:(amqps://" + amqpHost + ":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");
|
|
|
+
|
|
|
+ // 创建连接。
|
|
|
+ Connection connection = cf.createConnection(userName, password);
|
|
|
+
|
|
|
+ ((JmsConnection)connection).addConnectionListener(jmsConnectionListener);
|
|
|
+ // 创建会话。
|
|
|
+ // Session.CLIENT_ACKNOWLEDGE: 收到消息后,需要手动调用message.acknowledge()。
|
|
|
+ // Session.AUTO_ACKNOWLEDGE: SDK自动ACK(推荐)。
|
|
|
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
|
|
+
|
|
|
+ connection.start();
|
|
|
+ // 创建Receiver连接。
|
|
|
+ MessageConsumer consumer = session.createConsumer(queue);
|
|
|
+ consumer.setMessageListener(messageListener);
|
|
|
+ }
|
|
|
+
|
|
|
+ private 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) {
|
|
|
+ log.error("submit task occurs exception ", e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 在这里处理您收到消息后的具体业务逻辑。
|
|
|
+ */
|
|
|
+ private void processMessage(Message message) {
|
|
|
+ try {
|
|
|
+ String topic = message.getStringProperty("topic");
|
|
|
+ if(!topic.endsWith("/thing/event/property/post")) {
|
|
|
+ return ;
|
|
|
+ }
|
|
|
+ byte[] body = message.getBody(byte[].class);
|
|
|
+ String content = new String(body);
|
|
|
+ log.info("下发数据内容1:"+content);
|
|
|
+ JSONObject json = JSONObject.parseObject(content);
|
|
|
+ repackJson(json);
|
|
|
+ String deviceName = json.getString("deviceName");
|
|
|
+ String productKey = json.getString("productKey");
|
|
|
+ String hospitalCode = "";
|
|
|
+ JSONObject items = json.getJSONObject("items");
|
|
|
+ long ci = 0;
|
|
|
+ long lac = 0;
|
|
|
+ if(items.containsKey("userId")) {
|
|
|
+ hospitalCode = items.getString("userId");
|
|
|
+ }
|
|
|
+ if(items.containsKey("ci")) {
|
|
|
+ ci = items.getLong("ci");
|
|
|
+ }
|
|
|
+ if(items.containsKey("lac")) {
|
|
|
+ lac = items.getLong("lac");
|
|
|
+ }
|
|
|
+ if(ci > 0 && lac>0) {
|
|
|
+ String location = remoteUtil.localtion(String.valueOf(ci), String.valueOf(lac));
|
|
|
+ items.put("location", JSONObject.parseObject(location));
|
|
|
+ }
|
|
|
+ if(StringUtils.isEmpty(hospitalCode)) {
|
|
|
+// String sql = "SELECT h.code FROM bus_hospital h inner join bus_device d on d.tenant_id = h.tenant_id where device_id=? limit 1";
|
|
|
+// try {
|
|
|
+// hospitalCode = nbJdbcTemplate.queryForObject(sql, String.class, deviceName);
|
|
|
+// }catch (Exception e) {
|
|
|
+// // TODO: handle exception
|
|
|
+// }
|
|
|
+ hospitalCode = remoteUtil.getHospitalCode(deviceName);
|
|
|
+ }
|
|
|
+
|
|
|
+ if(StringUtils.isEmpty(hospitalCode)) {
|
|
|
+ log.info("设备没有对应医院:"+deviceName);
|
|
|
+ }else {
|
|
|
+ User owner = userMapper.selectByCode(hospitalCode);
|
|
|
+ if(owner == null) {
|
|
|
+ log.info("不存在该医院:"+hospitalCode);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ JSONObject mongoJson = new JSONObject();
|
|
|
+ Device exist = deviceMapper.selectByMac(deviceName);
|
|
|
+ if(exist == null) {
|
|
|
+ Device record = new Device();
|
|
|
+ String deviceId = UUIDUtil.get32UUID();
|
|
|
+ record.setId(deviceId);
|
|
|
+ record.setCreatetime(new Date());
|
|
|
+ record.setMac(deviceName);
|
|
|
+ record.setProductId(productKey);
|
|
|
+ record.setTenantId(owner.getId());
|
|
|
+ record.setData(items.toString());
|
|
|
+ deviceMapper.insertSelective(record);
|
|
|
+ mongoJson.put("deviceId", deviceId);
|
|
|
+ }else {
|
|
|
+ Device record = new Device();
|
|
|
+ record.setId(exist.getId());
|
|
|
+ record.setTenantId(owner.getId());
|
|
|
+ record.setData( items.toString());
|
|
|
+ deviceMapper.updateByPrimaryKeySelective2(record);
|
|
|
+ mongoJson.put("deviceId", exist.getId());
|
|
|
+ }
|
|
|
+ mongoJson.put("deviceMac", deviceName);
|
|
|
+ mongoJson.put("time", json.getLong("gmtCreate"));
|
|
|
+ mongoJson.put("tenantId", owner.getId());
|
|
|
+ mongoJson.put("items", items);
|
|
|
+ mongoTemplate.insert(mongoJson, CommonConstant.MONGO_COLLECTION_DEVICE_RECORD);
|
|
|
+
|
|
|
+ if(mqttUtil != null) {
|
|
|
+ log.info("mqtt publish1:"+ json.toString());
|
|
|
+ mqttUtil.pub(CommonConstant.MQTT_PUBLISH_PREFIX+hospitalCode, json.toString());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("processMessage occurs error ", e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private static JmsConnectionListener jmsConnectionListener = new JmsConnectionListener() {
|
|
|
+ /**
|
|
|
+ * 连接成功建立。
|
|
|
+ */
|
|
|
+ @Override
|
|
|
+ public void onConnectionEstablished(URI remoteURI) {
|
|
|
+ log.info("onConnectionEstablished, remoteUri:{}", remoteURI);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 尝试过最大重试次数之后,最终连接失败。
|
|
|
+ */
|
|
|
+ @Override
|
|
|
+ public void onConnectionFailure(Throwable error) {
|
|
|
+ log.error("onConnectionFailure, {}", error.getMessage());
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 连接中断。
|
|
|
+ */
|
|
|
+ @Override
|
|
|
+ public void onConnectionInterrupted(URI remoteURI) {
|
|
|
+ log.info("onConnectionInterrupted, remoteUri:{}", remoteURI);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 连接中断后又自动重连上。
|
|
|
+ */
|
|
|
+ @Override
|
|
|
+ public void onConnectionRestored(URI remoteURI) {
|
|
|
+ log.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);
|
|
|
+ }
|
|
|
+
|
|
|
+ public static void repackJson(JSONObject json) {
|
|
|
+ JSONObject items = json.getJSONObject("items");
|
|
|
+ for(String key:items.keySet()) {
|
|
|
+ JSONObject temp = items.getJSONObject(key);
|
|
|
+ items.replace(key, temp.get("value"));
|
|
|
+ }
|
|
|
+ json.replace("items", items);
|
|
|
+ }
|
|
|
+}
|