Merge remote-tracking branch 'origin/master'

src/main/java/com/techsor/datacenter/sender/dao/DashboardAlertDao.java

@@ -1,5 +1,6 @@
 package com.techsor.datacenter.sender.dao;
 
+import com.techsor.datacenter.sender.config.DataSourceContextHolder;
 import com.techsor.datacenter.sender.disruptor.AlertEvent;
 import com.techsor.datacenter.sender.entitiy.AlertHistoryDTO;
 import com.techsor.datacenter.sender.entitiy.DynamodbEntity;
@@ -9,6 +10,8 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang.StringUtils;
@@ -26,6 +29,8 @@ public class DashboardAlertDao {
 
     @Resource
     private JdbcTemplate jdbcTemplate;
+	@Resource
+	private CompanyInfoDao companyInfoDao;
 
 	public void upsertDeviceRawData(DynamodbEntity entity) {
 		if (StringUtils.isEmpty(entity.getDeviceId())) {
@@ -63,49 +68,62 @@ public class DashboardAlertDao {
 	}
 
 	public void batchUpsertRawData(List<AlertEvent> list) {
-		jdbcTemplate.batchUpdate(
-				"INSERT INTO device_rawdata_realtime (" +
-						"device_id, building_id, status, receive_ts, alert_title, alert_content, " +
-						"alert_cancel_title, alert_cancel_content, raw_data, upload_year, upload_month, upload_day) " +
-						"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " +
-						"ON DUPLICATE KEY UPDATE " +
-						"building_id = VALUES(building_id), " +
-						"status = VALUES(status), " +
-						"receive_ts = VALUES(receive_ts), " +
-						"alert_title = VALUES(alert_title), " +
-						"alert_content = VALUES(alert_content), " +
-						"alert_cancel_title = VALUES(alert_cancel_title), " +
-						"alert_cancel_content = VALUES(alert_cancel_content), " +
-						"raw_data = VALUES(raw_data), " +
-						"upload_year = VALUES(upload_year), " +
-						"upload_month = VALUES(upload_month), " +
-						"upload_day = VALUES(upload_day)",
-				new BatchPreparedStatementSetter() {
-
-					@Override
-					public void setValues(PreparedStatement ps, int i) throws SQLException {
-						AlertEvent e = list.get(i);
-
-						ps.setString(1, e.getEntity().getDeviceId());
-						ps.setLong(2, e.getEntity().getDbBuildingId());
-						ps.setString(3, e.getEntity().getStatus());
-						ps.setLong(4, e.getEntity().getReceive_ts());
-						ps.setString(5, e.getEntity().getAlertTitle());
-						ps.setString(6, e.getEntity().getAlertContent());
-						ps.setString(7, e.getEntity().getAlertCancelTitle());
-						ps.setString(8, e.getEntity().getAlertCancelContent());
-						ps.setString(9, e.getEntity().getRawData());
-						ps.setInt(10, Integer.parseInt(e.getEntity().getYearKey()));
-						ps.setInt(11, Integer.parseInt(e.getEntity().getMonthKey()));
-						ps.setInt(12, Integer.parseInt(e.getEntity().getDayKey()));
-					}
-
-					@Override
-					public int getBatchSize() {
-						return list.size();
-					}
-				}
-		);
+		// 按 companyId 分组
+		Map<Long, List<AlertEvent>> grouped = list.stream()
+				.filter(e -> e.getEntity() != null && e.getEntity().getCompanyId() != null)
+				.collect(Collectors.groupingBy(e -> e.getEntity().getCompanyId()));
+
+		grouped.forEach((companyId, subList) -> {
+			long topCompanyId = companyInfoDao.getTopCompanyId(String.valueOf(companyId));
+			String dsKey = "dataSourceForCompany_" + topCompanyId;
+			try {
+				DataSourceContextHolder.setCurrentDataSourceKey(dsKey);
+
+				// 批量执行 upsert
+				jdbcTemplate.batchUpdate(
+						"INSERT INTO device_rawdata_realtime (" +
+								"device_id, building_id, status, receive_ts, alert_title, alert_content, alert_cancel_title," +
+								"alert_cancel_content, raw_data, upload_year, upload_month, upload_day) " +
+								"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " +
+								"ON DUPLICATE KEY UPDATE " +
+								"building_id = VALUES(building_id), " +
+								"status = VALUES(status), " +
+								"receive_ts = VALUES(receive_ts)," +
+								"alert_title = VALUES(alert_title)," +
+								"alert_content = VALUES(alert_content)," +
+								"alert_cancel_title = VALUES(alert_cancel_title)," +
+								"alert_cancel_content = VALUES(alert_cancel_content)," +
+								"raw_data = VALUES(raw_data)," +
+								"upload_year = VALUES(upload_year)," +
+								"upload_month = VALUES(upload_month)," +
+								"upload_day = VALUES(upload_day)",
+						new BatchPreparedStatementSetter() {
+
+							public void setValues(PreparedStatement ps, int i) throws SQLException {
+								DynamodbEntity entity = subList.get(i).getEntity();
+								ps.setString(1, entity.getDeviceId());
+								ps.setLong(2, entity.getDbBuildingId());
+								ps.setString(3, entity.getStatus());
+								ps.setLong(4, entity.getReceive_ts());
+								ps.setString(5, entity.getAlertTitle());
+								ps.setString(6, entity.getAlertContent());
+								ps.setString(7, entity.getAlertCancelTitle());
+								ps.setString(8, entity.getAlertCancelContent());
+								ps.setString(9, entity.getRawData());
+								ps.setInt(10, Integer.parseInt(entity.getYearKey()));
+								ps.setInt(11, Integer.parseInt(entity.getMonthKey()));
+								ps.setInt(12, Integer.parseInt(entity.getDayKey()));
+							}
+
+							public int getBatchSize() {
+								return subList.size();
+							}
+						}
+				);
+			} finally {
+				DataSourceContextHolder.clearCurrentDataSourceKey();
+			}
+		});
 	}
 
 	public void insertAlertHistory(DynamodbEntity entity) {

src/main/java/com/techsor/datacenter/sender/dao/DeviceAlertConfigDao.java

@@ -2,6 +2,7 @@ package com.techsor.datacenter.sender.dao;
 
 import com.techsor.datacenter.sender.dto.DeviceAlertInfo;
 import com.techsor.datacenter.sender.entitiy.DeviceAlertConfigEntity;
+import org.springframework.cache.annotation.Cacheable;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Component;
 
@@ -31,6 +32,7 @@ public class DeviceAlertConfigDao {
 
     **/
 
+    @Cacheable(value = "DeviceAlertConfigDao::selectCurrentDeviceAlertConfigByDeviceId", key = "#deviceId")
     public List<DeviceAlertConfigEntity> selectCurrentDeviceAlertConfigByDeviceId(Long deviceId){
         String sql="select * from device_alert_config where device_config_id="+deviceId+" and flag!=1";
 

src/main/java/com/techsor/datacenter/sender/dao/DeviceAlertTemplateBindDao.java

@@ -1,6 +1,7 @@
 package com.techsor.datacenter.sender.dao;
 
 
+import org.springframework.cache.annotation.Cacheable;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Component;
 
@@ -13,6 +14,7 @@ public class DeviceAlertTemplateBindDao {
     @Resource
     private JdbcTemplate jdbcTemplate;
 
+    @Cacheable(value = "DeviceAlertTemplateBindDao::selectDeviceAlertTemplateBindByDeviceId", key = "#deviceId")
     public List<Long> selectDeviceAlertTemplateBindByDeviceId(Long deviceId){
         String sql="select device_alert_template_id from device_alert_template_bind where device_info_id="+deviceId;
         return this.jdbcTemplate.query(sql,(rs, rowNum)-> rs.getLong("device_alert_template_id"));

src/main/java/com/techsor/datacenter/sender/dao/DeviceAlertTemplateDao.java

@@ -1,6 +1,7 @@
 package com.techsor.datacenter.sender.dao;
 
 import com.techsor.datacenter.sender.entitiy.DeviceAlertTemplateEntity;
+import org.springframework.cache.annotation.Cacheable;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Component;
 
@@ -35,6 +36,7 @@ public class DeviceAlertTemplateDao {
     @Resource
     private JdbcTemplate jdbcTemplate;
 
+    @Cacheable(value = "DeviceAlertTemplateDao::selectDeviceAlertTemplatesByDeviceId", key = "#deviceId")
     public List<DeviceAlertInfo> selectDeviceAlertTemplatesByDeviceId(Long deviceId) {
         String sql = "SELECT tmp.id, " +
                 "tmp.company_id AS companyId, " +

src/main/java/com/techsor/datacenter/sender/dao/DeviceDao.java

@@ -227,6 +227,7 @@ public class DeviceDao {
     }
 
 
+    @Cacheable(value = "DeviceDao::getDeviceInfo", key = "#deviceId")
     public List<DeviceInfoVO> getDeviceInfo(String deviceId){
         if (StringUtils.isEmpty(deviceId)){
             return new ArrayList<>();

src/main/java/com/techsor/datacenter/sender/dao/DeviceForwardConfigDao.java

@@ -1,6 +1,7 @@
 package com.techsor.datacenter.sender.dao;
 
 import com.techsor.datacenter.sender.entitiy.DeviceForwardConfigEntity;
+import org.springframework.cache.annotation.Cacheable;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Component;
 
@@ -14,6 +15,7 @@ public class DeviceForwardConfigDao {
     private JdbcTemplate jdbcTemplate ;
 
 
+    @Cacheable(value = "DeviceForwardConfigDao::selectDeviceForwardConfigByDeviceId", key = "#deviceId")
     public List<DeviceForwardConfigEntity> selectDeviceForwardConfigByDeviceId(Long deviceId){
         String sql="select * from device_forward_config where device_id="+deviceId;
         sql=String.format(sql,deviceId);

src/main/java/com/techsor/datacenter/sender/dao/MqttConfigDao.java

@@ -3,6 +3,7 @@ package com.techsor.datacenter.sender.dao;
 import com.techsor.datacenter.sender.entitiy.MqttConfigEntity;
 import com.techsor.datacenter.sender.entitiy.bastatus.BaStatusEntity;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.cache.annotation.Cacheable;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Component;
 
@@ -23,6 +24,7 @@ public class MqttConfigDao {
      * @param purposeType 
      * @return
      */
+    @Cacheable(value = "MqttConfigDao::getByDeviceId", key = "#deviceId + '::' + #purposeType")
     public List<MqttConfigEntity> getByDeviceId(String deviceId, int purposeType) {
         String sql="SELECT "
         		+ "bmqtt.id mqttId, bmqtt.company_id, bcomp.company_name, "

src/main/java/com/techsor/datacenter/sender/dto/DeviceAlertInfo.java

@@ -3,8 +3,10 @@ package com.techsor.datacenter.sender.dto;
 
 import lombok.Data;
 
+import java.io.Serializable;
+
 @Data
-public class DeviceAlertInfo {
+public class DeviceAlertInfo implements Serializable {
 
     private Integer id;
     private Long deviceConfigId;

src/main/java/com/techsor/datacenter/sender/dto/DeviceInfoVO.java

@@ -2,12 +2,14 @@ package com.techsor.datacenter.sender.dto;
 
 import lombok.Data;
 
+import java.io.Serializable;
+
 /** 
 * @author Mr.Jiang
 * @time 2022年7月21日 下午8:50:31 
 */
 @Data
-public class DeviceInfoVO{
+public class DeviceInfoVO implements Serializable {
 
 	private String deviceId;
 	

src/main/java/com/techsor/datacenter/sender/service/impl/DataProcessServiceImpl.java

@@ -622,17 +622,17 @@ public class DataProcessServiceImpl implements IDataProcessService {
         }
         baseTransDataEntity.setHashId(UUID.randomUUID());
 
-//        try {
-//            handleDashboardAlert(baseTransDataEntity);
-//        } catch (Exception e) {
-//            log.error("dashboard alert error", e);
-//        }
-//
-//        try {
-//            minuteLevelStorage(baseTransDataEntity);
-//        } catch (Exception e) {
-//            log.error("minuteLevelStorage error", e);
-//        }
+        try {
+            handleDashboardAlert(baseTransDataEntity);
+        } catch (Exception e) {
+            log.error("dashboard alert error", e);
+        }
+
+        try {
+            minuteLevelStorage(baseTransDataEntity);
+        } catch (Exception e) {
+            log.error("minuteLevelStorage error", e);
+        }
 
 //        try {
 //        	if ("alert".equals(baseTransDataEntity.getStatus())) {