同步抄表和动态rabbit

smart-city-platform/src/main/java/com/bz/smart_city/commom/security/WebSecurityConfig.java

@@ -110,7 +110,8 @@ public class WebSecurityConfig extends WebSecurityConfigurerAdapter {
                 .antMatchers("/device/synArchies","/installPlan/syncPlan")
                 .antMatchers("/druid/**").antMatchers("/syncData/**")
                 .antMatchers("/device/valve/update")
-                .antMatchers("/integration/user/save","/integration/user/del");
+                .antMatchers("/integration/user/save","/integration/user/del")
+                .antMatchers("/rabbit/**");
 
     }
 

smart-city-platform/src/main/java/com/bz/smart_city/config/FanoutRabbitConfig.java

@@ -14,7 +14,7 @@ import java.util.Map;
 @Configuration
 public class FanoutRabbitConfig {
 
-    @Value("${spring.rabbitmq.zy.topic}")
+    /*@Value("${spring.rabbitmq.zy.topic}")
     String topicName ;
  
     @Bean
@@ -35,6 +35,6 @@ public class FanoutRabbitConfig {
     @Bean
     Binding bindingExchange() {
         return BindingBuilder.bind(queue()).to(fanoutExchange());
-    }
+    }*/
 
 }

smart-city-platform/src/main/java/com/bz/smart_city/controller/common/RabbitController.java

@@ -0,0 +1,100 @@
+package com.bz.smart_city.controller.common;
+
+import com.bz.smart_city.commom.exception.ServiceException;
+import com.bz.smart_city.commom.model.AjaxMessage;
+import com.bz.smart_city.commom.model.ResultStatus;
+import com.bz.smart_city.dao.ScRabbitConfigMapper;
+import com.bz.smart_city.entity.ScRabbitConfig;
+import com.bz.smart_city.rabbitmq.DynamicRoutingConnectionFactory;
+import com.bz.smart_city.service.impl.GeneralRabbitSyncService;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiParam;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.SimpleResourceHolder;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+
+@Slf4j
+@Controller
+@ResponseBody
+@RequestMapping("rabbit")
+@Api(tags = "测试")
+public class RabbitController {
+
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+    @Autowired
+    private ScRabbitConfigMapper scRabbitConfigMapper;
+    @Autowired
+    GeneralRabbitSyncService generalRabbitSyncService;
+
+
+    @GetMapping("test")
+    @ApiOperation(value = "test")
+    public AjaxMessage test(@ApiParam(value = "lookupKey", required = true) @RequestParam String lookupKey,
+                       @ApiParam(value = "payload", required = true) @RequestParam String payload) {
+
+        ScRabbitConfig scRabbitConfig = scRabbitConfigMapper.findByCode(lookupKey);
+        if (scRabbitConfig != null) {
+            //设置当前线程lookupKey，内部由ThreadLocal实现
+            SimpleResourceHolder.bind(rabbitTemplate.getConnectionFactory(), scRabbitConfig.getCode());
+            //业务操作会根据线程中的lookupKey从routeConnectionFactory的targetConnectionFactories中选择对应的connectionFactory
+            rabbitTemplate.convertAndSend(scRabbitConfig.getExchange(),scRabbitConfig.getRoutingKey(), payload);
+            //操作完以后记得解绑。不影响线程的后序其他工厂操作
+            SimpleResourceHolder.unbind(rabbitTemplate.getConnectionFactory());
+        }else {
+            throw new ServiceException(-900,"lookupKey不存在");
+        }
+
+        return new AjaxMessage(ResultStatus.OK);
+    }
+
+    @GetMapping("testPush")
+    @ApiOperation(value = "testPush")
+    public AjaxMessage testPush(@ApiParam(value = "lookupKey", required = true) @RequestParam String lookupKey) {
+
+        ScRabbitConfig scRabbitConfig = scRabbitConfigMapper.findByCode(lookupKey);
+        if (scRabbitConfig != null) {
+            generalRabbitSyncService.syncWaterMeterData(scRabbitConfig);
+        }else {
+            throw new ServiceException(-900,"lookupKey不存在");
+        }
+
+        return new AjaxMessage(ResultStatus.OK);
+    }
+
+    @GetMapping("addConnectionFactory")
+    @ApiOperation(value = "addConnectionFactory")
+    public AjaxMessage addConnectionFactory(@ApiParam(value = "lookupKey", required = true) @RequestParam String lookupKey) {
+        ScRabbitConfig config = scRabbitConfigMapper.findByCode(lookupKey);
+        if (config != null) {
+            DynamicRoutingConnectionFactory dynamicRoutingConnectionFactory = DynamicRoutingConnectionFactory.getInstance();
+            CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
+            connectionFactory.setHost(config.getHost());
+            connectionFactory.setPort(config.getPort());
+            connectionFactory.setUsername(config.getUsername());
+            connectionFactory.setPassword(config.getPassword());
+            connectionFactory.setVirtualHost(config.getVirtualHost());
+            dynamicRoutingConnectionFactory.addTargetConnectionFactory(lookupKey, connectionFactory);
+        }else {
+            throw new ServiceException(-900,"lookupKey不存在");
+        }
+
+        return new AjaxMessage(ResultStatus.OK);
+    }
+
+    @GetMapping("removeConnectionFactory")
+    @ApiOperation(value = "removeConnectionFactory")
+    public AjaxMessage removeConnectionFactory(@ApiParam(value = "lookupKey", required = true) @RequestParam String lookupKey) {
+        DynamicRoutingConnectionFactory dynamicRoutingConnectionFactory = DynamicRoutingConnectionFactory.getInstance();
+        dynamicRoutingConnectionFactory.removeTargetConnectionFactory(lookupKey);
+        return new AjaxMessage(ResultStatus.OK);
+    }
+}

