Skip to content

D
device-new-platform
Commit a3bcf020 authored by 赵啸非's avatar 赵啸非
Browse files
Options

修改消息组件

parent 6f294d8a
Hide whitespace changes
Inline Side-by-side
Showing with 504 additions and 339 deletions
common-lib/src/main/java/com/mortals/xhx/queue/TbQueueProducer.java
View file @ a3bcf020
...@@ -16,4 +16,7 @@ public interface TbQueueProducer<T extends TbQueueMsg> { ...@@ -16,4 +16,7 @@ public interface TbQueueProducer<T extends TbQueueMsg> {
void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback); void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback);
void stop(); void stop();
void queueDeclare(TopicPartitionInfo tpi, TbQueueCallback callback);
} }
common-lib/src/main/java/com/mortals/xhx/queue/kafka/TbKafkaProducerTemplate.java
View file @ a3bcf020
...@@ -44,7 +44,7 @@ public class TbKafkaProducerTemplate<T extends TbQueueMsg> implements TbQueuePro ...@@ -44,7 +44,7 @@ public class TbKafkaProducerTemplate<T extends TbQueueMsg> implements TbQueuePro
private Set<TopicPartitionInfo> topics; private Set<TopicPartitionInfo> topics;
@Builder @Builder
private TbKafkaProducerTemplate(TbKafkaSettings settings,String defaultTopic) { private TbKafkaProducerTemplate(TbKafkaSettings settings, String defaultTopic) {
this.settings = settings; this.settings = settings;
//初始化生产者参数 //初始化生产者参数
this.producer = new KafkaProducer<>(settings.toProducerProps()); this.producer = new KafkaProducer<>(settings.toProducerProps());
...@@ -90,5 +90,10 @@ public class TbKafkaProducerTemplate<T extends TbQueueMsg> implements TbQueuePro ...@@ -90,5 +90,10 @@ public class TbKafkaProducerTemplate<T extends TbQueueMsg> implements TbQueuePro
} }
} }
@Override
public void queueDeclare(TopicPartitionInfo tpi, TbQueueCallback callback) {
// TODO: 2022/4/29 创建kafka队列
}
} }
common-lib/src/main/java/com/mortals/xhx/queue/provider/TbCoreQueueProducerProvider.java
View file @ a3bcf020
...@@ -20,20 +20,13 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider { ...@@ -20,20 +20,13 @@ public class TbCoreQueueProducerProvider implements TbQueueProducerProvider {
* 消息队列提供 * 消息队列提供
*/ */
@Autowired @Autowired
private TbCoreQueueFactory tbQueueProvider; private TbCoreQueueFactory tbQueueProvider;
/** /**
* 消息队列生产者 * 消息队列生产者
*/ */
private TbQueueProducer<TbQueueMsg> queueProducer; private TbQueueProducer<TbQueueMsg> queueProducer;
// public TbCoreQueueProducerProvider(TbCoreQueueFactory tbQueueProvider) {
// this.tbQueueProvider = tbQueueProvider;
// }
//
@PostConstruct @PostConstruct
public void init() { public void init() {
log.info("消息队列生产服务开始..."); log.info("消息队列生产服务开始...");
......
common-lib/src/main/java/com/mortals/xhx/queue/rabbitmq/TbRabbitMqConsumerTemplate.java
View file @ a3bcf020
...@@ -56,7 +56,7 @@ public class TbRabbitMqConsumerTemplate<T extends TbQueueMsg> extends AbstractTb ...@@ -56,7 +56,7 @@ public class TbRabbitMqConsumerTemplate<T extends TbQueueMsg> extends AbstractTb
GetResponse getResponse = channel.basicGet(queue, true); GetResponse getResponse = channel.basicGet(queue, true);
return getResponse; return getResponse;
} catch (IOException e) { } catch (IOException e) {
log.error("Failed to get messages from queue: [{}]" , queue,e); log.error("Failed to get messages from queue: {},{}" , queue,e);
return null; return null;
} }
}).filter(Objects::nonNull).collect(Collectors.toList()); }).filter(Objects::nonNull).collect(Collectors.toList());
......
common-lib/src/main/java/com/mortals/xhx/queue/rabbitmq/TbRabbitMqProducerTemplate.java
View file @ a3bcf020
...@@ -79,13 +79,17 @@ public class TbRabbitMqProducerTemplate<T extends TbQueueMsg> implements TbQueue ...@@ -79,13 +79,17 @@ public class TbRabbitMqProducerTemplate<T extends TbQueueMsg> implements TbQueue
AMQP.BasicProperties properties = new AMQP.BasicProperties(); AMQP.BasicProperties properties = new AMQP.BasicProperties();
try { try {
if (!topicIfNotExist) { if (!topicIfNotExist) {
//topic不存在创建通道队列
channel.queueDeclare(tpi.getTopic(), true, false, false, null); channel.queueDeclare(tpi.getTopic(), true, false, false, null);
} }
if (!innerExists(tpi.getExchangeName(), channel)) { if (!innerExists(tpi.getExchangeName(), channel)) {
//判断交换机是否存在，如果不存在则创建后与新加入的队列绑定
channel = connection.createChannel(); channel = connection.createChannel();
channel.exchangeDeclare(tpi.getExchangeName(), BuiltinExchangeType.DIRECT); channel.exchangeDeclare(tpi.getExchangeName(), BuiltinExchangeType.DIRECT, true, false, null);
} }
channel.basicPublish(tpi.getExchangeName(), tpi.getFullTopicName(), properties, JSON.toJSONString(new DefaultTbQueueMsg(msg)).getBytes()); channel.queueBind(tpi.getTopic(), tpi.getExchangeName(), tpi.getTopic());
//channel.queueBind()
channel.basicPublish(tpi.getExchangeName(), tpi.getTopic(), properties, JSON.toJSONString(new DefaultTbQueueMsg(msg)).getBytes());
if (callback != null) { if (callback != null) {
callback.onSuccess(null); callback.onSuccess(null);
} }
...@@ -119,6 +123,28 @@ public class TbRabbitMqProducerTemplate<T extends TbQueueMsg> implements TbQueue ...@@ -119,6 +123,28 @@ public class TbRabbitMqProducerTemplate<T extends TbQueueMsg> implements TbQueue
} }
} }
@Override
public void queueDeclare(TopicPartitionInfo tpi, TbQueueCallback callback) {
Boolean topicIfNotExist = createTopicIfNotExist(tpi);
try {
if (!topicIfNotExist) {
//topic不存在创建通道队列
channel.queueDeclare(tpi.getTopic(), true, false, false, null);
}
if (!innerExists(tpi.getExchangeName(), channel)) {
//判断交换机是否存在，如果不存在则创建后与新加入的队列绑定
channel = connection.createChannel();
channel.exchangeDeclare(tpi.getExchangeName(), BuiltinExchangeType.DIRECT, true, false, null);
}
callback.onSuccess(null);
} catch (IOException e) {
log.error("Failed publish message: {}.", e);
if (callback != null) {
callback.onFailure(e);
}
}
}
private Boolean createTopicIfNotExist(TopicPartitionInfo tpi) { private Boolean createTopicIfNotExist(TopicPartitionInfo tpi) {
if (topics.contains(tpi)) { if (topics.contains(tpi)) {
return true; return true;
......
device-manager-ui/admin/src/assets/mixins/table.js
View file @ a3bcf020
...@@ -148,9 +148,6 @@ export default { ...@@ -148,9 +148,6 @@ export default {
formatterYES(row, column, val) { formatterYES(row, column, val) {
const content = formatter(this.tableData, column, val); const content = formatter(this.tableData, column, val);
console.log("content:"+content)
//return content;
if (content) { if (content) {
if (val == '0') { if (val == '0') {
return <el-tag type={'danger'} size='mini'>{content}</el-tag> return <el-tag type={'danger'} size='mini'>{content}</el-tag>
......
device-manager-ui/admin/src/router.js
View file @ a3bcf020
...@@ -30,7 +30,10 @@ const router = new Router({ ...@@ -30,7 +30,10 @@ const router = new Router({
...restBuilder('area', 'system/area'), // 系统管理-区域管理 ...restBuilder('area', 'system/area'), // 系统管理-区域管理
builder('site/list', 'system/site/index'),//站点 builder('site/list', 'system/site/index'),//站点
...restBuilder('device', 'device'),//
...restBuilder('platform', 'platform'),//平台
...restBuilder('product', 'product'),//产品
...restBuilder('device', 'device'),//设备
...restBuilder('device/log', 'device/log'),// ...restBuilder('device/log', 'device/log'),//
......
device-manager-ui/admin/src/views/device/dialogshow.vue
View file @ a3bcf020
...@@ -4,8 +4,9 @@ ...@@ -4,8 +4,9 @@
<el-form ref="form" :model="form" :rules="rules" label-width="120px"> <el-form ref="form" :model="form" :rules="rules" label-width="120px">
<el-row> <el-row>
<!-- <Field :span="20" label="设备名称" prop="deviceName" v-model="form.deviceName" placeholder="请输入设备名称"/> --> <!-- <Field :span="20" label="设备名称" prop="deviceName" v-model="form.deviceName" placeholder="请输入设备名称"/> -->
<Field :span="20" label="设备编码" prop="deviceCode" v-model="form.deviceCode" type="textarea" placeholder="请输入设备编码"/> <Field :span="20" label="设备编码" prop="deviceCode" v-model="form.deviceCode" placeholder="请输入设备编码"/>
<Field :span="20" label="设备类型" prop="deviceType" v-model="form.deviceType" type="select" :enumData="dict.deviceType" placeholder="请选择设备类型"/> <Field :span="20" label="平台类型" prop="platformId" v-model="form.platformId" type="select" :enumData="dict.platformId" placeholder="请选择平台类型"/>
<Field :span="20" label="产品类型" prop="productId" v-model="form.productId" type="select" :enumData="dict.productId" placeholder="请选择产品类型"/>
<Field :span="20" label="设备的MAC地址" prop="deviceMac" v-model="form.deviceMac" placeholder="请输入设备的MAC地址"/> <Field :span="20" label="设备的MAC地址" prop="deviceMac" v-model="form.deviceMac" placeholder="请输入设备的MAC地址"/>
<Field :span="20" label="中心设备编码" v-model="form.centernum" placeholder="请输入中心设备编码"/> <Field :span="20" label="中心设备编码" v-model="form.centernum" placeholder="请输入中心设备编码"/>
...@@ -23,8 +24,7 @@ ...@@ -23,8 +24,7 @@
/> />
<Field :span="20" label="备注" prop="deviceRemark" v-model="form.deviceRemark" type="textarea" placeholder="请输入备注"/> <Field :span="20" label="备注" prop="deviceRemark" v-model="form.deviceRemark" type="textarea" placeholder="请输入备注"/>
<Field :span="20" label="启用状态 " prop="status" v-model="form.status" type="radio" :enumData="dict.status" placeholder="请选择启用状态 "/>
<Field :span="20" label="启用状态 " prop="status" v-model="form.status" type="radio" :enumData="dict.status" placeholder="请选择启用状态 "/>
</el-row> </el-row>
......
device-manager-ui/admin/src/views/device/show.vue deleted 100644 → 0
View file @ 6f294d8a
<template>
<layout-form>
<el-form :model="form" :loading="loading" :rules="rules" size='small' style="width:100%" label-width='120px' ref="form">
<el-row>
<Field label="设备名称" prop="deviceName" v-model="form.deviceName" placeholder="请输入设备名称"/>
<Field label="设备编码" prop="deviceCode" v-model="form.deviceCode" type="textarea" placeholder="请输入设备编码"/>
<Field label="设备类型" prop="deviceType" v-model="form.deviceType" type="select" :enumData="dict.deviceType" placeholder="请选择设备类型"/>
<Field label="设备的MAC地址" prop="deviceMac" v-model="form.deviceMac" placeholder="请输入设备的MAC地址"/>
<Field label="设备访问ip" prop="ip" v-model="form.ip" placeholder="请输入设备访问ip"/>
<Field label="中心设备编码" prop="centernum" v-model="form.centernum" placeholder="请输入中心设备编码"/>
<Field label="端口" prop="port" v-model="form.port" placeholder="请输入端口"/>
<Field label="站点编号" prop="siteNum" v-model="form.siteNum" placeholder="请输入站点编号"/>
<Field label="设备生产厂商ID" prop="deviceFirmId" v-model="form.deviceFirmId" placeholder="请输入设备生产厂商ID"/>
<Field label="设备生产厂商名称" prop="deviceFirmname" v-model="form.deviceFirmname" placeholder="请输入设备生产厂商名称"/>
<Field label="在线状态 " prop="deviceOnlineStatus" v-model="form.deviceOnlineStatus" type="select" :enumData="dict.deviceOnlineStatus" placeholder="请选择在线状态 "/>
<Field label="启用状态 " prop="status" v-model="form.status" type="select" :enumData="dict.status" placeholder="请选择启用状态 "/>
<Field label="备注" prop="deviceRemark" v-model="form.deviceRemark" type="textarea" placeholder="请输入备注"/>
<Field label="最近上线时间" prop="onlineTime" v-model="form.onlineTime" type="date" />
<Field label="最近离线时间" prop="offlineTime" v-model="form.offlineTime" type="date" />
</el-row>
<form-buttons @submit='submitForm' :noSaveBtn="pageInfo.type === 'view'"/>
</el-form>
</layout-form>
</template>
<script>
import form from "@/assets/mixins/form";
export default {
mixins: [form],
components: {
},
methods: {
},
data() {
return {
toString:[
"deviceType",
"deviceOnlineStatus",
"status",
],
rules: {
deviceName: [
{required: true,message: "请输入设备名称", trigger: "blur" },
{max: 20,message: "最多只能录入20个字符",trigger: "blur",},
],
deviceType: [
{required: true,message: "请输入设备类型", trigger: "blur" },
],
deviceOnlineStatus: [
{required: true,message: "请输入在线状态 ", trigger: "blur" },
],
status: [
{required: true,message: "请输入启用状态 ", trigger: "blur" },
],
createTime: [
{required: true,message: "请选择创建时间" },
],
}
};
}
};
</script>
\ No newline at end of file
device-manager-ui/admin/src/views/device/view.vue deleted 100644 → 0
View file @ 6f294d8a
<template>
<layout-view>
<el-descriptions :title="title" :column="column" :size="size" :colon="false" border>
<template slot="title">
<i class="el-icon-tickets"></i>
基本详细信息
</template>
<template slot="extra">
<el-button type="primary" @click="$router.go(-1)" size="small">返回</el-button>
</template>
<el-descriptions-item label="设备名称" label-class-name="labelClass" content-class-name="contentClass">
{{form.deviceName}}
</el-descriptions-item>
<el-descriptions-item label="设备编码" label-class-name="labelClass" content-class-name="contentClass">
{{form.deviceCode}}
</el-descriptions-item>
<el-descriptions-item label="设备类型" label-class-name="labelClass" content-class-name="contentClass">
{{ util_formatters("deviceType", form.deviceType) }}
</el-descriptions-item>
<el-descriptions-item label="设备的MAC地址" label-class-name="labelClass" content-class-name="contentClass">
{{form.deviceMac}}
</el-descriptions-item>
<el-descriptions-item label="设备访问ip" label-class-name="labelClass" content-class-name="contentClass">
{{form.ip}}
</el-descriptions-item>
<el-descriptions-item label="中心设备编码" label-class-name="labelClass" content-class-name="contentClass">
{{form.centernum}}
</el-descriptions-item>
<el-descriptions-item label="端口" label-class-name="labelClass" content-class-name="contentClass">
{{form.port}}
</el-descriptions-item>
<el-descriptions-item label="站点编号" label-class-name="labelClass" content-class-name="contentClass">
{{form.siteNum}}
</el-descriptions-item>
<el-descriptions-item label="设备生产厂商ID" label-class-name="labelClass" content-class-name="contentClass">
{{form.deviceFirmId}}
</el-descriptions-item>
<el-descriptions-item label="设备生产厂商名称" label-class-name="labelClass" content-class-name="contentClass">
{{form.deviceFirmname}}
</el-descriptions-item>
<el-descriptions-item label="在线状态 " label-class-name="labelClass" content-class-name="contentClass">
{{ util_formatters("deviceOnlineStatus", form.deviceOnlineStatus) }}
</el-descriptions-item>
<el-descriptions-item label="启用状态 " label-class-name="labelClass" content-class-name="contentClass">
{{ util_formatters("status", form.status) }}
</el-descriptions-item>
<el-descriptions-item label="备注" label-class-name="labelClass" content-class-name="contentClass">
{{form.deviceRemark}}
</el-descriptions-item>
<el-descriptions-item label="最近上线时间" label-class-name="labelClass" content-class-name="contentClass">
{{ util_formatterDate(form.onlineTime)}}
</el-descriptions-item>
<el-descriptions-item label="最近离线时间" label-class-name="labelClass" content-class-name="contentClass">
{{ util_formatterDate(form.offlineTime)}}
</el-descriptions-item>
</el-descriptions>
</layout-view>
</template>
<script>
import view from "@/assets/mixins/view";
export default {
mixins: [view],
components: {
},
methods: {
},
data() {
return {
size:"small",
column:2,
toString:[
"deviceType",
"deviceOnlineStatus",
"status",
],
toArrays: [
],
toDate: [
]
}
}
}
</script>
<style lang="less">
.labelClass{
width: 200px;
}
.el-descriptions__body{
margin-left: 5px;
margin-right: 5px;
color: #606266;
background-color: #FFF;
}
.contentClass{
width: 600px;
}
</style>
\ No newline at end of file
device-manager/src/main/java/com/mortals/xhx/ManagerApplication.java
View file @ a3bcf020
package com.mortals.xhx; package com.mortals.xhx;
import com.mortals.framework.service.ICacheService;
import com.mortals.framework.service.impl.LocalCacheServiceImpl;
import com.mortals.framework.service.impl.RedisCacheServiceImpl;
import com.mortals.framework.springcloud.boot.BaseWebApplication; import com.mortals.framework.springcloud.boot.BaseWebApplication;
import org.springframework.boot.SpringApplication; import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.web.servlet.ServletComponentScan; import org.springframework.boot.web.servlet.ServletComponentScan;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ImportResource; import org.springframework.context.annotation.ImportResource;
@SpringBootApplication(scanBasePackages = {"com.mortals"}) @SpringBootApplication(scanBasePackages = {"com.mortals"})
...@@ -16,10 +11,10 @@ import org.springframework.context.annotation.ImportResource; ...@@ -16,10 +11,10 @@ import org.springframework.context.annotation.ImportResource;
@ImportResource(locations = {"classpath:config/spring-config.xml"}) @ImportResource(locations = {"classpath:config/spring-config.xml"})
public class ManagerApplication extends BaseWebApplication { public class ManagerApplication extends BaseWebApplication {
@Bean // @Bean
public ICacheService cacheService() { // public ICacheService cacheService() {
return new LocalCacheServiceImpl(); // return new LocalCacheServiceImpl();
} // }
public static void main(String[] args) { public static void main(String[] args) {
SpringApplication.run(ManagerApplication.class, args); SpringApplication.run(ManagerApplication.class, args);
......
device-manager/src/main/java/com/mortals/xhx/base/system/message/MessageService.java
View file @ a3bcf020
...@@ -18,4 +18,25 @@ public interface MessageService { ...@@ -18,4 +18,25 @@ public interface MessageService {
*/ */
void send(TopicPartitionInfo info, TbQueueMsgHeaders header, String message, TbQueueCallback callback); void send(TopicPartitionInfo info, TbQueueMsgHeaders header, String message, TbQueueCallback callback);
/**
* 请求队列
* @param info
* @param callback
*/
void queueDeclare(TopicPartitionInfo info, TbQueueCallback callback);
/**
* 获取鉴权token
*/
String getBasePlatformToken();
/**
* 获取站点树
* @return
*/
String siteTree();
} }
\ No newline at end of file
device-manager/src/main/java/com/mortals/xhx/base/system/message/impl/MessageServiceImpl.java
View file @ a3bcf020
package com.mortals.xhx.base.system.message.impl; package com.mortals.xhx.base.system.message.impl;
import cn.hutool.core.util.IdUtil; import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mortals.framework.exception.AppException;
import com.mortals.framework.model.Context; import com.mortals.framework.model.Context;
import com.mortals.framework.service.ICacheService;
import com.mortals.framework.service.impl.AbstractCRUDCacheServiceImpl; import com.mortals.framework.service.impl.AbstractCRUDCacheServiceImpl;
import com.mortals.framework.util.DateUtils; import com.mortals.framework.util.DateUtils;
import com.mortals.framework.util.HttpUtil;
import com.mortals.xhx.base.system.message.MessageService; import com.mortals.xhx.base.system.message.MessageService;
import com.mortals.xhx.busiz.rsp.ApiResp; import com.mortals.xhx.busiz.rsp.ApiResp;
import com.mortals.xhx.common.code.ApiRespCodeEnum; import com.mortals.xhx.common.code.ApiRespCodeEnum;
import com.mortals.xhx.common.code.YesNoEnum;
import com.mortals.xhx.common.key.Constant;
import com.mortals.xhx.common.model.DefaultTbQueueMsgHeaders; import com.mortals.xhx.common.model.DefaultTbQueueMsgHeaders;
import com.mortals.xhx.common.model.MessageHeader; import com.mortals.xhx.common.model.MessageHeader;
import com.mortals.xhx.module.device.dao.DeviceDao; import com.mortals.xhx.module.device.dao.DeviceDao;
...@@ -15,11 +22,20 @@ import com.mortals.xhx.module.device.service.DeviceService; ...@@ -15,11 +22,20 @@ import com.mortals.xhx.module.device.service.DeviceService;
import com.mortals.xhx.queue.*; import com.mortals.xhx.queue.*;
import com.mortals.xhx.queue.provider.TbCoreQueueProducerProvider; import com.mortals.xhx.queue.provider.TbCoreQueueProducerProvider;
import lombok.extern.apachecommons.CommonsLog; import lombok.extern.apachecommons.CommonsLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.entity.ContentType;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map;
import static com.mortals.framework.util.HttpUtil.HEADER_CONTENT_TYPE;
import static com.mortals.xhx.common.key.Constant.PATH_LOGIN;
import static com.mortals.xhx.common.key.Constant.PATH_SITETREE;
/** /**
* DeviceService * DeviceService
...@@ -29,9 +45,19 @@ import java.util.List; ...@@ -29,9 +45,19 @@ import java.util.List;
* @date 2022-03-09 * @date 2022-03-09
*/ */
@Service("messageService") @Service("messageService")
@CommonsLog @Slf4j
public class MessageServiceImpl implements MessageService { public class MessageServiceImpl implements MessageService {
@Value("${baseplatform.httpUrl:''}")
private String httpUrl;
@Value("${baseplatform.loginName:''}")
private String loginName;
@Value("${baseplatform.password:''}")
private String password;
@Autowired
private ICacheService cacheService;
@Autowired @Autowired
private TbCoreQueueProducerProvider producerProvider; private TbCoreQueueProducerProvider producerProvider;
...@@ -40,6 +66,75 @@ public class MessageServiceImpl implements MessageService { ...@@ -40,6 +66,75 @@ public class MessageServiceImpl implements MessageService {
TbQueueProducer<TbQueueMsg> producer = producerProvider.getTbCoreMsgProducer(); TbQueueProducer<TbQueueMsg> producer = producerProvider.getTbCoreMsgProducer();
TbQueueMsg queueMsg = new DefaultTbQueueMsg(IdUtil.fastUUID(), message == null ? "".getBytes() : message.getBytes(), header); TbQueueMsg queueMsg = new DefaultTbQueueMsg(IdUtil.fastUUID(), message == null ? "".getBytes() : message.getBytes(), header);
producer.send(info, queueMsg, callback); producer.send(info, queueMsg, callback);
}
@Override
public void queueDeclare(TopicPartitionInfo info, TbQueueCallback callback) {
TbQueueProducer<TbQueueMsg> producer = producerProvider.getTbCoreMsgProducer();
producer.queueDeclare(info, callback);
}
@Override
public String getBasePlatformToken() {
String token = cacheService.get(Constant.BASEPLATFORM_AUTHTOKEN);
if (ObjectUtils.isEmpty(token)) {
JSONObject obj = new JSONObject();
obj.put("loginName", loginName);
obj.put("password", password);
obj.put("securityCode", "8888");
String resp = null;
try {
Map<String, String> header = new HashMap<>();
header.put(HEADER_CONTENT_TYPE, "application/json");
resp = HttpUtil.doPost(httpUrl + PATH_LOGIN, header, JSON.toJSONString(obj));
JSONObject jsonObject = JSON.parseObject(resp);
Integer code = jsonObject.getInteger("code");
if (code == YesNoEnum.YES.getValue()) {
JSONObject dataObj = jsonObject.getJSONObject("data");
String authtoken = dataObj.getString("token");
cacheService.setnx(Constant.BASEPLATFORM_AUTHTOKEN,authtoken,7*24*60*60*1000);
return authtoken;
} else {
throw new AppException("异常");
}
} catch (Exception e) {
log.error("异常:", e);
}
log.info("resp:{}", resp);
}
return token;
}
@Override
public String siteTree() {
String authToken = this.getBasePlatformToken();
String resp = null;
try {
Map<String, String> header = new HashMap<>();
header.put(HEADER_CONTENT_TYPE, "application/json");
header.put("Authorization", Constant.TOKEN_PREFIX + authToken);
resp = HttpUtil.doGet(httpUrl + PATH_SITETREE, header,new HashMap<>());
JSONObject jsonObject = JSON.parseObject(resp);
Integer code = jsonObject.getInteger("code");
if (code == YesNoEnum.YES.getValue()) {
JSONObject dataObj = jsonObject.getJSONObject("data");
return resp;
} else {
throw new AppException("异常");
}
} catch (Exception e) {
log.error("异常:", e);
}
log.info("resp:{}", resp);
return resp;
} }
} }
\ No newline at end of file
device-manager/src/main/java/com/mortals/xhx/busiz/web/DeviceApiController.java
View file @ a3bcf020
...@@ -111,15 +111,16 @@ public class DeviceApiController { ...@@ -111,15 +111,16 @@ public class DeviceApiController {
if (ObjectUtils.isEmpty(platformEntity)) { if (ObjectUtils.isEmpty(platformEntity)) {
throw new AppException("当前设备未配置所属系统平台，请在后台配置后再激活！"); throw new AppException("当前设备未配置所属系统平台，请在后台配置后再激活！");
} }
// authInfo.setHost(platformEntity.getPlatformSn()); // authInfo.setHost(platformEntity.getPlatformSn());
ProductEntity productEntity = productService.get(deviceEntity.getProductId()); ProductEntity productEntity = productService.get(deviceEntity.getProductId());
if (ObjectUtils.isEmpty(productEntity)) { if (ObjectUtils.isEmpty(productEntity)) {
throw new AppException("当前设备未配置所属产品，请在后台配置后再激活！"); throw new AppException("当前设备未配置所属产品，请在后台配置后再激活！");
} }
authInfo.setExchangeName(platformEntity.getPlatformSn()+Constant.EXCHANGE_SPLIT+productEntity.getProductCode()); authInfo.setExchangeName(platformEntity.getPlatformSn() + Constant.EXCHANGE_SPLIT + productEntity.getProductCode());
authInfo.setUploadTopicFilter(Constant.UPLOAD_TOPIC + deviceEntity.getDeviceCode()); authInfo.setUploadTopicFilter(Constant.UPLOAD_TOPIC + deviceEntity.getDeviceCode());
authInfo.setDownTopicFilter(Constant.DOWN_TOPIC + deviceEntity.getDeviceCode()); authInfo.setDownTopicFilter(Constant.DOWN_TOPIC + deviceEntity.getDeviceCode());
String token = AESUtil.encryptForApp(JSON.toJSONString(authInfo), secret); String token = JSON.toJSONString(authInfo);
//String token = AESUtil.encryptForApp(JSON.toJSONString(authInfo), secret);
deviceResp.setToken(token); deviceResp.setToken(token);
} }
deviceEntity.setOnlineTime(new Date()); deviceEntity.setOnlineTime(new Date());
...@@ -174,8 +175,11 @@ public class DeviceApiController { ...@@ -174,8 +175,11 @@ public class DeviceApiController {
TopicPartitionInfo info = TopicPartitionInfo.builder().exchangeName(exchangeName).topic(Constant.UPLOAD_TOPIC + deviceEntity.getDeviceMac()).build(); TopicPartitionInfo info = TopicPartitionInfo.builder().exchangeName(exchangeName).topic(Constant.UPLOAD_TOPIC + deviceEntity.getDeviceMac()).build();
TbQueueMsgHeaders header = new DefaultTbQueueMsgHeaders(); TbQueueMsgHeaders header = new DefaultTbQueueMsgHeaders();
header.put(MessageHeader.MESSAGETYPE, Constant.MESSAGETYPE_HEARTBEAT); header.put(MessageHeader.MESSAGETYPE, Constant.MESSAGETYPE_HEARTBEAT);
ApiResp<String> sendDeviceMessageResp = deviceService.sendDeviceMessage(deviceEntity, info, header, JSON.toJSONString(req), null); for (int i = 0; i < 1000; i++) {
log.info(String.format("sendMsgResp:%s", JSON.toJSONString(sendDeviceMessageResp))); deviceService.sendDeviceMessage(deviceEntity, info, header, JSON.toJSONString(req), null);
// Thread.sleep(50);
}
//log.info(String.format("sendMsgResp:%s", JSON.toJSONString(sendDeviceMessageResp)));
} }
} catch (Exception e) { } catch (Exception e) {
log.error("接收数据失败", e); log.error("接收数据失败", e);
......
device-manager/src/main/java/com/mortals/xhx/common/key/Constant.java
View file @ a3bcf020
...@@ -42,6 +42,11 @@ public final class Constant { ...@@ -42,6 +42,11 @@ public final class Constant {
*/ */
public static final String TOKEN = "token"; public static final String TOKEN = "token";
/**
* 基础服务平台鉴权token
*/
public static final String BASEPLATFORM_AUTHTOKEN = "baseplatform_authtoken";
/** /**
* 令牌前缀 * 令牌前缀
*/ */
...@@ -81,4 +86,17 @@ public final class Constant { ...@@ -81,4 +86,17 @@ public final class Constant {
*/ */
public static final String MESSAGETYPE_UPGREAD = "UPGREAD"; public static final String MESSAGETYPE_UPGREAD = "UPGREAD";
/**
* 登录path
*/
public static final String PATH_LOGIN = "login/login";
/**
* 树path
*/
public static final String PATH_SITETREE = "site/siteTree";
} }
device-manager/src/main/java/com/mortals/xhx/common/utils/GateProtConfig.java
View file @ a3bcf020
...@@ -9,12 +9,8 @@ import org.dom4j.Element; ...@@ -9,12 +9,8 @@ import org.dom4j.Element;
@Data @Data
public abstract class GateProtConfig { public abstract class GateProtConfig {
/**提交URL*/
/** private String sendUrl;
* 协议ID
*/
private Integer protocol;
/** /**
* 流量数 * 流量数
*/ */
...@@ -30,9 +26,14 @@ public abstract class GateProtConfig { ...@@ -30,9 +26,14 @@ public abstract class GateProtConfig {
*/ */
private String paramConfig; private String paramConfig;
/**
* 发送延迟
*/
private Integer SendDelayTime = 1 * 1000; private Integer SendDelayTime = 1 * 1000;
/**
*
*/
private Integer maxDelayNum = 50; private Integer maxDelayNum = 50;
......
device-manager/src/main/java/com/mortals/xhx/common/utils/SendTask.java
View file @ a3bcf020
package com.mortals.xhx.common.utils; package com.mortals.xhx.common.utils;
import com.mortals.framework.util.HttpUtil;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils; import org.springframework.beans.BeanUtils;
...@@ -11,17 +13,18 @@ import org.springframework.beans.BeanUtils; ...@@ -11,17 +13,18 @@ import org.springframework.beans.BeanUtils;
* @description: * @description:
**/ **/
@Slf4j @Slf4j
@AllArgsConstructor
public class SendTask implements Runnable { public class SendTask implements Runnable {
private String sendUrl;
private String content;
@Override @Override
public void run() { public void run() {
// TODO: 2022/4/28 // TODO: 2022/4/28
try { try {
// log.debug("启动发送"+smsGateQueueEntity); // log.debug("启动发送"+smsGateQueueEntity);
String resp = HttpUtil.doPost(sendUrl, content);
log.debug("http resp:{}", resp);
} catch (Exception e) { } catch (Exception e) {
log.error("发送异常：" + e); log.error("发送异常：" + e);
} }
......
device-manager/src/main/java/com/mortals/xhx/daemon/DemoApiController.java
View file @ a3bcf020
package com.mortals.xhx.daemon; package com.mortals.xhx.daemon;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.JSONObject;
import com.mortals.framework.ap.GlobalSysInfo;
import com.mortals.framework.exception.AppException; import com.mortals.framework.exception.AppException;
import com.mortals.framework.model.Context;
import com.mortals.framework.model.PageInfo;
import com.mortals.framework.model.Result;
import com.mortals.framework.util.DateUtils; import com.mortals.framework.util.DateUtils;
import com.mortals.xhx.base.framework.config.InterceptorConfig; import com.mortals.xhx.base.system.message.MessageService;
import com.mortals.xhx.base.system.upload.service.UploadService; import com.mortals.xhx.busiz.req.UploadDeviceReq;
import com.mortals.xhx.base.system.user.model.UserEntity;
import com.mortals.xhx.busiz.rsp.ApiResp; import com.mortals.xhx.busiz.rsp.ApiResp;
import com.mortals.xhx.common.code.ApiRespCodeEnum; import com.mortals.xhx.common.code.ApiRespCodeEnum;
import com.mortals.xhx.common.code.OneThingRespCodeEnum; import com.mortals.xhx.common.code.OneThingRespCodeEnum;
import com.mortals.xhx.common.key.Constant;
import com.mortals.xhx.common.model.DefaultTbQueueMsgHeaders; import com.mortals.xhx.common.model.DefaultTbQueueMsgHeaders;
import com.mortals.xhx.common.model.MessageHeader; import com.mortals.xhx.common.model.MessageHeader;
import com.mortals.xhx.queue.*; import com.mortals.xhx.module.device.model.DeviceEntity;
import com.mortals.xhx.module.device.service.DeviceService;
import com.mortals.xhx.module.platform.model.PlatformEntity;
import com.mortals.xhx.module.platform.service.PlatformService;
import com.mortals.xhx.module.product.model.ProductEntity;
import com.mortals.xhx.module.product.service.ProductService;
import com.mortals.xhx.queue.TbQueueMsg;
import com.mortals.xhx.queue.TbQueueMsgHeaders;
import com.mortals.xhx.queue.TbQueueProducer;
import com.mortals.xhx.queue.TopicPartitionInfo;
import com.mortals.xhx.queue.provider.TbCoreQueueProducerProvider; import com.mortals.xhx.queue.provider.TbCoreQueueProducerProvider;
import lombok.extern.apachecommons.CommonsLog; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import java.util.HashMap;
import java.io.File; import java.util.Map;
import java.io.FileOutputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
/** /**
* 一件事微信对外服务接口 * 一件事微信对外服务接口
...@@ -46,21 +39,39 @@ import java.util.stream.IntStream; ...@@ -46,21 +39,39 @@ import java.util.stream.IntStream;
*/ */
@CrossOrigin(origins = "*") @CrossOrigin(origins = "*")
@RestController @RestController
@CommonsLog @Slf4j
@RequestMapping("/test/") @RequestMapping("/test/")
public class DemoApiController { public class DemoApiController {
@Autowired @Autowired
private TbCoreQueueProducerProvider producerProvider; private TbCoreQueueProducerProvider producerProvider;
@Autowired
private DeviceService deviceService;
@Autowired
private ProductService productService;
@Autowired
private PlatformService platformService;
@Autowired
private MessageService messageService;
@PostMapping("/sentry")
@GetMapping("/sentry") public String sentry(@RequestBody String body) {
public String sentry() { log.info("body==>{}",body);
log.error("[main][我就是展示下异常！]");
return "success"; return "success";
} }
@PostMapping("/getToken")
public String getToken() {
return messageService.getBasePlatformToken();
}
@PostMapping("/siteTree")
public String siteTree() {
return messageService.siteTree();
}
@GetMapping("/exception") @GetMapping("/exception")
public String exception() { public String exception() {
...@@ -122,4 +133,56 @@ public class DemoApiController { ...@@ -122,4 +133,56 @@ public class DemoApiController {
return JSON.toJSONString(rsp); return JSON.toJSONString(rsp);
} }
/**
* 设备数据上报
*
* @param
* @return
*/
@PostMapping("upload")
public String upload(@RequestParam(value = "deviceId") Long deviceId ) {
ApiResp<String> rsp = new ApiResp<>();
rsp.setMsg(ApiRespCodeEnum.SUCCESS.getLabel());
rsp.setCode(ApiRespCodeEnum.SUCCESS.getValue());
try {
//根据设备编码查询设备
DeviceEntity deviceEntity = deviceService.get(deviceId);
if (!ObjectUtils.isEmpty(deviceEntity)) {
//将上报信息转发到mq中
PlatformEntity platformEntity = platformService.get(deviceEntity.getPlatformId());
if (ObjectUtils.isEmpty(platformEntity)) {
throw new AppException("当前设备未配置所属系统平台，请在后台配置后再激活！");
}
// authInfo.setHost(platformEntity.getPlatformSn());
ProductEntity productEntity = productService.get(deviceEntity.getProductId());
if (ObjectUtils.isEmpty(productEntity)) {
throw new AppException("当前设备未配置所属产品，请在后台配置后再激活！");
}
String exchangeName = platformEntity.getPlatformSn() + Constant.EXCHANGE_SPLIT + productEntity.getProductCode();
TopicPartitionInfo info = TopicPartitionInfo.builder().exchangeName(exchangeName).topic(Constant.UPLOAD_TOPIC + deviceEntity.getDeviceMac()).build();
TbQueueMsgHeaders header = new DefaultTbQueueMsgHeaders();
header.put(MessageHeader.MESSAGETYPE, Constant.MESSAGETYPE_HEARTBEAT);
UploadDeviceReq uploadDeviceReq = new UploadDeviceReq();
uploadDeviceReq.setDeviceMac(deviceEntity.getDeviceMac());
uploadDeviceReq.setDeviceNum(deviceEntity.getDeviceCode());
for (int i = 0; i < 100; i++) {
deviceService.sendDeviceMessage(deviceEntity, info, header, JSON.toJSONString(deviceEntity), null);
}
}
} catch (Exception e) {
log.error("接收数据失败", e);
rsp.setCode(ApiRespCodeEnum.FAILED.getValue());
rsp.setMsg(e.getMessage());
return JSON.toJSONString(rsp);
}
log.debug("响应【设备数据上报】【响应体】--> " + JSONObject.toJSONString(rsp));
return JSON.toJSONString(rsp);
}
} }
device-manager/src/main/java/com/mortals/xhx/daemon/applicationservice/DeviceMsgComsumerStartedService.java
View file @ a3bcf020
...@@ -17,29 +17,31 @@ import com.mortals.xhx.module.device.service.DeviceLogService; ...@@ -17,29 +17,31 @@ import com.mortals.xhx.module.device.service.DeviceLogService;
import com.mortals.xhx.module.device.service.DeviceService; import com.mortals.xhx.module.device.service.DeviceService;
import com.mortals.xhx.module.platform.model.PlatformEntity; import com.mortals.xhx.module.platform.model.PlatformEntity;
import com.mortals.xhx.module.platform.service.PlatformService; import com.mortals.xhx.module.platform.service.PlatformService;
import com.mortals.xhx.module.product.model.ProductEntity;
import com.mortals.xhx.module.product.service.ProductService; import com.mortals.xhx.module.product.service.ProductService;
import com.mortals.xhx.queue.DefaultTbCoreConsumerService; import com.mortals.xhx.queue.DefaultTbCoreConsumerService;
import com.mortals.xhx.queue.TbQueueConsumer; import com.mortals.xhx.queue.TbQueueConsumer;
import com.mortals.xhx.queue.TbQueueMsg; import com.mortals.xhx.queue.TbQueueMsg;
import com.mortals.xhx.queue.TopicPartitionInfo; import com.mortals.xhx.queue.TopicPartitionInfo;
import com.mortals.xhx.utils.IotThreadFactory; import com.mortals.xhx.utils.IotThreadFactory;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import lombok.extern.apachecommons.CommonsLog; import lombok.extern.apachecommons.CommonsLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import java.util.Date; import java.util.*;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@Component @Component
@CommonsLog @Slf4j
public class DeviceMsgComsumerStartedService implements IApplicationStartedService { public class DeviceMsgComsumerStartedService implements IApplicationStartedService {
@Autowired @Autowired
...@@ -67,77 +69,42 @@ public class DeviceMsgComsumerStartedService implements IApplicationStartedServi ...@@ -67,77 +69,42 @@ public class DeviceMsgComsumerStartedService implements IApplicationStartedServi
if (!ObjectUtils.isEmpty(mainConsumer)) { if (!ObjectUtils.isEmpty(mainConsumer)) {
//订阅所有已几快活设备 //订阅所有已几快活设备
Set<TopicPartitionInfo> topicPartitionInfoSet = deviceService.find(new DeviceQuery().active(ActiveEnum.已激活.getValue()).status(StatusEnum.启用.getValue())).stream() Set<TopicPartitionInfo> topicPartitionInfoSet = deviceService.find(new DeviceQuery().active(ActiveEnum.已激活.getValue()).status(StatusEnum.启用.getValue())).stream()
.map(item -> .map(item -> {
new TopicPartitionInfo(Constant.UPLOAD_TOPIC + item.getDeviceMac(), null, null) PlatformEntity platformEntity = platformService.get(item.getPlatformId());
ProductEntity productEntity = productService.get(item.getProductId());
String exchangeName = platformEntity.getPlatformSn() + Constant.EXCHANGE_SPLIT + productEntity.getProductCode();
return new TopicPartitionInfo(Constant.UPLOAD_TOPIC + item.getDeviceMac(), null, exchangeName);
}
).collect(Collectors.toSet()); ).collect(Collectors.toSet());
mainConsumer.subscribe(topicPartitionInfoSet); mainConsumer.subscribe(topicPartitionInfoSet);
log.debug("消费线程订阅设备上报消息成功！：" + JSON.toJSONString(topicPartitionInfoSet)); log.info("消费线程订阅topic！：{}", JSON.toJSONString(topicPartitionInfoSet));
this.consumersExecutor = Executors.newCachedThreadPool(IotThreadFactory.forName("消费queue线程")); this.consumersExecutor = Executors.newCachedThreadPool(IotThreadFactory.forName("消费queue线程"));
consumersExecutor.submit(() -> { consumersExecutor.submit(() -> {
int waitTime = 1000;
while (!stopped) { while (!stopped) {
try { try {
//todo 批量读取
TbQueueMsg queueMsg = consumerService.getComsureQueue().poll();
if (!ObjectUtils.isEmpty(queueMsg)) {
//做相应业务,做日志操作
DeviceReq deviceReq = JSON.parseObject(new String(queueMsg.getData()), DeviceReq.class);
boolean bool = false;
DeviceEntity deviceEntity = deviceService.getExtCache(deviceReq.getDeviceNum());
if (!ObjectUtils.isEmpty(deviceEntity)) {
if (deviceEntity.getDeviceOnlineStatus() == DeviceOnlineStatusEnum.离线.getValue()) {
bool = true;
}
deviceEntity.setOnlineTime(new Date());
deviceEntity.setDeviceOnlineStatus(DeviceOnlineStatusEnum.在线.getValue());
deviceEntity.setIp(deviceReq.getIp());
deviceEntity.setPort(deviceReq.getPort());
deviceEntity.setCenternum(deviceReq.getCenternum());
deviceService.update(deviceEntity);
DeviceLogEntity deviceLogEntity = new DeviceLogEntity();
deviceLogEntity.initAttrValue();
deviceLogEntity.setDeviceId(deviceEntity.getId());
deviceLogEntity.setDeviceName(deviceEntity.getDeviceName());
deviceLogEntity.setDeviceNum(deviceEntity.getDeviceCode());
deviceLogEntity.setContent(JSONObject.toJSONString(deviceReq));
deviceLogEntity.setCreateTime(new Date());
deviceLogService.save(deviceLogEntity);
//查询消息是否归属了第三方平台 如果有则发送消息
//获取exchange,
PlatformEntity platformEntity = platformService.get(deviceEntity.getProductId());
if(!ObjectUtils.isEmpty(platformEntity)&&platformEntity.getSendSwitch()== YesNoEnum.YES.getValue()){
if(platformEntity.getSendMsgType()== SendMsgTypeEnum.http.getValue()){
//http方式
//通过线程池进行发送消息
SendTask sendTask = new SendTask();
sendTaskThreadPool.execute(sendTask);
}
}
//deviceEntity.getPlatformId()
//queueMsg.getHeaders()
if (bool) { List<TbQueueMsg> messageList = new ArrayList(32);
WebSocketUtil.broadcast(SendToAllRequest.TYPE, new SendToAllRequest().setContent(JSON.toJSONString(deviceEntity))); TbQueueMsg message;
} for (int i = 0; i < 32; i++) {
if (i == 0) {
message = consumerService.getComsureQueue().poll(waitTime, TimeUnit.MILLISECONDS);
} else {
message = consumerService.getComsureQueue().poll();
}
if (Objects.nonNull(message)) {
messageList.add(message);
} else {
break;
} }
}
if (messageList.size() > 0) {
log.debug("poll messageQueue messages: {}", JSON.toJSONString(messageList));
//异步消息分发
AsyncDispatcher dispatcher = new AsyncDispatcher(messageList);
sendTaskThreadPool.execute(dispatcher);
} }
} catch (Exception e) { } catch (Exception e) {
...@@ -149,7 +116,6 @@ public class DeviceMsgComsumerStartedService implements IApplicationStartedServi ...@@ -149,7 +116,6 @@ public class DeviceMsgComsumerStartedService implements IApplicationStartedServi
} }
} }
} }
mainConsumer.commit(); mainConsumer.commit();
}); });
...@@ -170,4 +136,68 @@ public class DeviceMsgComsumerStartedService implements IApplicationStartedServi ...@@ -170,4 +136,68 @@ public class DeviceMsgComsumerStartedService implements IApplicationStartedServi
return 10; return 10;
} }
class AsyncDispatcher implements Runnable {
private List<TbQueueMsg> messages;
AsyncDispatcher(List<TbQueueMsg> messages) {
this.messages = messages;
}
@Override
public void run() {
if (Objects.nonNull(messages)) {
try {
for (TbQueueMsg queueMsg : messages) {
log.debug("[WRITE MESSAGE] ->message:{}", JSON.toJSONString(queueMsg));
//做相应业务,做日志操作
DeviceReq deviceReq = JSON.parseObject(new String(queueMsg.getData()), DeviceReq.class);
boolean bool = false;
DeviceEntity deviceEntity = deviceService.getExtCache(deviceReq.getDeviceNum());
if (!ObjectUtils.isEmpty(deviceEntity)) {
if (deviceEntity.getDeviceOnlineStatus() == DeviceOnlineStatusEnum.离线.getValue()) {
bool = true;
}
deviceEntity.setOnlineTime(new Date());
deviceEntity.setDeviceOnlineStatus(DeviceOnlineStatusEnum.在线.getValue());
deviceEntity.setIp(deviceReq.getIp());
deviceEntity.setPort(deviceReq.getPort());
deviceEntity.setCenternum(deviceReq.getCenternum());
deviceService.update(deviceEntity);
DeviceLogEntity deviceLogEntity = new DeviceLogEntity();
deviceLogEntity.initAttrValue();
deviceLogEntity.setDeviceId(deviceEntity.getId());
deviceLogEntity.setDeviceName(deviceEntity.getDeviceName());
deviceLogEntity.setDeviceNum(deviceEntity.getDeviceCode());
deviceLogEntity.setContent(JSONObject.toJSONString(deviceReq));
deviceLogEntity.setLogType(LogTypeEnum.上报事件.getValue());
deviceLogEntity.setCreateUserId(1L);
deviceLogEntity.setCreateTime(new Date());
deviceLogService.save(deviceLogEntity);
//获取exchange,
PlatformEntity platformEntity = platformService.get(deviceEntity.getPlatformId());
if (!ObjectUtils.isEmpty(platformEntity) && platformEntity.getSendSwitch() == YesNoEnum.YES.getValue()) {
if (platformEntity.getSendMsgType() == SendMsgTypeEnum.http.getValue() && !ObjectUtils.isEmpty(platformEntity.getSendConfig())) {
//http方式
//通过线程池进行发送消息
SendTask sendTask = new SendTask(platformEntity.getSendConfig(), new String(queueMsg.getData()));
sendTaskThreadPool.execute(sendTask);
}
}
// if (bool) {
// WebSocketUtil.broadcast(SendToAllRequest.TYPE, new SendToAllRequest().setContent(JSON.toJSONString(deviceEntity)));
// }
}
}
} catch (Exception ex) {
log.error("异常", ex);
}
}
}
}
} }
device-manager/src/main/java/com/mortals/xhx/daemon/task/DeviceStatTaskImpl.java
View file @ a3bcf020
...@@ -48,10 +48,10 @@ public class DeviceStatTaskImpl implements ITaskExcuteService { ...@@ -48,10 +48,10 @@ public class DeviceStatTaskImpl implements ITaskExcuteService {
@Override @Override
public void excuteTask(ITask task) throws AppException { public void excuteTask(ITask task) throws AppException {
log.info("设备状态统计，开始执行"); log.debug("设备状态统计，开始执行");
doDeviceUpOrDown(); doDeviceUpOrDown();
doDeviceLogDel(); doDeviceLogDel();
log.info("设备状态统计，结束执行"); log.debug("设备状态统计，结束执行");
} }
/** /**
...@@ -81,7 +81,7 @@ public class DeviceStatTaskImpl implements ITaskExcuteService { ...@@ -81,7 +81,7 @@ public class DeviceStatTaskImpl implements ITaskExcuteService {
} catch (Exception e) { } catch (Exception e) {
log.error("更新设备状态任务异常，结束执行", e); log.error("更新设备状态任务异常，结束执行", e);
} }
log.info("更新设备状态任务，结束执行"); log.debug("更新设备状态任务，结束执行");
} }
...@@ -101,7 +101,7 @@ public class DeviceStatTaskImpl implements ITaskExcuteService { ...@@ -101,7 +101,7 @@ public class DeviceStatTaskImpl implements ITaskExcuteService {
} catch (Exception e) { } catch (Exception e) {
log.error("设备日志删除任务异常，结束执行", e); log.error("设备日志删除任务异常，结束执行", e);
} }
log.info("设备日志删除任务，结束执行"); log.debug("设备日志删除任务，结束执行");
} }
......
device-manager/src/main/java/com/mortals/xhx/module/device/service/DeviceService.java
View file @ a3bcf020
...@@ -21,11 +21,10 @@ public interface DeviceService extends ICRUDCacheService<DeviceEntity,Long>{ ...@@ -21,11 +21,10 @@ public interface DeviceService extends ICRUDCacheService<DeviceEntity,Long>{
ApiResp<String> sendDeviceMessage(DeviceEntity deviceEntity, TopicPartitionInfo info, TbQueueMsgHeaders header,String message , Context context); ApiResp<String> sendDeviceMessage(DeviceEntity deviceEntity, TopicPartitionInfo info, TbQueueMsgHeaders header,String message , Context context);
ApiResp<String> sendDeviceMessage(List<Long> deviceIds, TopicPartitionInfo info,TbQueueMsgHeaders header, String message , Context context);
ApiResp<String> sendDeviceMessage(List<Long> deviceIds, TopicPartitionInfo info,TbQueueMsgHeaders header, String message , Context context);
} }
\ No newline at end of file
device-manager/src/main/java/com/mortals/xhx/module/device/service/impl/DeviceServiceImpl.java
View file @ a3bcf020
package com.mortals.xhx.module.device.service.impl; package com.mortals.xhx.module.device.service.impl;
import cn.hutool.core.util.IdUtil; import com.mortals.framework.exception.AppException;
import com.mortals.framework.model.Context; import com.mortals.framework.model.Context;
import com.mortals.framework.util.DateUtils; import com.mortals.framework.service.impl.AbstractCRUDCacheServiceImpl;
import com.mortals.xhx.base.system.message.MessageService; import com.mortals.xhx.base.system.message.MessageService;
import com.mortals.xhx.busiz.rsp.ApiResp; import com.mortals.xhx.busiz.rsp.ApiResp;
import com.mortals.xhx.common.code.ApiRespCodeEnum; import com.mortals.xhx.common.code.ApiRespCodeEnum;
import com.mortals.xhx.common.code.MessageProtocolEnum;
import com.mortals.xhx.common.key.Constant; import com.mortals.xhx.common.key.Constant;
import com.mortals.xhx.common.model.DefaultTbQueueMsgHeaders; import com.mortals.xhx.module.device.dao.DeviceDao;
import com.mortals.xhx.common.model.MessageHeader; import com.mortals.xhx.module.device.model.DeviceEntity;
import com.mortals.xhx.module.device.model.DeviceQuery; import com.mortals.xhx.module.device.model.DeviceQuery;
import com.mortals.xhx.module.device.service.DeviceService;
import com.mortals.xhx.module.platform.model.PlatformEntity;
import com.mortals.xhx.module.platform.service.PlatformService;
import com.mortals.xhx.module.product.model.ProductEntity;
import com.mortals.xhx.module.product.service.ProductService;
import com.mortals.xhx.queue.*; import com.mortals.xhx.queue.*;
import com.mortals.xhx.queue.provider.TbCoreQueueProducerProvider; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import com.mortals.framework.service.impl.AbstractCRUDCacheServiceImpl;
import com.mortals.xhx.module.device.dao.DeviceDao;
import com.mortals.xhx.module.device.model.DeviceEntity;
import com.mortals.xhx.module.device.service.DeviceService;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
import java.util.HashMap; import java.util.HashSet;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Set;
import java.util.UUID;
/** /**
* DeviceService * DeviceService
...@@ -34,8 +33,16 @@ import java.util.UUID; ...@@ -34,8 +33,16 @@ import java.util.UUID;
* @date 2022-03-09 * @date 2022-03-09
*/ */
@Service("deviceService") @Service("deviceService")
@Slf4j
public class DeviceServiceImpl extends AbstractCRUDCacheServiceImpl<DeviceDao, DeviceEntity, Long> implements DeviceService { public class DeviceServiceImpl extends AbstractCRUDCacheServiceImpl<DeviceDao, DeviceEntity, Long> implements DeviceService {
@Autowired
private ProductService productService;
@Autowired
private PlatformService platformService;
@Autowired
private DefaultTbCoreConsumerService consumerService;
@Override @Override
protected String getExtKey(DeviceEntity data) { protected String getExtKey(DeviceEntity data) {
return data.getDeviceCode(); return data.getDeviceCode();
...@@ -49,7 +56,6 @@ public class DeviceServiceImpl extends AbstractCRUDCacheServiceImpl<DeviceDao, D ...@@ -49,7 +56,6 @@ public class DeviceServiceImpl extends AbstractCRUDCacheServiceImpl<DeviceDao, D
ApiResp<String> resp = new ApiResp<>(); ApiResp<String> resp = new ApiResp<>();
resp.setCode(ApiRespCodeEnum.SUCCESS.getValue()); resp.setCode(ApiRespCodeEnum.SUCCESS.getValue());
resp.setMsg(ApiRespCodeEnum.SUCCESS.getLabel()); resp.setMsg(ApiRespCodeEnum.SUCCESS.getLabel());
TbQueueCallback callback = new TbQueueCallback() { TbQueueCallback callback = new TbQueueCallback() {
@Override @Override
public void onSuccess(TbQueueMsgMetadata metadata) { public void onSuccess(TbQueueMsgMetadata metadata) {
...@@ -63,7 +69,6 @@ public class DeviceServiceImpl extends AbstractCRUDCacheServiceImpl<DeviceDao, D ...@@ -63,7 +69,6 @@ public class DeviceServiceImpl extends AbstractCRUDCacheServiceImpl<DeviceDao, D
} }
}; };
messageService.send(info, header, message, callback); messageService.send(info, header, message, callback);
return resp; return resp;
} }
...@@ -80,4 +85,35 @@ public class DeviceServiceImpl extends AbstractCRUDCacheServiceImpl<DeviceDao, D ...@@ -80,4 +85,35 @@ public class DeviceServiceImpl extends AbstractCRUDCacheServiceImpl<DeviceDao, D
} }
@Override
protected void saveAfter(DeviceEntity entity, Context context) throws AppException {
PlatformEntity platformEntity = platformService.get(entity.getPlatformId());
ProductEntity productEntity = productService.get(entity.getProductId());
if (!ObjectUtils.isEmpty(platformEntity) && ObjectUtils.isEmpty(productEntity)) {
//注册rabbmit相关队列与绑定
String exchangeName = platformEntity.getPlatformSn() + Constant.EXCHANGE_SPLIT + productEntity.getProductCode();
TopicPartitionInfo info = TopicPartitionInfo.builder().exchangeName(exchangeName).topic(Constant.UPLOAD_TOPIC + entity.getDeviceCode()).build();
TbQueueCallback callback = new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {
log.info("队列创建成功,设备编码:{}", entity.getDeviceCode());
}
@Override
public void onFailure(Throwable t) {
log.error("队列创建失败,设备通道编码:{},{}", entity.getDeviceCode(), t);
}
};
messageService.queueDeclare(info, callback);
//消费监听线程添加当前声明队列
Set<TopicPartitionInfo> partitions = new HashSet<>();
partitions.add(info);
consumerService.getMainConsumer().subscribe(partitions);
}
super.saveAfter(entity, context);
}
} }
\ No newline at end of file
device-manager/src/main/java/com/mortals/xhx/module/device/web/DeviceController.java
View file @ a3bcf020
...@@ -3,8 +3,10 @@ package com.mortals.xhx.module.device.web; ...@@ -3,8 +3,10 @@ package com.mortals.xhx.module.device.web;
import cn.hutool.crypto.SecureUtil; import cn.hutool.crypto.SecureUtil;
import cn.hutool.crypto.asymmetric.SignAlgorithm; import cn.hutool.crypto.asymmetric.SignAlgorithm;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.mortals.framework.exception.AppException; import com.mortals.framework.exception.AppException;
import com.mortals.framework.model.OrderCol; import com.mortals.framework.model.OrderCol;
import com.mortals.xhx.base.system.message.MessageService;
import com.mortals.xhx.base.system.param.service.ParamService; import com.mortals.xhx.base.system.param.service.ParamService;
import com.mortals.xhx.busiz.rsp.ApiResp; import com.mortals.xhx.busiz.rsp.ApiResp;
import com.mortals.xhx.common.code.ActiveEnum; import com.mortals.xhx.common.code.ActiveEnum;
...@@ -66,6 +68,8 @@ public class DeviceController extends BaseCRUDJsonMappingController<DeviceServic ...@@ -66,6 +68,8 @@ public class DeviceController extends BaseCRUDJsonMappingController<DeviceServic
private ProductService productService; private ProductService productService;
@Autowired @Autowired
private PlatformService platformService; private PlatformService platformService;
@Autowired
private MessageService messageService;
public DeviceController() { public DeviceController() {
super.setFormClass(DeviceForm.class); super.setFormClass(DeviceForm.class);
...@@ -99,12 +103,10 @@ public class DeviceController extends BaseCRUDJsonMappingController<DeviceServic ...@@ -99,12 +103,10 @@ public class DeviceController extends BaseCRUDJsonMappingController<DeviceServic
model.put("onlineCount", collect.get(true)); model.put("onlineCount", collect.get(true));
model.put("offlineCount", collect.get(false)); model.put("offlineCount", collect.get(false));
String resp = messageService.siteTree();
//离线设备 按类型分组 JSONObject jsonObject = JSON.parseObject(resp);
// if (collect.get(false) > 0) { JSONArray siteTreeArray = jsonObject.getJSONObject("data").getJSONArray("siteTree");
// Map<String, Long> collectTwo = this.service.find(new DeviceQuery().deviceOnlineStatus(DeviceOnlineStatusEnum.离线.getValue()).active(ActiveEnum.已激活.getValue())).stream().collect(Collectors.groupingBy(x -> deviceTypeMap.get(x.getDeviceType().toString()), Collectors.counting())); model.put("siteTree", siteTreeArray);
// model.put("offlineDeviceType", collectTwo);
// }
super.init(request, response, form, model, context); super.init(request, response, form, model, context);
} }
......
device-manager/src/main/resources/bootstrap.yml
View file @ a3bcf020
...@@ -36,8 +36,8 @@ spring: ...@@ -36,8 +36,8 @@ spring:
enabled: false enabled: false
datasource: datasource:
type: com.alibaba.druid.pool.DruidDataSource type: com.alibaba.druid.pool.DruidDataSource
driver-class-name: com.p6spy.engine.spy.P6SpyDriver #driver-class-name: com.p6spy.engine.spy.P6SpyDriver
#driver-class-name: com.mysql.cj.jdbc.Driver driver-class-name: com.mysql.cj.jdbc.Driver
url: @profiles.datasource.uri@ url: @profiles.datasource.uri@
username: @profiles.datasource.username@ username: @profiles.datasource.username@
password: @profiles.datasource.password@ password: @profiles.datasource.password@
...@@ -110,3 +110,8 @@ token: ...@@ -110,3 +110,8 @@ token:
expireTime: 60 expireTime: 60
# 令牌前缀 # 令牌前缀
prefix: Bearer prefix: Bearer
# 基础服务平台用户名与密码
baseplatform:
httpUrl: http://192.168.0.98:11071/zwfw/
loginName: admin
password: admin
device-manager/src/test/java/com/mortals/httpclient/device/DeviceController.http
View file @ a3bcf020
...@@ -28,7 +28,7 @@ Content-Type: application/json ...@@ -28,7 +28,7 @@ Content-Type: application/json
"status":0, "status":0,
"deviceRemark":"plxklr", "deviceRemark":"plxklr",
"onlineTime":"1646755200000", "onlineTime":"1646755200000",
"offlineTime":"1646755200000", "offlineTime":"1646755200000"
} }
> {% > {%
...@@ -76,5 +76,38 @@ Authorization: Bearer {{authToken}} ...@@ -76,5 +76,38 @@ Authorization: Bearer {{authToken}}
} }
###设备上报
POST {{baseUrl}}/api/upload
Content-Type: application/json
Authorization: Bearer {{authToken}}
{
"deviceNum":"AB:DD:DF:FD:AD:FA:DA:bb",
"action":"upload"
}
###设备上报
POST {{baseUrl}}/test/upload?deviceId=2
Content-Type: application/json
{
"req": 1
}
###getToken
POST {{baseUrl}}/test/getToken
Content-Type: application/json
{
"req": 1
}
###siteTree
POST {{baseUrl}}/test/siteTree
Content-Type: application/json
{}
pom.xml
View file @ a3bcf020
...@@ -33,7 +33,7 @@ ...@@ -33,7 +33,7 @@
<profiles.server.port>18222</profiles.server.port> <profiles.server.port>18222</profiles.server.port>
<profiles.queue.type>rabbitmq</profiles.queue.type> <profiles.queue.type>rabbitmq</profiles.queue.type>
<profiles.datasource.uri> <profiles.datasource.uri>
<![CDATA[jdbc:p6spy:mysql://localhost:3306/device-new-platform?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Hongkong]]></profiles.datasource.uri> <![CDATA[jdbc:mysql://localhost:3306/device-new-platform?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Hongkong]]></profiles.datasource.uri>
<profiles.datasource.username>root</profiles.datasource.username> <profiles.datasource.username>root</profiles.datasource.username>
<profiles.datasource.password>12345678</profiles.datasource.password> <profiles.datasource.password>12345678</profiles.datasource.password>
<profiles.redis.uri>127.0.0.1</profiles.redis.uri> <profiles.redis.uri>127.0.0.1</profiles.redis.uri>
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment