新增mock工程

ebc-mock-service/.gitignore

@@ -0,0 +1,37 @@
### Java template
# Compiled class file
*.class
# Log file
*.log
# BlueJ files
*.ctxt
# Mobile Tools for Java (J2ME)
.mtj.tmp/
# Package Files #
*.jar
*.war
*.ear
*.zip
*.tar.gz
*.rar
#eclipse
/.settings
/.classpath
/.project
# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
hs_err_pid*
*.iml
.idea
target/
logs/
.DS_Store

ebc-mock-service/pom.xml

@@ -0,0 +1,189 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>components</artifactId>
        <groupId>com.ai.bss</groupId>
        <version>2.1.5-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>ebc-mock-service</artifactId>
    <name>ebc-mock-service</name>
    <dependencies>
        <dependency>
            <groupId>com.ai.abc</groupId>
            <artifactId>persistence-mysql8</artifactId>
            <version>2.1.5-SNAPSHOT</version>
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-security</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>com.ai.bss</groupId>
            <artifactId>components-common</artifactId>
            <version>2.1.5-SNAPSHOT</version>
            <scope>compile</scope>
        </dependency>
        <dependency>
            <groupId>com.ai.bss</groupId>
            <artifactId>infrastructure</artifactId>
            <version>2.1.5-SNAPSHOT</version>
            <scope>compile</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.2.9.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <version>${spring-boot.version}</version>
            <type>pom</type>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.ai.ipu</groupId>
            <artifactId>ipu-basic</artifactId>
            <version>3.1-SNAPSHOT</version>
            <exclusions>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>log4j</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.logging</groupId>
                    <artifactId>log4j</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>com.ai.ipu</groupId>
            <artifactId>ipu-common</artifactId>
            <version>3.1-SNAPSHOT</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-slf4j-impl</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>dom4j</groupId>
                    <artifactId>dom4j</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>
    <build>
        <resources>
            <resource>
                <directory>src/main/resources</directory>
            </resource>
        </resources>
        <plugins>
            <!-- The configuration of maven-jar-plugin -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <version>2.4</version>
                <!-- The configuration of the plugin -->
                <configuration>
                    <!-- Configuration of the archiver -->
                    <archive>
                        <!--生成的jar中，不要包含pom.xml和pom.properties这两个文件-->
                        <addMavenDescriptor>false</addMavenDescriptor>
                        <!-- Manifest specific configuration -->
                        <manifest>
                            <!--是否要把第三方jar放到manifest的classpath中-->
                            <addClasspath>true</addClasspath>
                            <!--生成的manifest中classpath的前缀，因为要把第三方jar放到lib目录下，所以classpath的前缀是lib/-->
                            <classpathPrefix>lib/</classpathPrefix>
                            <!--应用的main class-->
                            <mainClass>com.ai.bss.mock.MockApp</mainClass>
                        </manifest>
                        <manifestEntries>
                            <Class-Path>./config/</Class-Path>
                        </manifestEntries>
                    </archive>
                    <!--过滤掉不希望包含在jar中的文件-->
                    <excludes>
                        <exclude>**/*.properties</exclude>
                        <exclude>**/*.xml</exclude>
                    </excludes>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-dependency-plugin</artifactId>
                <version>2.10</version>
                <executions>
                    <execution>
                        <id>copy-dependencies</id>
                        <phase>package</phase>
                        <goals>
                            <goal>copy-dependencies</goal>
                        </goals>
                        <configuration>
                            <outputDirectory>${project.build.directory}/lib</outputDirectory>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <!-- The configuration of maven-assembly-plugin -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>2.4</version>
                <!-- The configuration of the plugin -->
                <configuration>
                    <!-- Specifies the configuration file of the assembly plugin -->
                    <descriptors>
                        <descriptor>src/main/assembly/package.xml</descriptor>
                    </descriptors>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <configuration>
                    <skip>true</skip>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

ebc-mock-service/src/main/assembly/package.xml

@@ -0,0 +1,59 @@
<assembly>
    <id>bin</id>
    <!-- 最终打包成一个用于发布的zip文件 -->
    <formats>
        <format>zip</format>
    </formats>
    <!-- Adds dependencies to zip package under lib directory -->
    <dependencySets>
        <dependencySet>
            <!-- 不使用项目的artifact，第三方jar不要解压，打包进zip文件的lib目录 -->
            <useProjectArtifact>false</useProjectArtifact>
            <outputDirectory>lib</outputDirectory>
            <unpack>false</unpack>
            <scope>runtime</scope>
        </dependencySet>
    </dependencySets>
    <fileSets>
        <!-- 把项目相关的说明文件，打包进zip文件的根目录 -->
        <fileSet>
            <directory>${project.basedir}</directory>
            <outputDirectory>/</outputDirectory>
        <includes>
            <include>README*</include>
            <include>LICENSE*</include>
            <include>NOTICE*</include>
        </includes>
    </fileSet>
        <!-- 把项目的配置文件，打包进zip文件的config目录 -->
        <fileSet>
            <directory>src/main/resources</directory>
            <outputDirectory>/config</outputDirectory>
            <includes>
                <include>**/*.properties</include>
                <include>**/*.xml</include>
            </includes>
        </fileSet>
        <!-- 把项目的脚本文件目录（ src/main/scripts ）中的启动脚本文件，打包进zip文件的跟目录 -->
        <fileSet>
            <directory>${project.build.scriptSourceDirectory}</directory>
            <outputDirectory></outputDirectory>
            <includes>
                <include>startup.*</include>
            </includes>
        </fileSet>
        <!-- 把项目自己编译出来的jar文件，打包进zip文件的根目录 -->
        <fileSet>
            <directory>${project.build.directory}</directory>
            <outputDirectory></outputDirectory>
            <includes>
                <include>*.jar</include>
            </includes>
        </fileSet>
    </fileSets>
</assembly>

ebc-mock-service/src/main/java/com/ai/bss/mock/MockApp.java

@@ -0,0 +1,33 @@
package com.ai.bss.mock;
import com.ai.bss.mock.utils.tcp.IpuTcpLongConnectClient;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.domain.EntityScan;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
import org.springframework.scheduling.annotation.EnableAsync;
/**
 * @author zhangfeng
 */
@EnableJpaRepositories(basePackages = "com.ai.bss")
@EntityScan(basePackages = {"com.ai.bss","com.ai.abc"})
@ComponentScan(basePackages = {"com.ai.bss","com.ai.abc","com.ai.ipu"})
@EnableAsync
@SpringBootApplication
public class MockApp {
    private static final String HOST = "47.105.130.83";
    private static final int PORT = 8042;
    public static void main(String[] args) throws Exception {
//        IpuTcpLongConnectClient.getInstance(HOST, PORT).test();
        SpringApplication.run(MockApp.class, args);
    }
}

ebc-mock-service/src/main/java/com/ai/bss/mock/constants/MockConstant.java

@@ -0,0 +1,11 @@
package com.ai.bss.mock.constants;
import java.util.HashMap;
import java.util.Map;
public class MockConstant {
    public static Map MACK_STATUS = new HashMap();
    public static final String SPLIT_DASH = "-";
}

