package com.whyc.controller;
|
|
import com.alibaba.fastjson2.JSON;
|
import com.whyc.dto.Response;
|
import com.whyc.dto.interfaceB.DataB;
|
import com.whyc.dto.interfaceC.DataC;
|
import com.whyc.dto.interfaceC.PointC;
|
import org.apache.kafka.clients.admin.KafkaAdminClient;
|
import org.apache.kafka.common.KafkaFuture;
|
import org.springframework.kafka.core.KafkaAdmin;
|
import org.springframework.kafka.core.KafkaTemplate;
|
import org.springframework.web.bind.annotation.*;
|
|
import javax.annotation.Resource;
|
import java.util.LinkedList;
|
import java.util.List;
|
import java.util.Set;
|
import java.util.concurrent.LinkedBlockingDeque;
|
import java.util.concurrent.ThreadPoolExecutor;
|
import java.util.concurrent.TimeUnit;
|
|
import static java.lang.Thread.sleep;
|
|
@RestController
|
@RequestMapping("dcim_north")
|
public class DcimNorthController{
|
|
@Resource
|
private KafkaTemplate template;
|
|
@Resource
|
private KafkaTemplate<Object,List<DataC>> batchTemplate;
|
|
/**
|
* 这个仅测试使用,正式环境不使用
|
* @param dataB
|
* @return
|
*/
|
@PostMapping("upload_config")
|
public Response uploadConfig(@RequestBody DataB dataB){
|
//不处理
|
return new Response().set(1,"配置上传完成");
|
}
|
|
@GetMapping("benchmarkTest")
|
public void benchmarkTest(){
|
long currentTimeMillis = System.currentTimeMillis();
|
//生产80万条数据,然后发送
|
List<String> dataCList = new LinkedList<>();
|
for (int i = 0; i < 8000; i++) {
|
DataC dataC = new DataC();
|
dataC.setDevice_id(i+"");
|
dataC.setDevice_type("test");
|
dataC.setIdc("test");
|
|
List<PointC> pointCList = new LinkedList<>();
|
PointC pointC = new PointC();
|
pointC.setPoint_code(i);
|
pointC.setTag("test");
|
pointC.setTimestamp(currentTimeMillis);
|
pointC.setValue(Double.parseDouble(i+""));
|
pointCList.add(pointC);
|
dataC.setPoints(pointCList);
|
|
String dataCStr = JSON.toJSONString(dataC);
|
dataCList.add(dataCStr);
|
}
|
//long startTime = System.currentTimeMillis();
|
while (true) {
|
for (int i = 0; i < dataCList.size(); i++) {
|
String dataCStr = dataCList.get(i);
|
template.send("topic_device_data2", dataCStr);
|
/*.addCallback(
|
success->{
|
//System.out.println(success);
|
},
|
failure->{
|
//System.err.println(failure.getMessage());
|
//failure.printStackTrace();
|
});*/
|
}
|
long endTime = System.currentTimeMillis();
|
//System.out.println("花费时长(毫秒):" + (endTime - startTime));
|
try {
|
sleep(1000);
|
} catch (InterruptedException e) {
|
e.printStackTrace();
|
}
|
}
|
|
}
|
|
@GetMapping("benchmarkTest3")
|
public void benchmarkTest3(){
|
long currentTimeMillis = System.currentTimeMillis();
|
//生产80万条数据,然后发送
|
List<DataC> dataCList = new LinkedList<>();
|
for (int i = 0; i < 800000; i++) {
|
DataC dataC = new DataC();
|
dataC.setDevice_id(i+"");
|
dataC.setDevice_type("test");
|
dataC.setIdc("test");
|
|
List<PointC> pointCList = new LinkedList<>();
|
PointC pointC = new PointC();
|
pointC.setPoint_code(i);
|
pointC.setTag("test");
|
pointC.setTimestamp(currentTimeMillis);
|
pointC.setValue(Double.parseDouble(i+""));
|
pointCList.add(pointC);
|
dataC.setPoints(pointCList);
|
|
dataCList.add(dataC);
|
}
|
long startTime = System.currentTimeMillis();
|
batchTemplate.send("topic_device_data2",dataCList);
|
/*for (int i = 0; i < dataCList.size(); i++) {
|
pool.execute(()->{
|
template.send("topic_device_data2",dataCStr).addCallback(
|
success->{
|
//System.out.println(success);
|
},
|
failure->{
|
//System.err.println(failure.getMessage());
|
//failure.printStackTrace();
|
});
|
});
|
}*/
|
long endTime = System.currentTimeMillis();
|
System.out.println("花费时长(毫秒):"+(endTime-startTime));
|
|
}
|
|
@PostMapping("push_data")
|
public Response pushData(@RequestBody DataC dataC){
|
//点位数据发送到kafka topic_device_data
|
String dataCStr = JSON.toJSONString(dataC);
|
template.send("topic_device_data2",dataCStr).addCallback(
|
success->{
|
System.out.println(success);
|
},
|
failure->{
|
System.err.println(failure.getMessage());
|
failure.printStackTrace();
|
});
|
return new Response().set(1,"推送完成");
|
}
|
|
}
|
