增加Mock服务指定topic功能

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

@@ -1,6 +1,5 @@
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;
@@ -9,7 +8,7 @@ import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
import org.springframework.scheduling.annotation.EnableAsync;
/**
 * @author zhangfeng
 * @author wangdong
 */
@EnableJpaRepositories(basePackages = "com.ai.bss")
@@ -19,15 +18,8 @@ import org.springframework.scheduling.annotation.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/service/impl/MockProcessImpl.java

@@ -10,6 +10,7 @@ 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.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
@@ -24,18 +25,18 @@ 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;
//    private static String kafkaServers = "47.105.160.21:9090";
    private static String kafkaServers = "10.19.90.34:9090";
//    @Value("${kafka.producer.servers:Empty}")
//    private static String kafkaServers;
    private static String kafkaServers = "47.105.160.21:9090";
    @Value("${kafka.producer.servers}")
    private static String servers;
    @Async
    @Override
@@ -174,19 +175,20 @@ public class MockProcessImpl implements MockProcess {
    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);
        }
//        KafkaTemplate kafkaTemplate = kafkaTemplateMap.get(kafkaServers);
//        if(kafkaTemplate == null){
//            //new 实例
//            KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(kafkaServers);
//            kafkaTemplate = kafkaProducerConfig.kafkaTemplate();
//            kafkaTemplateMap.put(kafkaServers,kafkaTemplate);
//        }
        System.out.println(servers);
        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);
//                Object result = kafkaTemplate.send(dataPoint.getPushTopic(), JSONObject.toJSONString(dataPoint)).get();
                Object result = KafkaProducerConfig.sendForKafka(dataPoint.getPushTopic(), JSONObject.toJSONString(dataPoint));
//                kafkaTemplate.flush();
                TimeUnit.MILLISECONDS.sleep(frequency*1000);
@@ -212,17 +214,18 @@ public class MockProcessImpl implements MockProcess {
            kafkaTemplate = kafkaProducerConfig.kafkaTemplate();
            kafkaTemplateMap.put(kafkaServers,kafkaTemplate);
        }
        try {
//            for(DataPoint dataPoint:dataPointList){
            Object result = kafkaTemplate.send(dataPoint.getPushTopic(), JSONObject.toJSONString(dataPoint)).get();
            kafkaTemplate.flush();
//            Object result = KafkaProducerConfig.sendForKafka(dataPoint.getPushTopic(), JSONObject.toJSONString(dataPoint));
//            kafkaTemplate.flush();
//                TimeUnit.MILLISECONDS.sleep(frequency*1000);
            log.debug("It's successful send msg to Kafka:" + kafkaServers + "服务器地址以及topic:" + dataPoint.getPushTopic() + " 内容：" + JSONObject.toJSONString(dataPoint) + " 结果" + result);
//            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 (InterruptedException e) {
//            e.printStackTrace();
//            return false;
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage());

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

@@ -1,8 +1,5 @@
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 {

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

@@ -1,5 +1,6 @@
package com.ai.bss.mock.utils;
import com.alibaba.fastjson.JSONObject;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerConfig;
@@ -11,6 +12,7 @@ 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 org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
@@ -19,10 +21,12 @@ import java.util.Map;
@Configuration
@EnableKafka
@NoArgsConstructor
@Component
public class KafkaProducerConfig {
    @Value("${kafka.producer.servers:Empty}")
    private static String servers;
//    private static String servers = "47.105.160.21:9090";
    private static String servers = "10.19.90.34:9090";
    @Value("${kafka.producer.retries:0}")
    private  int retries;
    @Value("${kafka.producer.batch.size:4096}")
@@ -37,9 +41,9 @@ public class KafkaProducerConfig {
    public KafkaProducerConfig(String inputServers){
        servers = inputServers;
        this.retries = 0;
        this.batchSize = 4096;
        this.batchSize = 16384;
        this.linger = 1;
        this.bufferMemory = 40960;
        this.bufferMemory = 33554432;
        this.maxBlockTime = 6000;
    }
@@ -66,6 +70,10 @@ public class KafkaProducerConfig {
        return new KafkaTemplate<String, String>(producerFactory());
    }
//
    private static Map<String, KafkaTemplate>  kafkaTemplateMap = new HashMap<>();
    static KafkaTemplate  kafkaTemplate = null;
    /**
     * 推送消息到订阅的KAFKA服务器地址以及topic
     * @param kafkaTopic  订阅的kafka topic
@@ -73,12 +81,25 @@ public class KafkaProducerConfig {
     * @return
     */
    public static Boolean sendForKafka(String kafkaTopic,String content) {
        KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(servers);
        KafkaTemplate kafkaTemplate =  kafkaProducerConfig.kafkaTemplate();
//        KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(servers);
//        KafkaTemplate kafkaTemplate =  kafkaProducerConfig.kafkaTemplate();
        kafkaTemplate = kafkaTemplateMap.get(servers);
        if(kafkaTemplate == null){
            //new 实例
//            KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig(servers);
            KafkaProducerConfig kafkaProducerConfig = new KafkaProducerConfig();
            kafkaTemplate = kafkaProducerConfig.kafkaTemplate();
            kafkaTemplateMap.put(servers,kafkaTemplate);
        }
        try {
            System.out.println("send msg to Kafka:" + servers + "服务器地址以及topic:" + kafkaTopic + " 内容：" + content );
            Object result = kafkaTemplate.send(kafkaTopic,content ).get();
            log.debug("It's successful send msg to Kafka:" + servers + "服务器地址以及topic:" + kafkaTopic + " 内容：" + content + " 结果" + result);
            kafkaTemplate.flush();
        } catch (InterruptedException e) {
            log.error(e.getMessage());

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

@@ -11,10 +11,13 @@ server.port=8086
#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.url=jdbc:mysql://47.105.160.21:3309/dmp?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
spring.datasource.url=jdbc:mysql://10.19.90.34:3307/energy?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
#spring.datasource.username=cmp
#spring.datasource.password=cmp@123
spring.datasource.username=ebc
spring.datasource.password=ebc@123
# JPA (JpaBaseConfiguration, HibernateJpaAutoConfiguration)
#spring.jpa.database=default