smart-city-platform/src/main/java/com/bz/smart_city/dao/ScRabbitConfigMapper.java

@@ -0,0 +1,18 @@
+package com.bz.smart_city.dao;
+
+import com.bz.smart_city.entity.ScRabbitConfig;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+@Mapper
+public interface ScRabbitConfigMapper {
+    int insertSelective(ScRabbitConfig record);
+
+    int updateByPrimaryKeySelective(ScRabbitConfig record);
+
+    List<ScRabbitConfig> findByType(@Param("type") Integer type);
+
+    ScRabbitConfig findByCode(@Param("code") String code);
+}

smart-city-platform/src/main/java/com/bz/smart_city/entity/ScRabbitConfig.java

@@ -0,0 +1,67 @@
+package com.bz.smart_city.entity;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import java.time.LocalDateTime;
+import lombok.Data;
+
+@ApiModel(value="com-bz-smart_city-entity-ScRabbitConfig")
+@Data
+public class ScRabbitConfig {
+    @ApiModelProperty(value="")
+    private Integer id;
+
+    /**
+    * 名称
+    */
+    @ApiModelProperty(value="名称")
+    private String name;
+
+    /**
+    * 唯一编号
+    */
+    @ApiModelProperty(value="唯一编号")
+    private String code;
+
+    @ApiModelProperty(value="")
+    private Integer type;
+
+    @ApiModelProperty(value="")
+    private Integer customerId;
+
+    @ApiModelProperty(value="")
+    private String host;
+
+    @ApiModelProperty(value="")
+    private Integer port;
+
+    @ApiModelProperty(value="")
+    private String username;
+
+    @ApiModelProperty(value="")
+    private String password;
+
+    @ApiModelProperty(value="")
+    private String virtualHost;
+
+    @ApiModelProperty(value="")
+    private String exchange;
+
+    @ApiModelProperty(value="")
+    private String routingKey;
+
+    @ApiModelProperty(value="")
+    private Integer status;
+
+    @ApiModelProperty(value="")
+    private String createBy;
+
+    @ApiModelProperty(value="")
+    private LocalDateTime dateCreate;
+
+    @ApiModelProperty(value="")
+    private String updateBy;
+
+    @ApiModelProperty(value="")
+    private LocalDateTime dateUpdate;
+}

smart-city-platform/src/main/java/com/bz/smart_city/excel/download/GeneralDownloadExcelTemplate.java

@@ -40,7 +40,7 @@ public class GeneralDownloadExcelTemplate extends AbstractDownloadExcelTemplate
     protected void setParam() {
         LoginUser loginUser = UserUtil.getCurrentUser();
         filePath = getFilePath();
-        templatePath = "eexcel/deviceTemplateV1.xlsx";
+        templatePath = "excel/deviceTemplateV1.xlsx";
 
         Channel channel = channelMapper.findById(sysId);
         downloadName = channel.getChannelName()+"批量导入模板";

smart-city-platform/src/main/java/com/bz/smart_city/quartz/job/GeneralRabbitSyncJob.java

@@ -0,0 +1,50 @@
+package com.bz.smart_city.quartz.job;
+
+import com.alibaba.fastjson.JSON;
+import com.bz.smart_city.dao.ScRabbitConfigMapper;
+import com.bz.smart_city.entity.ScRabbitConfig;
+import com.bz.smart_city.service.SyncWaterMeterDataService;
+import com.bz.smart_city.service.impl.GeneralRabbitSyncService;
+import lombok.extern.slf4j.Slf4j;
+import org.quartz.Job;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * <p>通rabbit mq数据同步</p>
+ *
+ * @Author linqingwei
+ * @Date 2020/11/17 15:45
+ * @Version 1.0
+ */
+@Slf4j
+@Component
+public class GeneralRabbitSyncJob implements Job, Serializable {
+
+    @Resource
+    private ScRabbitConfigMapper scRabbitConfigMapper;
+    @Autowired
+    GeneralRabbitSyncService generalRabbitSyncService;
+
+    @Override
+    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
+        log.info("begin GeneralRabbitSyncJob !");
+        List<ScRabbitConfig> configs = scRabbitConfigMapper.findByType(1);
+        log.info("ScRabbitConfig configs = {}", JSON.toJSONString(configs));
+        if (configs != null && configs.size() > 0) {
+            for (ScRabbitConfig config : configs) {
+                if (config.getCustomerId() != null) {
+                    generalRabbitSyncService.syncWaterMeterData(config);
+                }
+            }
+        }
+
+        log.info("end GeneralRabbitSyncJob !");
+    }
+}

smart-city-platform/src/main/java/com/bz/smart_city/rabbitmq/DynamicRoutingConnectionFactory.java

@@ -0,0 +1,42 @@
+package com.bz.smart_city.rabbitmq;
+
+import org.springframework.amqp.rabbit.connection.AbstractRoutingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.SimpleResourceHolder;
+import org.springframework.lang.Nullable;
+
+public class DynamicRoutingConnectionFactory extends AbstractRoutingConnectionFactory {
+
+    //单例句柄
+    private static DynamicRoutingConnectionFactory instance;
+    private static byte[] lock=new byte[0];
+
+    @Override
+    @Nullable
+    protected Object determineCurrentLookupKey() {
+        return SimpleResourceHolder.get(this);
+    }
+
+    /**
+     * 单例方法
+     * @return
+     */
+    public static synchronized DynamicRoutingConnectionFactory getInstance(){
+        if(instance==null){
+            synchronized (lock){
+                if(instance==null){
+                    instance=new DynamicRoutingConnectionFactory();
+                }
+            }
+        }
+        return instance;
+    }
+
+    public void addTargetConnectionFactory(Object key, ConnectionFactory connectionFactory) {
+        super.addTargetConnectionFactory(key,connectionFactory);
+    }
+
+    public ConnectionFactory removeTargetConnectionFactory(Object key) {
+        return super.removeTargetConnectionFactory(key);
+    }
+}