ebc-mock-service/src/main/java/com/ai/bss/mock/controller/MockController.java

@@ -0,0 +1,56 @@
package com.ai.bss.mock.controller;
import com.ai.bss.mock.constants.MockConstant;
import com.ai.bss.mock.service.interfaces.MockManageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import java.util.Map;
/**
 * mock
 */
@Controller
@RequestMapping("/mock")
public class MockController {
    @Autowired
    MockManageService mockManageService;
    Logger logger = LoggerFactory.getLogger(MockController.class);
    /**
     * 查询mock执行状态
     */
    @ResponseBody
    @RequestMapping("/getMockStatus")
    public Map getMockStatus(){
        return mockManageService.findMockStatusByScenarioId();
    }
    /**
     * 查询历史轨迹查询
     */
    @ResponseBody
    @RequestMapping("/startMackData")
    public Object startMackData(@RequestParam Long sId,@RequestParam Long f,@RequestParam(required = false) String topic0 ,@RequestParam(required = false) String topic1 ) {
        return mockManageService.startMackData(sId,f,topic0,topic1);
    }
    /**
     * 查询历史轨迹查询
     */
    @ResponseBody
    @RequestMapping("/stopMackData")
    public Object stopMackData(@RequestParam Long sId){
        return mockManageService.stopMackData(sId);
    }
}

ebc-mock-service/src/main/java/com/ai/bss/mock/model/Data.java

@@ -0,0 +1,15 @@
package com.ai.bss.mock.model;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
public class Data {
    private String unit;
    private String name;
    private String description;
    private String value;
}

ebc-mock-service/src/main/java/com/ai/bss/mock/model/DataPoint.java

@@ -0,0 +1,30 @@
package com.ai.bss.mock.model;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
public class DataPoint {
    
    private String dataPointId;
    private String dataPointValue;
    private String detailInfo;
    private String eventSpecId;
    private String eventSpecName;
    private String eventTime;
    private String eventTimeStr;
    private String eventTypeName;
    private String messageTopic;
    private String messageTypeId;
    private String resourceId;
    private String resourceName;
    private String resourceSpecId;
    private String specName;
    private String subcriberPush;
    private String subscriberDataType;
    private String terminalEventId;
    private String pushTopic;
    private long orderNo;
}

ebc-mock-service/src/main/java/com/ai/bss/mock/model/EBCData.java

@@ -0,0 +1,18 @@
package com.ai.bss.mock.model;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
public class EBCData {
    private Data orientation;
    private Data businessState;
    private Data latitude;
    private Data speed;
    private Data msgDate;
    private Data longitude;
    private Data alarmType;
}

ebc-mock-service/src/main/java/com/ai/bss/mock/model/MockScenarioData.java

@@ -0,0 +1,71 @@
package com.ai.bss.mock.model;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.hibernate.annotations.SQLDelete;
import org.hibernate.annotations.Where;
import javax.persistence.*;
import java.io.Serializable;
import java.sql.Timestamp;
@Entity
@Table( name="mock_scenario_data")
@NoArgsConstructor
@Getter
@Setter
public class MockScenarioData implements Serializable {
    @Id
    @GeneratedValue(strategy = GenerationType.AUTO)
    @Column(name="ID")
    private long id;
    //场景ID
    @Column(name="SCENARIO_ID")
    private long scenarioId;
    //场景ID
    @Column(name="DEVICE_ID")
    private String deviceId;
    //执行顺序
    @Column(name="ORDER_NO")
    private long orderNo;
    //经度
    @Column(name="LONGITUDE")
    private String longitude;
    //纬度
    @Column(name="LATITUDE")
    private String latitude;
    //速度
    @Column(name="SPEED")
    private String speed;
    //方位
    @Column(name="ORIENTATION")
    private String orientation;
    //时间
    @Column(name="TIME")
    private String time;
    //类型(0 定位；1 自动报警；2手动报警；4离线)
    @Column(name="TYPE")
    private String type;
    //类型(0 定位；1 自动报警；2手动报警；4离线)
    @Column(name="TOPIC")
    private String topic;
    //模拟数据-扩展json格式
    @Column(name="MOCK_DATA")
    private String mock_data;
}

ebc-mock-service/src/main/java/com/ai/bss/mock/repository/MockScenarioDataRepository.java

@@ -0,0 +1,25 @@
package com.ai.bss.mock.repository;
import com.ai.bss.mock.model.MockScenarioData;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Modifying;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;
import org.springframework.stereotype.Repository;
import java.io.Serializable;
import java.util.List;
@Repository
public interface MockScenarioDataRepository extends JpaRepository<MockScenarioData, Serializable> {
    List<MockScenarioData> findByScenarioIdOrderByOrderNo(Long scenarioId);
    @Modifying
    @Query(value = "select DISTINCT order_no from mock_scenario_data where scenario_id = :scenarioId order by order_no\n", nativeQuery = true)
    List<Long> findByOrderNoByScenarioId(@Param("scenarioId") Long scenarioId);
    
}

ebc-mock-service/src/main/java/com/ai/bss/mock/service/impl/MockManageServiceImpl.java

@@ -0,0 +1,70 @@
package com.ai.bss.mock.service.impl;
import com.ai.bss.mock.constants.MockConstant;
import com.ai.bss.mock.model.MockScenarioData;
import com.ai.bss.mock.repository.MockScenarioDataRepository;
import com.ai.bss.mock.service.interfaces.MockManageService;
import com.ai.bss.mock.service.interfaces.MockProcess;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@Slf4j
@Service
public class MockManageServiceImpl implements MockManageService {
    @Autowired
    private MockScenarioDataRepository mockScenarioDataRepository;
    @Autowired
    private MockProcess mockProcess;
//    @Value("${kafka.producer.servers:Empty}")
    private static String kafkaServers = "47.105.160.21:9090";
    @Override
    public Map findMockStatusByScenarioId()  {
        List<Long> result = new ArrayList(MockConstant.MACK_STATUS.keySet());
        Map  tempmap = new HashMap();
        for(Long sid:result){
            tempmap.put(sid,"场景 "+sid+" 执行中...");
        }
        return tempmap;
    }
    @Override
    public String  startMackData(Long sId,  Long frequency,String topic0,String topic1)  {
        //        验证场景是否正在执行
        if(MockConstant.MACK_STATUS.get(sId)!=null){
            return "场景" + sId + " 正在执行,请耐心等待...";
        }
        //根据场景ID从数据库中查询模拟数据
        List<MockScenarioData>  mockScenarioDataList = mockScenarioDataRepository.findByScenarioIdOrderByOrderNo(sId);
        List<Long> orderNoList = mockScenarioDataRepository.findByOrderNoByScenarioId(sId);
        if(mockScenarioDataList==null || mockScenarioDataList.size()==0){
            return "场景" + sId + " 没有配置模拟 ，请在数据库中添加数据再执行...";
        }
        //记录场景执行状态
        MockConstant.MACK_STATUS.put(sId,frequency);
        //异步执行模拟场景 通过IOT-DMP连接服务模拟
//        mockProcess.processMock(sId,mockScenarioDataList,frequency);
        //异步执行模拟场景  直接模拟最终数据发送到kafka
        mockProcess.processMockKafka(sId,mockScenarioDataList,orderNoList,frequency,topic0, topic1);
        return "场景" + sId + " 开始执行...";
    }
    @Override
    public String  stopMackData(Long sId){
        MockConstant.MACK_STATUS.remove(sId);
        return "场景" + sId + " 终止执行...";
    }
}

