herb.git - Gitblit

			@@ -6,6 +6,7 @@
			import com.alibaba.fastjson.JSONObject;
			import com.alibaba.fastjson.TypeReference;
			import lombok.extern.slf4j.Slf4j;
			import org.apache.poi.ss.formula.functions.T;
			import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
			import org.eclipse.paho.client.mqttv3.MqttCallback;
			import org.eclipse.paho.client.mqttv3.MqttMessage;
			@@ -22,13 +23,17 @@
			import org.jeecg.modules.dry.entity.DryShop;
			import org.jeecg.modules.dry.service.*;
			import org.jeecg.modules.dry.vo.DryEquipmentVo;
			import org.jeecg.modules.dry.vo.DryFaultRecordVo;
			import org.jeecg.modules.dry.vo.RealTimeDataVo;
			import org.springframework.beans.factory.annotation.Autowired;
			import org.springframework.beans.factory.annotation.Value;
			import org.springframework.context.annotation.Scope;
			import org.springframework.stereotype.Component;

			import java.util.ArrayList;
			import java.util.List;
			import java.util.Map;
			import java.util.stream.Collectors;

			@Slf4j
			@Component
			@@ -62,8 +67,6 @@
			private IDryFaultRecordService faultRecordService;




			@Override
			public void connectionLost(Throwable throwable) {
			System.err.println("连接断开：：掉线");
			@@ -82,7 +85,7 @@
			JSONObject messageJson = JSONObject.parseObject(message);

			if (topic.startsWith("$SYS/brokers/") && topic.endsWith("connected")) {
			JSONObject client = (JSONObject) redisUtil.get(MqttConstant.MQTT_ONLINE_CLIENT + messageJson.get("clientid"));
			JSONObject client = (JSONObject) redisUtil.get(String.format(MqttConstant.MQTT_ONLINE_CLIENT,messageJson.get("clientid").toString().split("-")[1]) + messageJson.get("clientid"));
			if (client == null) {
			JSONObject item = new JSONObject();
			//username
			@@ -105,7 +108,7 @@
			}catch (Exception e){
			e.printStackTrace();
			}
			redisUtil.set(MqttConstant.MQTT_ONLINE_CLIENT + clientid, item);
			redisUtil.hset(String.format(MqttConstant.MQTT_ONLINE_CLIENT,item.get("tenantId") ), clientid, item);
			System.err.println(String.format("设备: %s上线", clientid));
			}

			@@ -113,7 +116,7 @@
			if (topic.startsWith("$SYS/brokers/") && topic.endsWith("disconnected")) {
			try {
			String clientid = messageJson.getString("clientid");
			redisUtil.del(MqttConstant.MQTT_ONLINE_CLIENT + clientid);
			redisUtil.hdel(String.format(MqttConstant.MQTT_ONLINE_CLIENT,clientid.split("-")[1]), clientid);
			System.err.println(String.format("设备: %s下线", clientid));
			} catch (Exception e) {
			e.printStackTrace();
			@@ -160,7 +163,7 @@
			}
			// 实时数据上传太频繁且数据内容超过字段大小不记录日志
			if (!topic.equals(MqttConstant.TENANT_UP_PREFIX_REALTIME_DATA) && !topic.equals(MqttConstant.TENANT_UP_PREFIX_FAULT_DATA)){
			baseCommonService.addLog(message, CommonConstant.LOG_TYPE_MQTT, CommonConstant.OPERATE_MQTT_1);
			// baseCommonService.addLog(message, CommonConstant.LOG_TYPE_MQTT, CommonConstant.OPERATE_MQTT_1);
			}

			switch (topic) {
			@@ -169,7 +172,7 @@
			System.err.println("admin收到" + topic);
			// 根据设备id查询设备mqtt在线状态
			String clientId = messageJson.getString("clientId");
			JSONObject client = (JSONObject) redisUtil.get(MqttConstant.MQTT_ONLINE_CLIENT + clientId);
			JSONObject client = (JSONObject) redisUtil.hget(String.format(MqttConstant.MQTT_ONLINE_CLIENT,clientId.split("-")[1]) , clientId);

			ThreadUtil.execute(() -> {

			@@ -212,15 +215,44 @@
			});

			break;
			//各租户上传的实时报警数据
			case MqttConstant.TENANT_UP_PREFIX_REAL_FAULT_DATA:
			MqMessage<Map<String, DryFaultRecordVo>> realFaultMessage = JSON.parseObject(message, new TypeReference<MqMessage<Map<String, DryFaultRecordVo>>>() {
			});
			//故障数据
			Map<String, DryFaultRecordVo> dryFaultMap = realFaultMessage.getData();
			//租户id
			String tentId = realFaultMessage.getTentId();
			//收到租户实时报警数据存入redis
			//转换为 Map<String, Object>
			Map<String, Object> objectMap = dryFaultMap.entrySet().stream()
			.collect(Collectors.toMap(
			Map.Entry::getKey,
			entry -> (Object) entry.getValue()
			));
			redisUtil.hmset(String.format(MqttConstant.MQTT_CLOUD_REAL_FAULT,realFaultMessage.getTentId()), objectMap);
			//广播发送给各租户下移动设备
			if(dryFaultMap.isEmpty()){
			return;
			}
			String recTopic = String.format(MqttConstant.SERVICE_BROADCAST_TENANT_REAL_FAULT, tentId);
			//数据转换
			List<DryFaultRecordVo> faultList = new ArrayList<DryFaultRecordVo>((dryFaultMap.values()));
			MqMessage< List<DryFaultRecordVo>> mqMessage = new MqMessage<>(faultList,tentId,recTopic);
			//发送广播
			System.err.println("广播给：" + recTopic);
			sendMqttMessage(MqttConstant.SERVICE_BROADCAST_TENANT_REAL_FAULT,mqMessage);

			break;
			// 接收设备报警数据
			case MqttConstant.TENANT_UP_PREFIX_FAULT_DATA:
			ThreadUtil.execute(() -> {
			try {
			MqMessage<List<DryFaultRecord>> listMqMessage = JSON.parseObject(message, new TypeReference<MqMessage<List<DryFaultRecord>>>() {
			MqMessage<List<DryFaultRecord>> faultMessage = JSON.parseObject(message, new TypeReference<MqMessage<List<DryFaultRecord>>>() {
			});
			// List<DryFaultRecord> faultRecords = JSON.parseArray(jsonObject.get("data").toString(), DryFaultRecord.class);
			System.err.println(listMqMessage.toString());
			faultRecordService.saveBatch(listMqMessage.getData());
			System.err.println(faultMessage.toString());
			faultRecordService.saveBatch(faultMessage.getData());

			} catch (Exception e) {
			e.printStackTrace();
			@@ -335,11 +367,30 @@
			});



			break;

			}



			}

			/**
			* 发送消息
			* @param topic 订阅
			* @param mqMessage 消息体
			*/
			private void sendMqttMessage(String topic, MqMessage mqMessage){
			ThreadUtil.execute(() -> {
			try {
			MqttMessage sendMessage = new MqttMessage(JSONObject.toJSONString(mqMessage).getBytes());
			sendMessage.setQos(0);
			mqttUtil.getMqttClient().publish(String.format(topic, mqMessage.getTentId()), sendMessage);
			}catch (Exception e){
			e.printStackTrace();
			}
			});
			}


			}