smart-city-platform/src/main/java/com/bz/smart_city/rabbitmq/RabbitConfig.java

@@ -0,0 +1,57 @@
+package com.bz.smart_city.rabbitmq;
+
+
+import com.bz.smart_city.dao.ScRabbitConfigMapper;
+import com.bz.smart_city.entity.ScRabbitConfig;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.Resource;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+
+@Configuration
+public class RabbitConfig {
+
+    @Resource
+    private ScRabbitConfigMapper scRabbitConfigMapper;
+
+
+
+    private ConnectionFactory createConnectionFactory(String host, Integer port, String username, String password,String virtualHost) {
+        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
+        connectionFactory.setHost(host);
+        connectionFactory.setPort(port);
+        connectionFactory.setUsername(username);
+        connectionFactory.setPassword(password);
+        connectionFactory.setVirtualHost(virtualHost);
+        return connectionFactory;
+    }
+
+    @Bean
+    public ConnectionFactory routeConnectionFactory(){
+
+        List<ScRabbitConfig> list = scRabbitConfigMapper.findByType(1);
+        if (list != null && list.size() > 0) {
+            DynamicRoutingConnectionFactory dynamicRoutingConnectionFactory = DynamicRoutingConnectionFactory.getInstance();
+            Map<Object, ConnectionFactory> map = new HashMap<>();
+            for (ScRabbitConfig config : list) {
+                if(config.getCode() != null){
+                    ConnectionFactory connectionFactory = createConnectionFactory(config.getHost(),config.getPort(),config.getUsername(),config.getPassword(),config.getVirtualHost());
+                    map.put(config.getCode(),connectionFactory);
+                }
+            }
+            dynamicRoutingConnectionFactory.setTargetConnectionFactories(map);
+            return dynamicRoutingConnectionFactory;
+        }
+        return null;
+
+
+    }
+}
+

smart-city-platform/src/main/java/com/bz/smart_city/service/impl/DeviceServiceImpl.java

@@ -538,11 +538,13 @@ public  class DeviceServiceImpl implements DeviceService {
         //2、同步设备维度关系
         DeviceDimension deviceDimension = new DeviceDimension();
         deviceDimension.setDeviceId(device.getId());
-        deviceDimension.setProvince(String.valueOf(building.getProvince()));
-        deviceDimension.setCity(String.valueOf(building.getCity()));
-        deviceDimension.setRegion(String.valueOf(building.getRegion()));
-        deviceDimension.setCommunity(String.valueOf(building.getCommunity()));
-        deviceDimension.setBuilding(String.valueOf(building.getId()));
+        if (building != null) {
+            deviceDimension.setProvince(String.valueOf(building.getProvince()));
+            deviceDimension.setCity(String.valueOf(building.getCity()));
+            deviceDimension.setRegion(String.valueOf(building.getRegion()));
+            deviceDimension.setCommunity(String.valueOf(building.getCommunity()));
+            deviceDimension.setBuilding(String.valueOf(building.getId()));
+        }
         deviceDimension.setFloor(String.valueOf(device.getFloor()));
         deviceDimension.setDevice(device.getDeviceNo());
         deviceDimension.setCustomer(String.valueOf(device.getCustomerId()));
@@ -741,11 +743,13 @@ public  class DeviceServiceImpl implements DeviceService {
         //2、同步设备维度关系
         DeviceDimension deviceDimension = new DeviceDimension();
         deviceDimension.setDeviceId(device.getId());
-        deviceDimension.setProvince(String.valueOf(building.getProvince()));
-        deviceDimension.setCity(String.valueOf(building.getCity()));
-        deviceDimension.setRegion(String.valueOf(building.getRegion()));
-        deviceDimension.setCommunity(String.valueOf(building.getCommunity()));
-        deviceDimension.setBuilding(String.valueOf(building.getId()));
+        if (building != null) {
+            deviceDimension.setProvince(String.valueOf(building.getProvince()));
+            deviceDimension.setCity(String.valueOf(building.getCity()));
+            deviceDimension.setRegion(String.valueOf(building.getRegion()));
+            deviceDimension.setCommunity(String.valueOf(building.getCommunity()));
+            deviceDimension.setBuilding(String.valueOf(building.getId()));
+        }
         deviceDimension.setFloor(String.valueOf(device.getFloor()));
         deviceDimension.setDevice(device.getDeviceNo());
         deviceDimension.setCustomer(String.valueOf(device.getCustomerId()));
@@ -772,98 +776,7 @@ public  class DeviceServiceImpl implements DeviceService {
 
         }
 
-        /*//先删除旧的设备维度关系
-        deviceDimInfoMapper.delByDeviceId(device.getId(), loginUser.getUsername());
-
-        List<DeviceDimInfo> deviceDimInfoList = newArrayList();
-        //2.1、省
-        if (building.getProvince() != null) {
-            DeviceDimInfo deviceDimInfoProvince = new DeviceDimInfo();
-            deviceDimInfoProvince.setDeviceId(device.getId());
-            deviceDimInfoProvince.setDimensionId(1);
-            deviceDimInfoProvince.setDimensionCode("PROVINCE");
-            deviceDimInfoProvince.setDimensionValue(building.getProvince().toString());
-            deviceDimInfoProvince.setStatus(1);
-            deviceDimInfoProvince.setCreateBy(loginUser.getUsername());
-            deviceDimInfoProvince.setDateCreate(LocalDateTime.now());
-            deviceDimInfoProvince.setUpdateBy(loginUser.getUsername());
-            deviceDimInfoProvince.setDateUpdate(LocalDateTime.now());
-            deviceDimInfoList.add(deviceDimInfoProvince);
-        }
-
-        //2.2、市
-        if (building.getCity() != null) {
-            DeviceDimInfo deviceDimInfoCity = new DeviceDimInfo();
-            deviceDimInfoCity.setDeviceId(device.getId());
-            deviceDimInfoCity.setDimensionId(2);
-            deviceDimInfoCity.setDimensionCode("CITY");
-            deviceDimInfoCity.setDimensionValue(building.getCity().toString());
-            deviceDimInfoCity.setStatus(1);
-            deviceDimInfoCity.setCreateBy(loginUser.getUsername());
-            deviceDimInfoCity.setDateCreate(LocalDateTime.now());
-            deviceDimInfoCity.setUpdateBy(loginUser.getUsername());
-            deviceDimInfoCity.setDateUpdate(LocalDateTime.now());
-            deviceDimInfoList.add(deviceDimInfoCity);
-        }
-
-        //2.3、区
-        if (building.getRegion() != null) {
-            DeviceDimInfo deviceDimInfoRegion = new DeviceDimInfo();
-            deviceDimInfoRegion.setDeviceId(device.getId());
-            deviceDimInfoRegion.setDimensionId(3);
-            deviceDimInfoRegion.setDimensionCode("REGION");
-            deviceDimInfoRegion.setDimensionValue(building.getRegion().toString());
-            deviceDimInfoRegion.setStatus(1);
-            deviceDimInfoRegion.setCreateBy(loginUser.getUsername());
-            deviceDimInfoRegion.setDateCreate(LocalDateTime.now());
-            deviceDimInfoRegion.setUpdateBy(loginUser.getUsername());
-            deviceDimInfoRegion.setDateUpdate(LocalDateTime.now());
-            deviceDimInfoList.add(deviceDimInfoRegion);
-        }
-
-        //2.4、小区
-        if (building.getCommunity() != null) {
-            DeviceDimInfo deviceDimInfoCommunity = new DeviceDimInfo();
-            deviceDimInfoCommunity.setDeviceId(device.getId());
-            deviceDimInfoCommunity.setDimensionId(4);
-            deviceDimInfoCommunity.setDimensionCode("COMMUNITY");
-            deviceDimInfoCommunity.setDimensionValue(building.getCommunity().toString());
-            deviceDimInfoCommunity.setStatus(1);
-            deviceDimInfoCommunity.setCreateBy(loginUser.getUsername());
-            deviceDimInfoCommunity.setDateCreate(LocalDateTime.now());
-            deviceDimInfoCommunity.setUpdateBy(loginUser.getUsername());
-            deviceDimInfoCommunity.setDateUpdate(LocalDateTime.now());
-            deviceDimInfoList.add(deviceDimInfoCommunity);
-        }
 
-        //2.5、建筑
-        DeviceDimInfo deviceDimInfoBuilding = new DeviceDimInfo();
-        deviceDimInfoBuilding.setDeviceId(device.getId());
-        deviceDimInfoBuilding.setDimensionId(5);
-        deviceDimInfoBuilding.setDimensionCode("BUILDING");
-        deviceDimInfoBuilding.setDimensionValue(building.getId().toString());
-        deviceDimInfoBuilding.setStatus(1);
-        deviceDimInfoBuilding.setCreateBy(loginUser.getUsername());
-        deviceDimInfoBuilding.setDateCreate(LocalDateTime.now());
-        deviceDimInfoBuilding.setUpdateBy(loginUser.getUsername());
-        deviceDimInfoBuilding.setDateUpdate(LocalDateTime.now());
-        deviceDimInfoList.add(deviceDimInfoBuilding);
-
-
-        //2.6、设备
-        DeviceDimInfo deviceDimInfoDevice = new DeviceDimInfo();
-        deviceDimInfoDevice.setDeviceId(device.getId());
-        deviceDimInfoDevice.setDimensionId(8);
-        deviceDimInfoDevice.setDimensionCode("DEVICE");
-        deviceDimInfoDevice.setDimensionValue(device.getDeviceNo());
-        deviceDimInfoDevice.setStatus(1);
-        deviceDimInfoDevice.setCreateBy(loginUser.getUsername());
-        deviceDimInfoDevice.setDateCreate(LocalDateTime.now());
-        deviceDimInfoDevice.setUpdateBy(loginUser.getUsername());
-        deviceDimInfoDevice.setDateUpdate(LocalDateTime.now());
-        deviceDimInfoList.add(deviceDimInfoDevice);
-
-        deviceDimInfoMapper.insertList(deviceDimInfoList);*/
         log.info("end editDevice,result = " + result);
     }
 

smart-city-platform/src/main/java/com/bz/smart_city/service/impl/GeneralRabbitSyncService.java

@@ -0,0 +1,322 @@
+package com.bz.smart_city.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.bz.smart_city.commom.model.CommonQueryCondition;
+import com.bz.smart_city.commom.util.DateTimeUtil;
+import com.bz.smart_city.dao.DeviceMapper;
+import com.bz.smart_city.dao.MeterPushDataMapper;
+import com.bz.smart_city.dao.MeterReadRecordMapper;
+import com.bz.smart_city.dto.MessageData;
+import com.bz.smart_city.entity.Device;
+import com.bz.smart_city.entity.MeterPushData;
+import com.bz.smart_city.entity.MeterReadRecord;
+import com.bz.smart_city.entity.ScRabbitConfig;
+import com.bz.smart_city.service.SyncWaterMeterDataService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.amqp.rabbit.connection.SimpleResourceHolder;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.math.BigDecimal;
+import java.util.*;
+
+/**
+ * <p></p>
+ *
+ * @Author wilian.peng
+ * @Date 2020/3/5 18:56
+ * @Version 1.0
+ */
+@Slf4j
+@Service
+public class GeneralRabbitSyncService {
+    @Autowired
+    RabbitTemplate rabbitTemplate;
+
+    @Autowired
+    MeterReadRecordMapper meterReadRecordMapper;
+
+    @Autowired
+    DeviceMapper  deviceMapper ;
+
+    @Autowired
+    MeterPushDataMapper meterPushDataMapper ;
+
+    @Value("${spring.profiles.active}")
+    String env ;
+
+    /**
+     */
+    public void syncWaterMeterData(ScRabbitConfig config){
+        // 1,查询昨日抄表数据
+        int yesterday = Integer.parseInt(DateTimeUtil.formatDate(DateTimeUtil.beforeNow(1), "yyyyMMdd"));
+        int today = Integer.parseInt(DateTimeUtil.formatNow("yyyyMMdd"));
+        CommonQueryCondition condition = new CommonQueryCondition();
+        condition.setCustormerId(config.getCustomerId());
+        if("prd".equals(env)){
+            condition.setStartDate(yesterday);
+            condition.setEndDate(yesterday);
+        }
+        else{
+            condition.setStartDate(20201113);
+            condition.setEndDate(20201113);
+        }
+        // 只传已验收上线的
+        condition.setChannelId(40);
+        log.info("begin query meter read records ,condition = {}",JSON.toJSONString(condition));
+        List<MeterReadRecord> meterReadRecords = meterReadRecordMapper.queryMeterReadRecordWithCondtion(condition);
+        log.info("end query meter read records ,result size is = {}",meterReadRecords.size());
+        // 2,构建消息
+        for(MeterReadRecord record : meterReadRecords){
+            try{
+                Long deviceId = record.getDeviceId();
+                Device device = deviceMapper.findByDeviceId(deviceId);
+                // 设备为故障时不进行数据同步
+                if(device.getDeviceStatus() == 2){
+                    continue;
+                }
+                else {
+                    MeterPushData mpd = new MeterPushData();
+                    mpd.setCustomerId(device.getCustomerId());
+                    mpd.setDeviceId(device.getId());
+                    mpd.setDeviceNo(device.getDeviceNo());
+                    mpd.setDeviceStatus(device.getDeviceStatus());
+                    mpd.setMeterNo(device.getWaterMeterNo());
+                    mpd.setSyncDataDate(new Date());
+                    mpd.setSyncDate(today);
+                    // 非故障设备时判断是否有抄表数据
+                    if (record.getReadStatus().equals("1")) {
+                        // 无抄表数据,则继续取上日的推送数据进行推送,若上一日无推送数据则取最近一次的抄表数据进行推送
+                        MeterReadRecord meterLastReadRecord = meterReadRecordMapper.findMeterLastReadRecord(deviceId);
+                        if (meterLastReadRecord == null) {
+                            continue;
+                        }
+                        mpd.setRealData(meterLastReadRecord.getReadData());
+                        mpd.setRealDataDate(meterLastReadRecord.getReadTime());
+                        mpd.setSyncData(meterLastReadRecord.getReadData());
+                        mpd.setGap(new BigDecimal(0));
+                    } else {
+                        // 有抄表数据,则判断与上日推送数据是否存在-3内的波动差
+                        MeterPushData pushData = meterPushDataMapper.selectByPrimaryKey(device.getId(), yesterday);
+                        // 若不存在推送数据,则直接推送今日抄表数据
+                        if (pushData == null) {
+                            mpd.setRealData(record.getReadData());
+                            mpd.setRealDataDate(record.getReadTime());
+                            mpd.setSyncData(record.getReadData());
+                            mpd.setGap(new BigDecimal(0));
+                        } else {
+                            // 若存在推送数据则比较今日数据真实上报数据进行比较,若比上日推送数据大或者波动差>=-3则直接进行推送,
+                            // 若波动差<-3则推送上日的推送数据
+                            mpd.setRealData(record.getReadData());
+                            mpd.setRealDataDate(record.getReadTime());
+                            BigDecimal syncData = new BigDecimal(pushData.getSyncData());
+                            BigDecimal readData = new BigDecimal(record.getReadData());
+                            BigDecimal gap = readData.subtract(syncData);
+                            // 今日上报数据大于等于昨日推送数据,则推送今日上报数据
+                            if (gap.compareTo(new BigDecimal(0)) > -1) {
+                                mpd.setSyncData(record.getReadData());
+                                mpd.setGap(gap);
+                            } else {
+                                // 波动差>=-3则直接进行推送
+                                if (gap.abs().compareTo(new BigDecimal(3)) < 1) {
+                                    mpd.setSyncData(record.getReadData());
+                                    mpd.setGap(gap);
+                                }
+                                // 若波动差<-3则推送上日的推送数据
+                                else {
+                                    mpd.setSyncData(pushData.getSyncData());
+                                    mpd.setGap(gap);
+                                }
+                            }
+                        }
+                    }
+                    // 将推送记录进行存储,并将数据进行推送
+                    meterPushDataMapper.insert(mpd);
+                    Map<String, Object> message = buildSendMessage(mpd);
+                    if (message != null) {
+                        // 3,发送消息
+                        sendMessage(config,JSON.toJSONString(message));
+                    }
+                }
+            }catch (Exception e){
+                log.error("Synchronize Data Error ! ",e);
+            }
+        }
+
+    }
+
+
+    public void mockSyncWaterMeterData(ScRabbitConfig config,Long deviceId) {
+        // 1,查询昨日抄表数据
+        int yesterday = Integer.parseInt(DateTimeUtil.formatDate(DateTimeUtil.beforeNow(1), "yyyyMMdd"));
+        int today = Integer.parseInt(DateTimeUtil.formatNow("yyyyMMdd"));
+        MeterReadRecord record = meterReadRecordMapper.getLastDayData(deviceId, yesterday);
+        // 2,构建消息
+        if(record != null){
+            try{
+                Device device = deviceMapper.findByDeviceId(deviceId);
+                // 设备为故障时不进行数据同步
+                if(device.getDeviceStatus() == 2){
+                    return;
+                }
+                else {
+                    MeterPushData mpd = new MeterPushData();
+                    mpd.setCustomerId(device.getCustomerId());
+                    mpd.setDeviceId(device.getId());
+                    mpd.setDeviceNo(device.getDeviceNo());
+                    mpd.setDeviceStatus(device.getDeviceStatus());
+                    mpd.setMeterNo(device.getWaterMeterNo());
+                    mpd.setSyncDataDate(new Date());
+                    mpd.setSyncDate(today);
+                    // 非故障设备时判断是否有抄表数据
+                    if ("1".equals(record.getReadStatus())) {
+                        // 无抄表数据,则继续取上日的推送数据进行推送,若上一日无推送数据则取最近一次的抄表数据进行推送
+                        MeterPushData pushData = meterPushDataMapper.selectByPrimaryKey(device.getId(), yesterday);
+                        if(pushData == null ){
+                            MeterReadRecord meterLastReadRecord = meterReadRecordMapper.findMeterLastReadRecord(deviceId);
+                            mpd.setRealData(meterLastReadRecord.getReadData());
+                            mpd.setRealDataDate(meterLastReadRecord.getReadTime());
+                            mpd.setSyncData(meterLastReadRecord.getReadData());
+                            mpd.setGap(new BigDecimal(0));
+                        }
+                        else{
+                            mpd.setSyncData(pushData.getSyncData());
+                            mpd.setGap(new BigDecimal(0));
+                        }
+                    } else {
+                        // 有抄表数据,则判断与上日推送数据是否存在-3内的波动差
+                        MeterPushData pushData = meterPushDataMapper.selectByPrimaryKey(device.getId(), yesterday);
+                        // 若不存在推送数据,则直接推送今日抄表数据
+                        if (pushData == null) {
+                            mpd.setRealData(record.getReadData());
+                            mpd.setRealDataDate(record.getReadTime());
+                            mpd.setSyncData(record.getReadData());
+                            mpd.setGap(new BigDecimal(0));
+                        } else {
+                            // 若存在推送数据则比较今日数据真实上报数据进行比较,若比上日推送数据大或者波动差>=-3则直接进行推送,
+                            // 若波动差<-3则推送上日的推送数据
+                            mpd.setRealData(record.getReadData());
+                            mpd.setRealDataDate(record.getReadTime());
+                            BigDecimal syncData = new BigDecimal(pushData.getSyncData());
+                            BigDecimal readData = new BigDecimal(record.getReadData());
+                            BigDecimal gap = readData.subtract(syncData);
+                            // 今日上报数据大于等于昨日推送数据,则推送今日上报数据
+                            if (gap.compareTo(new BigDecimal(0)) > -1) {
+                                mpd.setSyncData(record.getReadData());
+                                mpd.setGap(gap);
+                            } else {
+                                // 若波动差>-3,即波动差的绝对值小于3 ,则推送上日的推送数据
+                                if (gap.abs().compareTo(new BigDecimal(3)) < 1) {
+                                    mpd.setSyncData(pushData.getSyncData());
+                                    mpd.setGap(gap);
+                                }
+                                //  波动差<=-3,即波动差的绝对值>3,则直接进行推送
+                                else {
+                                    mpd.setSyncData(record.getReadData());
+                                    mpd.setGap(gap);
+                                }
+                            }
+                        }
+                    }
+                    // 将推送记录进行存储,并将数据进行推送
+                    meterPushDataMapper.insert(mpd);
+                    Map<String, Object> message = buildSendMessage(mpd);
+                    if (message != null) {
+                        // 3,发送消息
+                        sendMessage(config,JSON.toJSONString(message));
+                    }
+                }
+            }catch (Exception e){
+                log.error("Synchronize Data Error ! ",e);
+            }
+        }
+    }
+
+
+    protected Map<String,Object> buildSendMessage(MeterPushData pushData){
+        HashMap<String, Object> msg = new HashMap<>();
+        msg.put("type","nbxb");
+        // 设备编号
+        String deviceId= pushData.getMeterNo();
+        if(StringUtils.isNotBlank(deviceId)) {
+            // 设备编码补齐14位后再添加30
+            msg.put("deviceId","30"+deviceId) ;
+            List<MessageData> datas = new ArrayList<>() ;
+            MessageData data = new MessageData();
+            data.setS("v");
+            data.setT(System.currentTimeMillis());
+            data.setV(Double.parseDouble(pushData.getSyncData()));
+            datas.add(data);
+            msg.put("real",datas);
+        }
+        else{
+            msg = null ;
+        }
+        return msg ;
+    }
+    /**
+     *  {"real":[{"s":"v","t":1578534831000,"v":39.0}],"type":"nbxb","deviceId":"30000161034460"}
+     * @param record
+     * @return
+     */
+    @Deprecated
+    protected Map<String,Object> buildSendMessage(MeterReadRecord record){
+        HashMap<String, Object> msg = new HashMap<>();
+        msg.put("type","nbxb");
+
+        // 设备编号
+        String deviceId= record.getMeterNo();
+        if(StringUtils.isNotBlank(deviceId)) {
+            // 设备编码补齐14位后再添加30
+            msg.put("deviceId","30"+deviceId) ;
+            List<MessageData> datas = new ArrayList<>() ;
+            MessageData data = new MessageData();
+            data.setS("v");
+            //data.setT(record.getReadTime().getTime());
+            data.setT(System.currentTimeMillis());
+            data.setV(Double.parseDouble(record.getReadData()));
+            datas.add(data);
+            msg.put("real",datas);
+        }
+        else{
+            msg = null ;
+        }
+        return msg ;
+    }
+
+
+    protected void sendMessage(ScRabbitConfig config, String message){
+        log.info("begin send message ,config = {},message = {}", config, message);
+        //设置当前线程lookupKey，内部由ThreadLocal实现
+        SimpleResourceHolder.bind(rabbitTemplate.getConnectionFactory(), config.getCode());
+        //业务操作会根据线程中的lookupKey从routeConnectionFactory的targetConnectionFactories中选择对应的connectionFactory
+        rabbitTemplate.convertAndSend(config.getExchange(),config.getRoutingKey(), message);
+        //操作完以后记得解绑。不影响线程的后序其他工厂操作
+        SimpleResourceHolder.unbind(rabbitTemplate.getConnectionFactory());
+        log.info("end send message ,config = {},message = {}", config, message);
+    }
+
+    /**
+     * 字符串高位补零
+     * @param str
+     * @param strLength
+     * @return
+     */
+    protected static String addZeroForString(String str, int strLength) {
+        int strLen = str.length();
+        if (strLen < strLength) {
+            while (strLen < strLength) {
+                StringBuffer sb = new StringBuffer();
+                // 左补0
+                sb.append("0").append(str);
+                str = sb.toString();
+                strLen = str.length();
+            }
+        }
+        return str;
+    }
+
+}

smart-city-platform/src/main/resources/application-dev.properties

@@ -152,12 +152,12 @@ crm.login.password=111111
 
 
 #rabbitmq
-spring.rabbitmq.host=222.86.88.126
-spring.rabbitmq.port=5627
-spring.rabbitmq.username=hdkj
-spring.rabbitmq.password=hdkj123
-spring.rabbitmq.virtual-host=/
-spring.rabbitmq.zy.topic=hello
+#spring.rabbitmq.host=222.86.88.126
+#spring.rabbitmq.port=5627
+#spring.rabbitmq.username=hdkj
+#spring.rabbitmq.password=hdkj123
+#spring.rabbitmq.virtual-host=/
+#spring.rabbitmq.zy.topic=hello
 
 #UDIP平台集成元管理url
 udip.unit.url=http://114.135.61.188:48322

smart-city-platform/src/main/resources/application-prd.properties

@@ -106,7 +106,7 @@ spring.quartz.properties.org.quartz.jobStore.clusterCheckinInterval=10000
 spring.quartz.properties.org.quartz.jobStore.useProperties=false
 spring.quartz.properties.org.quartz.jobStore.misfireThreshold = 1000
 spring.quartz.properties.org.quartz.threadPool.class=org.quartz.simpl.SimpleThreadPool
-spring.quartz.properties.org.quartz.threadPool.threadCount=15
+spring.quartz.properties.org.quartz.threadPool.threadCount=20
 spring.quartz.properties.org.quartz.threadPool.threadPriority=5
 spring.quartz.properties.org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread=true
 
@@ -156,12 +156,12 @@ crm.api.ur=http://183.62.175.139:5889/hxcrm/api/Crm.SnMgt.getMeterInformationLis
 crm.login.username=CRM
 crm.login.password=CRM2019
 #rabbitmq
-spring.rabbitmq.host=222.86.88.126
-spring.rabbitmq.port=5627
-spring.rabbitmq.username=hdkj
-spring.rabbitmq.password=hdkj123
-spring.rabbitmq.virtual-host=/
-spring.rabbitmq.zy.topic=zyqueue
+#spring.rabbitmq.host=222.86.88.126
+#spring.rabbitmq.port=5627
+#spring.rabbitmq.username=hdkj
+#spring.rabbitmq.password=hdkj123
+#spring.rabbitmq.virtual-host=/
+#spring.rabbitmq.zy.topic=zyqueue
 
 #UDIP平台集成元管理url
 udip.unit.url=http://47.112.15.78:8322

smart-city-platform/src/main/resources/application-sit.properties

@@ -155,12 +155,12 @@ crm.api.ur=http://meetrend.f3322.net:8089/crm/api/Crm.SnMgt.getMeterInformationL
 crm.login.username=揭健
 crm.login.password=111111
 #rabbitmq
-spring.rabbitmq.host=222.86.88.126
-spring.rabbitmq.port=5627
-spring.rabbitmq.username=hdkj
-spring.rabbitmq.password=hdkj123
-spring.rabbitmq.virtual-host=/
-spring.rabbitmq.zy.topic=hello
+#spring.rabbitmq.host=222.86.88.126
+#spring.rabbitmq.port=5627
+#spring.rabbitmq.username=hdkj
+#spring.rabbitmq.password=hdkj123
+#spring.rabbitmq.virtual-host=/
+#spring.rabbitmq.zy.topic=hello
 
 #UDIP平台集成元管理url
 udip.unit.url=http://114.135.61.188:48322

smart-city-platform/src/main/resources/application.properties

@@ -1,2 +1,2 @@
-#����������dev  ���Ի�����sit  ���ϻ�����prd  ��ʾ������uat
-spring.profiles.active=dev
+#开发环境：dev  测试环境：sit  线上环境：prd  演示环境：uat
+spring.profiles.active=prd

smart-city-platform/src/main/resources/mapper/ScRabbitConfigMapper.xml

@@ -0,0 +1,204 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.bz.smart_city.dao.ScRabbitConfigMapper">
+  <resultMap id="BaseResultMap" type="com.bz.smart_city.entity.ScRabbitConfig">
+    <!--@mbg.generated-->
+    <!--@Table sc_rabbit_config-->
+    <id column="id" property="id" />
+    <result column="name" property="name" />
+    <result column="code" property="code" />
+    <result column="type" property="type" />
+    <result column="customer_id" property="customerId" />
+    <result column="host" property="host" />
+    <result column="port" property="port" />
+    <result column="username" property="username" />
+    <result column="password" property="password" />
+    <result column="virtual_host" property="virtualHost" />
+    <result column="exchange" property="exchange" />
+    <result column="routing_key" property="routingKey" />
+    <result column="status" property="status" />
+    <result column="create_by" property="createBy" />
+    <result column="date_create" property="dateCreate" />
+    <result column="update_by" property="updateBy" />
+    <result column="date_update" property="dateUpdate" />
+  </resultMap>
+  <sql id="Base_Column_List">
+    <!--@mbg.generated-->
+    id, `name`, code, `type`, customer_id, `host`, port, username, `password`, virtual_host, 
+    exchange, routing_key, `status`, create_by, date_create, update_by, date_update
+  </sql>
+  <insert id="insertSelective" parameterType="com.bz.smart_city.entity.ScRabbitConfig">
+    <!--@mbg.generated-->
+    insert into sc_rabbit_config
+    <trim prefix="(" suffix=")" suffixOverrides=",">
+      <if test="id != null">
+        id,
+      </if>
+      <if test="name != null">
+        `name`,
+      </if>
+      <if test="code != null">
+        code,
+      </if>
+      <if test="type != null">
+        `type`,
+      </if>
+      <if test="customerId != null">
+        customer_id,
+      </if>
+      <if test="host != null">
+        `host`,
+      </if>
+      <if test="port != null">
+        port,
+      </if>
+      <if test="username != null">
+        username,
+      </if>
+      <if test="password != null">
+        `password`,
+      </if>
+      <if test="virtualHost != null">
+        virtual_host,
+      </if>
+      <if test="exchange != null">
+        exchange,
+      </if>
+      <if test="routingKey != null">
+        routing_key,
+      </if>
+      <if test="status != null">
+        `status`,
+      </if>
+      <if test="createBy != null">
+        create_by,
+      </if>
+      <if test="dateCreate != null">
+        date_create,
+      </if>
+      <if test="updateBy != null">
+        update_by,
+      </if>
+      <if test="dateUpdate != null">
+        date_update,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides=",">
+      <if test="id != null">
+        #{id},
+      </if>
+      <if test="name != null">
+        #{name},
+      </if>
+      <if test="code != null">
+        #{code},
+      </if>
+      <if test="type != null">
+        #{type},
+      </if>
+      <if test="customerId != null">
+        #{customerId},
+      </if>
+      <if test="host != null">
+        #{host},
+      </if>
+      <if test="port != null">
+        #{port},
+      </if>
+      <if test="username != null">
+        #{username},
+      </if>
+      <if test="password != null">
+        #{password},
+      </if>
+      <if test="virtualHost != null">
+        #{virtualHost},
+      </if>
+      <if test="exchange != null">
+        #{exchange},
+      </if>
+      <if test="routingKey != null">
+        #{routingKey},
+      </if>
+      <if test="status != null">
+        #{status},
+      </if>
+      <if test="createBy != null">
+        #{createBy},
+      </if>
+      <if test="dateCreate != null">
+        #{dateCreate},
+      </if>
+      <if test="updateBy != null">
+        #{updateBy},
+      </if>
+      <if test="dateUpdate != null">
+        #{dateUpdate},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.bz.smart_city.entity.ScRabbitConfig">
+    <!--@mbg.generated-->
+    update sc_rabbit_config
+    <set>
+      <if test="name != null">
+        `name` = #{name},
+      </if>
+      <if test="code != null">
+        code = #{code},
+      </if>
+      <if test="type != null">
+        `type` = #{type},
+      </if>
+      <if test="customerId != null">
+        customer_id = #{customerId},
+      </if>
+      <if test="host != null">
+        `host` = #{host},
+      </if>
+      <if test="port != null">
+        port = #{port},
+      </if>
+      <if test="username != null">
+        username = #{username},
+      </if>
+      <if test="password != null">
+        `password` = #{password},
+      </if>
+      <if test="virtualHost != null">
+        virtual_host = #{virtualHost},
+      </if>
+      <if test="exchange != null">
+        exchange = #{exchange},
+      </if>
+      <if test="routingKey != null">
+        routing_key = #{routingKey},
+      </if>
+      <if test="status != null">
+        `status` = #{status},
+      </if>
+      <if test="createBy != null">
+        create_by = #{createBy},
+      </if>
+      <if test="dateCreate != null">
+        date_create = #{dateCreate},
+      </if>
+      <if test="updateBy != null">
+        update_by = #{updateBy},
+      </if>
+      <if test="dateUpdate != null">
+        date_update = #{dateUpdate},
+      </if>
+    </set>
+    where id = #{id}
+  </update>
+
+  <select id="findByType" resultMap="BaseResultMap">
+    select <include refid="Base_Column_List"/> from sc_rabbit_config where status = 1
+    <if test="type != null"> and type = #{type}</if>
+  </select>
+
+  <select id="findByCode" resultMap="BaseResultMap">
+    select <include refid="Base_Column_List"/> from sc_rabbit_config where status = 1 and code = #{code}
+  </select>
+</mapper>