ebc-mock-service/src/main/java/com/ai/bss/mock/service/impl/MockProcessImpl.java

@@ -0,0 +1,234 @@
package com.ai.bss.mock.service.impl;
import com.ai.bss.mock.constants.MockConstant;
import com.ai.bss.mock.model.Data;
import com.ai.bss.mock.model.DataPoint;
import com.ai.bss.mock.model.EBCData;
import com.ai.bss.mock.model.MockScenarioData;
import com.ai.bss.mock.service.interfaces.MockProcess;
import com.ai.bss.mock.utils.KafkaProducerConfig;
import com.ai.bss.mock.utils.tcp.IpuTcpLongConnectClient;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@Slf4j
@Configuration
@Service
public class MockProcessImpl implements MockProcess {
//    EBC设备tcp连接服务地址
    private static final String HOST = "47.105.130.83";
    private static final int PORT = 8042;
//    @Value("${kafka.producer.servers:Empty}")
//    private static String kafkaServers;
    private static String kafkaServers = "47.105.160.21:9090";
    @Async
    @Override
    public void processMock(Long sId,List<MockScenarioData> mockScenarioDataList, Long frequency){
        try {
            IpuTcpLongConnectClient.getInstance(HOST, PORT).sendMockData(sId,mockScenarioDataList, frequency);
            MockConstant.MACK_STATUS.remove(sId);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    @Async
    @Override
    public void processMockKafka(Long sId,List<MockScenarioData> mockScenarioDataList,List<Long> orderNoList, Long frequency,String topic0,String topic1){
        List<DataPoint> dataPointList =  parseDataPoint(mockScenarioDataList,topic0,topic1);
        try {
            for (Long orderNo : orderNoList) {
                for (DataPoint dataPoint : dataPointList) {
                    if (dataPoint.getOrderNo() == orderNo) {
                        sendKafkaDataPoint(dataPoint);
                    }
                }
                TimeUnit.MILLISECONDS.sleep(frequency * 1000);
            }
            MockConstant.MACK_STATUS.remove(sId);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    private List<DataPoint> parseDataPoint(List<MockScenarioData> mockScenarioDataList,String topic0,String topic1){
        List<DataPoint> dataPointList = new ArrayList<>();
        DataPoint dataPoint = new DataPoint();
        EBCData ebcData = new EBCData();
        Data data = new Data();
        if(mockScenarioDataList!=null){
            for(MockScenarioData mockScenarioData:mockScenarioDataList){
                dataPoint = new DataPoint();
                ebcData = new EBCData();
                dataPoint.setResourceId(mockScenarioData.getDeviceId());
                dataPoint.setResourceName(mockScenarioData.getDeviceId());
                dataPoint.setOrderNo(mockScenarioData.getOrderNo());
                if(mockScenarioData.getType().equals("0")){
                    dataPoint.setEventSpecName("定位");
                    dataPoint.setMessageTopic("DW");
                    dataPoint.setPushTopic(topic0);
                    if(StringUtils.isEmpty(dataPoint.getPushTopic())) {
                        dataPoint.setPushTopic(mockScenarioData.getTopic() == null ? "Topic_IoT_DeviceLocation" : mockScenarioData.getTopic());
                    }
                    if(mockScenarioData.getTime()!=null){
                        data = new Data();
                        data.setName("msgDate");
                        data.setDescription("报告时间");
                        data.setValue(mockScenarioData.getTime());
                        ebcData.setMsgDate(data);
                    }
                }else{
                    dataPoint.setEventSpecName("报警类型");
                    dataPoint.setMessageTopic("BJ");
                    dataPoint.setPushTopic(topic1);
                    if(StringUtils.isEmpty(dataPoint.getPushTopic())) {
                        dataPoint.setPushTopic(mockScenarioData.getTopic() == null ? "Topic_IoT_IndividualAlarm" : mockScenarioData.getTopic());
                    }
                    data = new Data();
                    data.setName("alarmType");
                    data.setDescription("报警类型");
                    data.setValue(mockScenarioData.getType());
                    ebcData.setAlarmType(data);
                    if(mockScenarioData.getTime()!=null){
                        data = new Data();
                        data.setName("msgDate");
                        data.setDescription("报警时间");
                        data.setValue(mockScenarioData.getTime());
                        ebcData.setMsgDate(data);
                    }
                }
                if(mockScenarioData.getLatitude()!=null){
                    data = new Data();
                    data.setName("latitude");
                    data.setDescription("纬度");
                    data.setValue(mockScenarioData.getLatitude());
                    data.setUnit("度");
                    ebcData.setLatitude(data);
                }
                if(mockScenarioData.getLongitude()!=null){
                    data = new Data();
                    data.setName("longitude");
                    data.setDescription("经度");
                    data.setValue(mockScenarioData.getLongitude());
                    data.setUnit("度");
                    ebcData.setLongitude(data);
                }
                if(mockScenarioData.getSpeed()!=null){
                    data = new Data();
                    data.setName("speed");
                    data.setDescription("速度");
                    data.setValue(mockScenarioData.getSpeed());
                    data.setUnit("米/秒");
                    ebcData.setSpeed(data);
                }
                if(mockScenarioData.getOrientation()!=null){
                    data = new Data();
                    data.setName("orientation");
                    data.setDescription("方位");
                    data.setValue(mockScenarioData.getOrientation());
                    data.setUnit("度");
                    ebcData.setOrientation(data);
                }
                dataPoint.setDataPointValue(JSONObject.toJSONString(ebcData));
                dataPointList.add(dataPoint);
            }
        }
        return dataPointList;
    }
    private Map<String, KafkaTemplate> kafkaTemplateMap = new HashMap<>();
    private boolean sendKafkamsg(List<DataPoint> dataPointList,Long frequency){
//        String kafkaServers = "";
        KafkaTemplate kafkaTemplate = kafkaTemplateMap.get(kafkaServers);
        if(kafkaTemplate == null){
            //new 实例
            KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(kafkaServers);
            kafkaTemplate = kafkaProducerConfig.kafkaTemplate();
            kafkaTemplateMap.put(kafkaServers,kafkaTemplate);
        }
        try {
            for(DataPoint dataPoint:dataPointList){
                Object result = kafkaTemplate.send(dataPoint.getPushTopic(), JSONObject.toJSONString(dataPoint)).get();
//                Object result = KafkaProducerConfig.sendForKafka(dataPoint.getPushTopic(), JSONObject.toJSONString(dataPoint));
                kafkaTemplate.flush();
                log.debug("It's successful send msg to Kafka:" + kafkaServers + "服务器地址以及topic:" + dataPoint.getPushTopic() + " 内容：" + JSONObject.toJSONString(dataPoint) + " 结果" + result);
                TimeUnit.MILLISECONDS.sleep(frequency*1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            return false;
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage());
            return false;
        }
        return true;
    }
    private boolean sendKafkaDataPoint(DataPoint dataPoint){
//        String kafkaServers = "";
        KafkaTemplate kafkaTemplate = kafkaTemplateMap.get(kafkaServers);
        if(kafkaTemplate == null){
            //new 实例
            KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(kafkaServers);
            kafkaTemplate = kafkaProducerConfig.kafkaTemplate();
            kafkaTemplateMap.put(kafkaServers,kafkaTemplate);
        }
        try {
//            for(DataPoint dataPoint:dataPointList){
            Object result = kafkaTemplate.send(dataPoint.getPushTopic(), JSONObject.toJSONString(dataPoint)).get();
            kafkaTemplate.flush();
//                TimeUnit.MILLISECONDS.sleep(frequency*1000);
            log.debug("It's successful send msg to Kafka:" + kafkaServers + "服务器地址以及topic:" + dataPoint.getPushTopic() + " 内容：" + JSONObject.toJSONString(dataPoint) + " 结果" + result);
//            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            return false;
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage());
            return false;
        }
        return true;
    }
}

ebc-mock-service/src/main/java/com/ai/bss/mock/service/interfaces/MockManageService.java

@@ -0,0 +1,30 @@
package com.ai.bss.mock.service.interfaces;
import com.ai.bss.mock.model.MockScenarioData;
import java.util.List;
import java.util.Map;
public interface MockManageService {
    /**
     * 查询当前正在 的场景
     * @return
     */
    Map findMockStatusByScenarioId();
    /**
     * 开始执行模拟数据
     * @param sId
     * @param f
     * @return
     */
    String  startMackData(Long sId,  Long f,String topic0,String topic1) ;
    /**
     * 停止执行模拟数据
     * @param sId
     * @return
     */
    String  stopMackData(Long sId);
}

ebc-mock-service/src/main/java/com/ai/bss/mock/service/interfaces/MockProcess.java

@@ -0,0 +1,12 @@
package com.ai.bss.mock.service.interfaces;
import com.ai.bss.mock.model.MockScenarioData;
import java.util.List;
public interface MockProcess {
    void processMock(Long sId,List<MockScenarioData> mockScenarioDataList, Long frequency);
    void processMockKafka(Long sId,List<MockScenarioData> mockScenarioDataList,List<Long> orderNoList, Long frequency,String topic0,String topic1);
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/HexStringUtil.java

@@ -0,0 +1,48 @@
package com.ai.bss.mock.utils;
/**
 * 16进制转换工具类
 * */
public class HexStringUtil {
	/**
	 * 将16进制字符串转换为byte[]
	 */
	public static byte[] decodeHextoByte(String str) {
		if (str == null || str.trim().equals("")) {
			return new byte[0];
		}
		byte[] bytes = new byte[str.length() / 2];
		for (int i = 0; i < str.length() / 2; i++) {
			String subStr = str.substring(i * 2, i * 2 + 2);
			bytes[i] = (byte) Integer.parseInt(subStr, 16);
		}
		return bytes;
	}
	/**
	 * 将byte转换为16进制字符串
	 * @param b
	 * @return
	 */
	public static String decodeBytetoHexString(byte b) {
		return Integer.toHexString(b & 0xff);
	}
	/**
	 * 将byte数组转换为16进制字符串
	 * @param bs
	 * @return
	 */
	public static String decodeByteArraytoHexString(byte[] bs) {
		StringBuilder builder = new StringBuilder();
		for (int i = 0; i < bs.length; i++) {
			if (bs[i]<0x10 && bs[i]>=0) {
				builder.append("0");
			}
			builder.append(decodeBytetoHexString(bs[i]));
		}
		return builder.toString();
	}
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/KafkaProducerConfig.java

@@ -0,0 +1,92 @@
package com.ai.bss.mock.utils;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import java.util.HashMap;
import java.util.Map;
@Slf4j
@Configuration
@EnableKafka
@NoArgsConstructor
public class KafkaProducerConfig {
    @Value("${kafka.producer.servers:Empty}")
    private static String servers;
    @Value("${kafka.producer.retries:0}")
    private  int retries;
    @Value("${kafka.producer.batch.size:4096}")
    private  int batchSize;
    @Value("${kafka.producer.linger:1}")
    private  int linger;
    @Value("${kafka.producer.buffer.memory:40960}")
    private  int bufferMemory;
    private  int maxBlockTime = 60000;
    public KafkaProducerConfig(String inputServers){
        servers = inputServers;
        this.retries = 0;
        this.batchSize = 4096;
        this.linger = 1;
        this.bufferMemory = 40960;
        this.maxBlockTime = 6000;
    }
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG,maxBlockTime);
        return props;
    }
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<String, String>(producerFactory());
    }
//
    /**
     * 推送消息到订阅的KAFKA服务器地址以及topic
     * @param kafkaTopic  订阅的kafka topic
     * @param content  消息内容
     * @return
     */
    public static Boolean sendForKafka(String kafkaTopic,String content) {
        KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(servers);
        KafkaTemplate kafkaTemplate =  kafkaProducerConfig.kafkaTemplate();
        try {
            Object result = kafkaTemplate.send(kafkaTopic,content ).get();
            kafkaTemplate.flush();
        } catch (InterruptedException e) {
            log.error(e.getMessage());
            return false;
        } catch (Exception e) {
            log.error(e.getMessage());
            return false;
        }
        return true;
    }
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/MessageUtil.java

@@ -0,0 +1,19 @@
package com.ai.bss.mock.utils;
public class MessageUtil {
    private MessageUtil() {
    }
    public static String getMessageKeyWord(String strContent) {
        return null;//  CommonConfig.hasBase64Flag() ? JSON.parseObject(new String(Base64.getDecoder().decode(strContent), StandardCharsets.UTF_8)).getString(CommonConfig.getMessageKeyWord()) : JSON.parseObject(strContent).getString(CommonConfig.getMessageKeyWord());
    }
    public static String getDecodeContent(String strContentEncode) {
        return null;//  CommonConfig.hasBase64Flag() ? new String(Base64.getDecoder().decode(strContentEncode), StandardCharsets.UTF_8) : strContentEncode;
    }
    public static String getEncodeContent(String strContent) {
        return null;// CommonConfig.hasBase64Flag() ? new String(Base64.getEncoder().encode(strContent.getBytes(StandardCharsets.UTF_8)), StandardCharsets.UTF_8) : strContent;
    }
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/SecurityConfigSEVEN.java

@@ -0,0 +1,19 @@
package com.ai.bss.mock.utils;
import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.annotation.web.builders.HttpSecurity;
import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter;
@Configuration
@EnableWebSecurity
public class SecurityConfigSEVEN extends WebSecurityConfigurerAdapter {
    @Override
    protected void configure(HttpSecurity http) throws Exception {
        //super.configure(http);
        //配置不需要登陆验证
        http.authorizeRequests().anyRequest().permitAll().and().logout().permitAll();
    }
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/makedata/MakeBeidouData.java

@@ -0,0 +1,197 @@
package com.ai.bss.mock.utils.makedata;
import com.ai.bss.mock.utils.HexStringUtil;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.Calendar;
public class MakeBeidouData {
	/**
	 * 
	 * @param deviceId 设备ID
	 * @param type 类型(0 定位；1 自动报警；2手动报警；4离线)
	 * @param longitudeDouble 经度
	 * @param latitudeDouble 纬度
	 * @param dateString 时间
	 * @return
	 */
	public static String makeOneBeidouData(String deviceId, String type, String longitudeDouble, String latitudeDouble,String dateString) {
		return makeOneBeidouData(Integer.parseInt(deviceId), Integer.parseInt(type), Double.parseDouble(longitudeDouble), Double.parseDouble(latitudeDouble), dateString);
	}
	
	/**
	 * 
	 * @param deviceId 设备ID
	 * @param type 类型(0 定位；1 自动报警；2手动报警；4离线)
	 * @param longitudeDouble 经度
	 * @param latitudeDouble 纬度
	 * @param dateString 时间
	 * @return
	 */
	public static String makeOneBeidouData(int deviceId, int type, double longitudeDouble, double latitudeDouble,String dateString) {
		//时间
		String time = getHexDate(dateString);
		
		//经度
		String longitude="45"+getHexCoord(longitudeDouble);
		
		//纬度
		String latitude="4e"+getHexCoord(latitudeDouble);
		
		int speed = 0; // 速度
		int orientation = 0; // 方位
		StringBuilder sb = new StringBuilder();
		sb.append("2454585858002a02426c60").append(getHex(deviceId, 6)).append("000000b000");
		switch (type) {
		case 0:
			// 定位
			sb.append("b0").append(time).append(longitude).append(latitude);
			sb.append(getHex(speed)).append(getHex(orientation));
			sb.append("87");// 业务状态
			break;
		case 1:
			// 自动报警
			sb.append("a2").append(time).append("01").append(longitude).append(latitude);
			sb.append(getHex(speed)).append(getHex(orientation));
			break;
		case 2:
			// 手动报警
			sb.append("a2").append(time).append("02").append(longitude).append(latitude);
			sb.append(getHex(speed)).append(getHex(orientation));
			break;
		case 4:
			// 离线
			sb.append("a2").append(time).append("04").append(longitude).append(latitude);
			sb.append(getHex(speed)).append(getHex(orientation));
			break;
		default:
			break;
		}
		sb.append(xorCheck(sb.toString().substring(36)));
		sb.append("00");
		sb.append(xorCheck(sb.toString()));
		return sb.toString();
	}
	private static String getHexDate(String dateString) {
		try {
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			 Calendar date = Calendar.getInstance();
			 date.setTime(sdf.parse(dateString));
		     String time = getHex(date.get(Calendar.YEAR) - 2000) + getHex(date.get(Calendar.MONTH) + 1) 
		        + getHex( date.get(Calendar.DAY_OF_MONTH)) + getHex(date.get(Calendar.HOUR_OF_DAY)) 
		        + getHex(date.get(Calendar.MINUTE)) + getHex(date.get(Calendar.SECOND));
		     return time;
		} catch (Exception e) {
			System.out.println(e.getMessage());
		}
		return "";
	}
	
	private static String getHexCoord(double data) {
		BigDecimal b1=new BigDecimal(String.valueOf(data));
		
		BigDecimal degreeBigDecimal=b1.setScale(0, BigDecimal.ROUND_DOWN);
		int degree=degreeBigDecimal.intValue();
		
		b1=b1.subtract(degreeBigDecimal).multiply(new BigDecimal("60"));
		BigDecimal minutesBigDecimal=b1.setScale(0, BigDecimal.ROUND_DOWN);
		int minutes=minutesBigDecimal.intValue();
		
		b1=b1.subtract(minutesBigDecimal).multiply(new BigDecimal("60"));
		BigDecimal secondBigDecimal=b1.setScale(0, BigDecimal.ROUND_DOWN);
		int second=secondBigDecimal.intValue();
		
		b1=b1.subtract(secondBigDecimal).multiply(new BigDecimal("10"));
		BigDecimal littleSecondBigDecimal=b1.setScale(0, BigDecimal.ROUND_HALF_UP);
		int littleSecond=littleSecondBigDecimal.intValue();
		
		String value=getHex(degree)+getHex(minutes)+getHex(second)+getHex(littleSecond);
		return value;
	}
	
	private static  String xorCheck(String data) {
		byte[] value = HexStringUtil.decodeHextoByte(data);
		byte result = 0;
		for (int i = 0; i < value.length; i++) {
			result ^= value[i];
		}
		return HexStringUtil.decodeBytetoHexString(result);
	}
	private static  String getHex(int data) {
		return getHex(data, 2);
	}
	private static  String getHex(int data, int len) {
		String result = Integer.toString(data, 16);
		if (result.length() < len) {
			StringBuilder sb = new StringBuilder();
			for (int i = 0; i < len - result.length(); i++) {
				sb.append("0");
			}
			sb.append(result);
			result = sb.toString();
		}
		return result;
	}
	
	
	public void test1111() {
		int deviceId = 318414; // 设备id
		int type = 4; // 0 定位；1 自动报警；2手动报警；4离线
		int year = 2020; // 年
		int month = 10; // 月
		int day = 16; // 日
		int hour = 10; // 时
		int minute = 50; // 分
		int second = 42; // 秒
		String longitude = "45762e0f00"; // 经度
		String latitude = "4e200e1608"; // 纬度
		int speed = 0; // 速度
		int orientation = 0; // 方位
		
		String time = getHex(year - 2000) + getHex(month) + getHex(day) + getHex(hour) + getHex(minute)
				+ getHex(second);
		StringBuilder sb = new StringBuilder();
		sb.append("2454585858002a02426c60").append(getHex(deviceId, 6)).append("000000b000");
		switch (type) {
		case 0:
			// 定位
			sb.append("b0").append(time).append(longitude).append(latitude);
			sb.append(getHex(speed)).append(getHex(orientation));
			sb.append("87");// 业务状态
			break;
		case 1:
			// 自动报警
			sb.append("a2").append(time).append("01").append(longitude).append(latitude);
			sb.append(getHex(speed)).append(getHex(orientation));
			break;
		case 2:
			// 手动报警
			sb.append("a2").append(time).append("02").append(longitude).append(latitude);
			sb.append(getHex(speed)).append(getHex(orientation));
			break;
		case 4:
			// 离线
			sb.append("a2").append(time).append("04").append(longitude).append(latitude);
			sb.append(getHex(speed)).append(getHex(orientation));
			break;
		default:
			break;
		}
		sb.append(xorCheck(sb.toString().substring(36)));
		sb.append("00");
		sb.append(xorCheck(sb.toString()));
		System.out.println(sb.toString());
	}
	
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/tcp/IpuTcpLongConnectClient.java

@@ -0,0 +1,46 @@
package com.ai.bss.mock.utils.tcp;
import com.ai.bss.mock.model.MockScenarioData;
import com.ai.bss.mock.utils.tcp.message.MessageGenerationManager;
import java.util.List;
/**
 * Ipu Tcp长连接测试
 *
 * @author lilb3@asiainfo.com
 * @since 2020-06-23
 **/
@SuppressWarnings("unused")
public class IpuTcpLongConnectClient {
    private static IpuTcpLongConnectClient ipuClient;
    private TcpLongConnectClient nettyClient;
    private IpuTcpLongConnectClient() {
    }
    public static IpuTcpLongConnectClient getInstance(String host, int serverPort) {
        if (ipuClient == null) {
            ipuClient = new IpuTcpLongConnectClient();
            ipuClient.nettyClient = new TcpLongConnectClient(host, serverPort, MessageGenerationManager.getMessage());
        }
        return ipuClient;
    }
    /**
     * Tcp长连接方式发送消息
     * @throws Exception 
     */
    public void test() throws Exception {
        nettyClient.test();
    }
    /**
     * Tcp长连接方式发送消息
     * @throws Exception
     */
    public void sendMockData(Long sId,List<MockScenarioData> mockScenarioDataList,Long frequency) throws Exception {
        nettyClient.sendMockData(sId,mockScenarioDataList,frequency);
    }
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/tcp/TcpLongConnectClient.java

@@ -0,0 +1,287 @@
package com.ai.bss.mock.utils.tcp;
import com.ai.bss.mock.constants.MockConstant;
import com.ai.bss.mock.model.MockScenarioData;
import com.ai.bss.mock.utils.HexStringUtil;
import com.ai.bss.mock.utils.makedata.MakeBeidouData;
import com.ai.ipu.common.date.TimeUtil;
import com.alibaba.fastjson.JSONObject;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
 * Tcp长连接客户端
 *
 * @author lilb3@asiainfo.com
 * @since 2019-10-16
 **/
@Slf4j
class TcpLongConnectClient {
//	private static final ILogger LOGGER = IpuLoggerFactory.createLogger(TcpLongConnectClient.class);
	// private static final int PERIOD = 1;
	static final long PER_SLEEP_TIME = 2000L;
	// private static final int SEND_TIMES = 6;
	private final String host;
	private final int port;
	private final String message;
	private Channel channel;
	private ChannelFuture channelFuture;
	private Bootstrap bootstrap;
	TcpLongConnectClient(String host, int port, String message) {
		this.host = host;
		this.port = port;
		this.message = message;
	}
	/**
	 * 抽取出该方法 (断线重连时使用)
	 *
	 * @throws InterruptedException 异常
	 */
	void doConnect() throws InterruptedException {
		channelFuture = bootstrap.connect(host, port).sync();
		channel = channelFuture.channel();
	}
	public void test() throws Exception {
		EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
		bootstrap = new Bootstrap();
		bootstrap.group(eventLoopGroup);
		bootstrap.channel(NioSocketChannel.class);
		bootstrap.option(ChannelOption.SO_REUSEADDR, true);
		ScheduledExecutorService scheduledExecutorService;
		TcpLongConnectClient tcpLongConnectClient = this;
		bootstrap.handler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel socketChannel) {
				// 消息最大长度:1*1024*1024=1M表示单条消息的最大长度
				// int maxFrameLength = 1*1024*1024;
				// netty基于分割符的自带解码器，根据提供的分隔符解析报文，这里是0x7e;
				// 解码器在查找分隔符的时候，达到该长度还没找到的话会抛异常
				// 设置分隔符为分号(###)
				// socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(maxFrameLength,
				// Unpooled.copiedBuffer(TCP_DELIMITER.getBytes(StandardCharsets.UTF_8))));
				socketChannel.pipeline().addLast(new TcpLongConnectClientHandler(tcpLongConnectClient));
			}
		});
		// 连接Tcp服务器
		int count=1;
		try {
			doConnect();
			if (channelFuture.isSuccess()) {
				String sendMessage = null;
				String deviceId=null;
				String type=null; //类型(0 定位；1 自动报警；2手动报警；4离线) 
				String longitudeDouble=null;
				String latitudeDouble=null; 
				String dateString=null;
				while (count<50) {
					try {
					System.out.println("请输入第"+count+"条数据(设备ID,类型,经度,纬度,时间):");
					
					//建立从控制台输入的类  
					BufferedReader stdin = new BufferedReader(new InputStreamReader(System.in));
					
					// 循环读取每行数据
					//while(true) {
					String parameters = stdin.readLine().trim();
					
					if (parameters==null||"".equals(parameters)) {
						continue;
						//break;
						
					}else if ("0".equals(parameters)) {
						//断开
						break;
						
					}else if ("1".equals(parameters)&&sendMessage!=null) {
						//重复发送
						dateString=TimeUtil.getSysTime();
						sendMessage = MakeBeidouData.makeOneBeidouData(deviceId, type, longitudeDouble, latitudeDouble, dateString);
						
					}else {
						String[] params=parameters.split(",");
						if (params.length==5) {//设备ID,类型,经度,纬度,时间
							deviceId=params[0].trim();
							type=params[1].trim();
							longitudeDouble=params[2].trim();
							latitudeDouble=params[3].trim();
							dateString=params[4].trim();
							sendMessage = MakeBeidouData.makeOneBeidouData(deviceId, type, longitudeDouble, latitudeDouble, dateString);
						}else if(params.length==4){//设备ID,类型,经度,纬度
							deviceId=params[0].trim();
							type=params[1].trim();
							longitudeDouble=params[2].trim();
							latitudeDouble=params[3].trim();
							dateString=TimeUtil.getSysTime();
							sendMessage = MakeBeidouData.makeOneBeidouData(deviceId, type, longitudeDouble, latitudeDouble, dateString);
						}else if(params.length==3){//类型,经度,纬度
							type=params[0].trim();
							longitudeDouble=params[1].trim();
							latitudeDouble=params[2].trim();
							dateString=TimeUtil.getSysTime();
							sendMessage = MakeBeidouData.makeOneBeidouData(deviceId, "0", longitudeDouble, latitudeDouble, dateString);
						}else if(params.length==2){//经度,纬度
							longitudeDouble=params[0].trim();
							latitudeDouble=params[1].trim();
							dateString=TimeUtil.getSysTime();
							sendMessage = MakeBeidouData.makeOneBeidouData(deviceId, "0", longitudeDouble, latitudeDouble, dateString);
						}else if(params.length==1&&parameters.startsWith("2454585858")){
							sendMessage = parameters;
						}else {
//							LOGGER.debug("输入错误");
							continue;
						}
					}
						
						String typeZh="";//类型(0 定位；1 自动报警；2手动报警；4离线) 
						switch (type) {
						case "0":
							typeZh="定位";
							break;
						case "1":
							typeZh="自动报警";
							break;
						case "2":
							typeZh="手动报警";
							break;
						case "4":
							typeZh="离线";
							break;
						default:
							break;
						}
					
						log.debug("设备("+deviceId+")的["+typeZh+"]数据: 经度="+longitudeDouble+",纬度="+latitudeDouble+",时间="+dateString);
						log.debug(String.format(count + ":Send Message: %s", sendMessage));
						//LOGGER.debug("send data:" + Arrays.toString(HexStringUtil.decodeHextoByte(sendMessage)));
						channel.writeAndFlush(Unpooled.copiedBuffer(HexStringUtil.decodeHextoByte(sendMessage)));
						count++;
						
						//TimeUnit.MILLISECONDS.sleep(PER_SLEEP_TIME);
					//}
					}catch (Exception e) {
						log.debug("输入错误，请重新输入");
					}
				}
			}
		} finally {
			try {
				if (null != channel) {
					channel.disconnect();
				}
			} catch (Exception e) {
				log.error("出现错误，连接断开");
			}
		}
		log.info("发送成功，共发送"+count+"条数据");
	}
	public void sendMockData(Long sId,List<MockScenarioData> mockScenarioDataList,Long frequency) throws Exception {
		EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
		bootstrap = new Bootstrap();
		bootstrap.group(eventLoopGroup);
		bootstrap.channel(NioSocketChannel.class);
		bootstrap.option(ChannelOption.SO_REUSEADDR, true);
		ScheduledExecutorService scheduledExecutorService;
		TcpLongConnectClient tcpLongConnectClient = this;
		bootstrap.handler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel socketChannel) {
				// 消息最大长度:1*1024*1024=1M表示单条消息的最大长度
				// int maxFrameLength = 1*1024*1024;
				// netty基于分割符的自带解码器，根据提供的分隔符解析报文，这里是0x7e;
				// 解码器在查找分隔符的时候，达到该长度还没找到的话会抛异常
				// 设置分隔符为分号(###)
				// socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(maxFrameLength,
				// Unpooled.copiedBuffer(TCP_DELIMITER.getBytes(StandardCharsets.UTF_8))));
				socketChannel.pipeline().addLast(new TcpLongConnectClientHandler(tcpLongConnectClient));
			}
		});
		// 连接Tcp服务器
		int count=0;
		try {
			doConnect();
			if (channelFuture.isSuccess()) {
				String sendMessage = null;
				String deviceId=null;
				String type=null; //类型(0 定位；1 自动报警；2手动报警；4离线)
				String longitudeDouble=null;
				String latitudeDouble=null;
				String dateString=null;
				for(MockScenarioData mockScenarioData:mockScenarioDataList) {
					if(MockConstant.MACK_STATUS.get(sId)==null){
						break;
					}
					type = mockScenarioData.getType();
					deviceId = mockScenarioData.getDeviceId();
					longitudeDouble = mockScenarioData.getLongitude();
					latitudeDouble = mockScenarioData.getLatitude();
					dateString = mockScenarioData.getTime();
					if (StringUtils.isEmpty(mockScenarioData.getTime())){
						dateString = TimeUtil.getSysTime();
					}
					try {
						System.out.println("data:" + JSONObject.toJSONString(mockScenarioData));
					log.debug("data:" + JSONObject.toJSONString(mockScenarioData));
						sendMessage = MakeBeidouData.makeOneBeidouData(deviceId, type, longitudeDouble, latitudeDouble, dateString);
						String typeZh="";//类型(0 定位；1 自动报警；2手动报警；4离线)
						switch (type) {
						case "0":
							typeZh="定位";
							break;
						case "1":
							typeZh="自动报警";
							break;
						case "2":
							typeZh="手动报警";
							break;
						case "4":
							typeZh="离线";
							break;
						default:
							break;
						}
						System.out.println("设备("+deviceId+")的["+typeZh+"]数据: 经度="+longitudeDouble+",纬度="+latitudeDouble+",时间="+dateString);
						System.out.println(String.format(count + ":Send Message: %s", sendMessage));
						channel.writeAndFlush(Unpooled.copiedBuffer(HexStringUtil.decodeHextoByte(sendMessage)));
						count++;
						TimeUnit.MILLISECONDS.sleep(frequency*1000);
					}catch (Exception e) {
						e.printStackTrace();
						log.error("数据错误，请重新输入");
					}
				}
			}
		} finally {
			try {
				if (null != channel) {
					channel.disconnect();
				}
			} catch (Exception e) {
				log.error("出现错误，连接断开");
			}
		}
		log.info("发送成功，共发送"+count+"条数据");
	}
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/tcp/TcpLongConnectClientHandler.java

@@ -0,0 +1,77 @@
package com.ai.bss.mock.utils.tcp;
import com.ai.bss.mock.utils.MessageUtil;
import com.ai.ipu.basic.log.ILogger;
import com.ai.ipu.basic.log.IpuLoggerFactory;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import java.nio.charset.StandardCharsets;
/**
 * Tcp长连接客户端处理类
 *
 * @author lilb3@asiainfo.com
 * @since 2020-04-09
 **/
@Slf4j
class TcpLongConnectClientHandler extends ChannelInboundHandlerAdapter {
//    private static final ILogger LOGGER = IpuLoggerFactory.createLogger(TcpLongConnectClientHandler.class);
    private static final long SLEEP_TIME = 10 * 1000L;
    private final TcpLongConnectClient tcpLongConnectClient;
    TcpLongConnectClientHandler(TcpLongConnectClient tcpLongConnectClient) {
        this.tcpLongConnectClient = tcpLongConnectClient;
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        log.info("TCP长连接客户端接入成功");
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
    	log.debug("连接断开");
    	System.out.println("连接断开");
		/*LOGGER.debug("连接断开，10s之后尝试重新连接服务器。");
		try {
		    TimeUnit.MILLISECONDS.sleep(SLEEP_TIME);
		} catch (InterruptedException e) {
		    LOGGER.error(e.getMessage(), e);
		    Thread.currentThread().interrupt();
		}
		try {
		    tcpLongConnectClient.doConnect();
		} catch (InterruptedException e) {
		    LOGGER.error(e.getMessage(), e);
		    Thread.currentThread().interrupt();
		}*/
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf byteBuf = (ByteBuf) msg;
        byte[] content = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(content);
        String strContentEncode = new String(content, StandardCharsets.UTF_8);
//            String strContentDecode = MessageUtil.getDecodeContent(strContentEncode);
        log.debug(String.format("长连接TCP客户端收到的信息: %s", strContentEncode));
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("客户端捕获到异常。", cause);
    }
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            //如果规定时内没有读写时间，则触发一个消息
            //ctx.writeAndFlush((CommonConfig.getTcpHeartBeat() + TCP_DELIMITER).getBytes(StandardCharsets.UTF_8));
        	
        }
    }
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/tcp/message/DefaultMessageGeneration.java

@@ -0,0 +1,27 @@
package com.ai.bss.mock.utils.tcp.message;
import com.ai.bss.mock.utils.MessageUtil;
import com.alibaba.fastjson.JSONObject;
/**
 * 消息测试
 *
 * @author lilb3@asiainfo.com
 * @since 2020-03-30
 **/
public class DefaultMessageGeneration implements IMessageGeneration {
    private static final String IMEI = "imeiTest";
    private static String getMessage0(String topic) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("topic", topic);
        jsonObject.put("imei", IMEI);
        jsonObject.put("sendTime", System.currentTimeMillis() + "");
        return jsonObject.toString();
    }
    @Override
    public String getMessage(String topic) {
        return MessageUtil.getEncodeContent(getMessage0(topic));
    }
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/tcp/message/IMessageGeneration.java

@@ -0,0 +1,19 @@
package com.ai.bss.mock.utils.tcp.message;
/**
 * 消息发送接口
 *
 * @author lilb3@asiainfo.com
 * @since 2020-06-22
 **/
@SuppressWarnings("unused")
public interface IMessageGeneration {
    /**
     * 获取测试消息
     *
     * @param topic 消息主题
     * @return 测试消息
     */
    String getMessage(String topic);
}

ebc-mock-service/src/main/java/com/ai/bss/mock/utils/tcp/message/MessageGenerationManager.java

@@ -0,0 +1,74 @@
package com.ai.bss.mock.utils.tcp.message;
import com.ai.bss.mock.constants.MockConstant;
import com.ai.ipu.basic.instance.InstanceManager;
import java.util.HashMap;
import java.util.Map;
/**
 * 增加message默认处理类，分离消息消费逻辑
 *
 * @author lilb3@asiainfo.com
 * @since 2020-06-22
 **/
@SuppressWarnings("unused")
public class MessageGenerationManager {
//    private static final ILogger LOGGER = IpuLoggerFactory.createLogger(MessageGenerationManager.class);
    private static final String SPLIT_DASH = MockConstant.SPLIT_DASH;
    private static final Map<String, IMessageGeneration> INSTANCES = new HashMap<>();
    private static IMessageGeneration defaultMessageGeneration;
    private MessageGenerationManager() {
        throw new RuntimeException("MessageGenerationManager无法被实例化");
    }
    @SuppressWarnings("unchecked")
    private static IMessageGeneration createMessageGeneration(String className) throws Exception {
        IMessageGeneration messageGeneration = INSTANCES.get(className);
        if (messageGeneration == null) {
            synchronized (INSTANCES) {
                if (INSTANCES.get(className) == null) {
                    Class<?> clazz = Class.forName(className);
                    messageGeneration = InstanceManager.createBean((Class<IMessageGeneration>) clazz);
                    INSTANCES.put(className, messageGeneration);
                }
            }
        }
        return INSTANCES.get(className);
    }
//    public static <Type> Type createBean(Class<Type> clazz) throws Exception {
//        return createBean(clazz, true);
//    }
    private static IMessageGeneration getDefaultMessageGeneration() {
        if (defaultMessageGeneration == null) {
            defaultMessageGeneration = new DefaultMessageGeneration();
        }
        return defaultMessageGeneration;
    }
    public static void setDefaultMessageGeneration(IMessageGeneration defaultMessageGeneration) {
        MessageGenerationManager.defaultMessageGeneration = defaultMessageGeneration;
    }
    public static String getMessage() {
    	//String topic = CommonConfig.getTopics()[new Random().nextInt(CommonConfig.getTopics().length)];
        String topic = "networkPacket";
        return getMessage(topic);
    }
    public static String getMessage(String topic) {
        try {
            IMessageGeneration messageGeneration = MessageGenerationManager.getDefaultMessageGeneration();;
            if (messageGeneration != null) {
                return messageGeneration.getMessage(topic);
            }
        } catch (Exception e) {
//            LOGGER.error("Test Message生成异常", e);
        }
        return "";
    }
}

ebc-mock-service/src/main/resources/application.properties

@@ -0,0 +1,55 @@
spring.application.name=WorkTaskSpec
server.port=8086
# DATASOURCE (DataSourceAutoConfiguration & DataSourceProperties)
#spring.datasource.url=jdbc:mysql://localhost:3306/cmp
#spring.datasource.url=jdbc:mysql://10.19.14.28:3306/work_order?useUnicode=true&ampcharacterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
#spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
#spring.datasource.username=aibp
#spring.datasource.password=Aibp@123
#spring.datasource.url=jdbc:mysql://10.11.20.120:3306/common_frm?useUnicode=true&ampcharacterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
#spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
#spring.datasource.username=comon_frm
#spring.datasource.password=1qaz@WSX
spring.datasource.url=jdbc:mysql://47.105.160.21:3309/dmp?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.username=cmp
spring.datasource.password=cmp@123
# JPA (JpaBaseConfiguration, HibernateJpaAutoConfiguration)
#spring.jpa.database=default
spring.jpa.database-platform=org.hibernate.dialect.MySQL5Dialect
spring.jpa.hibernate.ddl-auto=update
spring.jpa.show-sql=true
spring.jpa.properties.hibernate.format_sql=true
spring.jpa.properties.hibernate.generate_statistics=false
spring.main.allow-bean-definition-overriding=true
# CACHE
#spring.cache.type=ehcache
#spring.cache.ehcache.config=ehcache.xml
# LOGGING
logging.level.com.ai=info
logging.level.org.springframework.data=info
#============== kafka ===================
kafka.consumer.zookeeper.connect=47.105.160.21:2100
kafka.consumer.servers=47.105.160.21:9090
kafka.consumer.enable.auto.commit=true
kafka.consumer.session.timeout=6000
kafka.consumer.auto.commit.interval=100
kafka.consumer.auto.offset.reset=latest
kafka.consumer.topic=productMessage
kafka.consumer.group.id=productMessage
kafka.consumer.concurrency=10
kafka.producer.servers=47.105.160.21:9090
kafka.producer.retries=0
kafka.producer.batch.size=4096
kafka.producer.linger=1
kafka.producer.buffer.memory=40960
kafka.producer.topic=67message

ebc-mock-service/src/test/resources/application.properties

@@ -0,0 +1,34 @@
spring.application.name=WorkTaskSpec
server.port=8086
# DATASOURCE (DataSourceAutoConfiguration & DataSourceProperties)
#spring.datasource.url=jdbc:mysql://localhost:3306/cmp
#spring.datasource.url=jdbc:mysql://10.19.14.28:3306/work_order?useUnicode=true&ampcharacterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
#spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
#spring.datasource.username=aibp
#spring.datasource.password=Aibp@123
#spring.datasource.url=jdbc:mysql://10.11.20.120:3306/common_frm?useUnicode=true&ampcharacterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
#spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
#spring.datasource.username=comon_frm
#spring.datasource.password=1qaz@WSX
spring.datasource.url=jdbc:mysql://10.19.90.34:3307/energy?useUnicode=true&ampcharacterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.username=ebc
spring.datasource.password=ebc@123
# JPA (JpaBaseConfiguration, HibernateJpaAutoConfiguration)
#spring.jpa.database=default
spring.jpa.database-platform=org.hibernate.dialect.MySQL5Dialect
spring.jpa.hibernate.ddl-auto=update
spring.jpa.show-sql=true
spring.jpa.properties.hibernate.format_sql=true
spring.jpa.properties.hibernate.generate_statistics=false
spring.main.allow-bean-definition-overriding=true
# CACHE
#spring.cache.type=ehcache
#spring.cache.ehcache.config=ehcache.xml
# LOGGING
logging.level.com.ai=info
logging.level.org.springframework.data=info