Browse Source

修改处理报警消息的bug

lixing 3 years ago
parent
commit
8414894b9f

+ 5 - 5
src/main/java/com/persagy/cache/AlarmRedisCache.java

@@ -40,7 +40,7 @@ public class AlarmRedisCache implements RedisCache {
 	public AlarmConditionState getAlarmDefineState(String defineId) {
 		AlarmConditionState alarmConditionState = null;
 		try {
-			Object obj = alarmRedisTemplate.opsForHash().get(RedisConstant.DEFINE_ID, defineId);
+			Object obj = alarmRedisTemplate.opsForHash().get(RedisConstant.ALARM_CONDITION_STATE, defineId);
 			if (null != obj && StringUtils.isNotBlank(obj.toString())) {
 				try {
 					alarmConditionState =JSONUtil.toBean(obj.toString(), AlarmConditionState.class);
@@ -70,13 +70,13 @@ public class AlarmRedisCache implements RedisCache {
 	}
 
 	public void setAlarmState(String defineId, AlarmConditionState alarmConditionState) {
-		alarmRedisTemplate.opsForHash().put(RedisConstant.DEFINE_ID, defineId, JSON.toJSONString(alarmConditionState));
+		alarmRedisTemplate.opsForHash().put(RedisConstant.ALARM_CONDITION_STATE, defineId, JSON.toJSONString(alarmConditionState));
 	}
 
 	public AlarmCondition getAlarmConditionRedis(String conditionId) {
 		String str = alarmRedisTemplate.execute((RedisCallback<String>) conn -> {
 			conn.select(14);
-			byte[] condition = conn.hGet(RedisConstant.ALARMCONDITION_ID.getBytes(), conditionId.getBytes());
+			byte[] condition = conn.hGet(RedisConstant.ALARM_CONDITIONS.getBytes(), conditionId.getBytes());
 			return new String(condition);
 		});
 		JSONObject object = JSONObject.parseObject(str);
@@ -86,7 +86,7 @@ public class AlarmRedisCache implements RedisCache {
 
 	public ObjConditionInfo getAlarmConditionsByObjId(String meterId) {
 		try {
-			String str = alarmRedisTemplate.opsForHash().get(RedisConstant.METER_ID, meterId).toString();
+			String str = alarmRedisTemplate.opsForHash().get(RedisConstant.OBJ_CONDITION_REL, meterId).toString();
 			if (StringUtils.isNotBlank(str)) {
 				JSONObject object = JSONObject.parseObject(str);
 				JSONArray conditions = object.getJSONArray("conditions");
@@ -106,7 +106,7 @@ public class AlarmRedisCache implements RedisCache {
 
 	public AlarmCondition getAlarmCondition(String conditionId) {
 		try {
-			Object obj = alarmRedisTemplate.opsForHash().get(RedisConstant.ALARMCONDITION_ID, conditionId);
+			Object obj = alarmRedisTemplate.opsForHash().get(RedisConstant.ALARM_CONDITIONS, conditionId);
 			if (null != obj && StringUtils.isNotBlank(obj.toString())) {
 				JSONObject object = JSONObject.parseObject(obj.toString());
 				AlarmCondition alarmCondition = JSON.toJavaObject(object, AlarmCondition.class);

+ 0 - 0
src/main/java/com/persagy/config/RabbitmqMsgConverterConfig.java


+ 15 - 62
src/main/java/com/persagy/constant/RedisConstant.java

@@ -1,82 +1,35 @@
 package com.persagy.constant;
 
 /**
- * @description: redis常量
- * @author: fenghanchao
- * @company: Persagy Technology Co.,Ltd
- * @since: 2020/7/20 10:29
- * @version: V1.0
- **/
+ * redis常量
+ *
+ * @author lixing
+ * @version V1.0 2021/11/20 6:36 下午
+ */
 public class RedisConstant {
 
     /**
-     * 根据图类型获取数据平台图实例
-     * 格式为Hash格式  "DATA_PLATFORM_GRAPH_INSTANCE:projectId" : {"图类型":"图实例"}
+     * itemCode-objId拼接,用于获取报警定义状态
      */
-    public static final String DATA_PLATFORM_GRAPH_INSTANCE_KEY = "DATA_PLATFORM_GRAPH_INSTANCE";
+    public static final String ALARM_CONDITION_STATE = "ALARM_CONDITION_STATE";
 
     /**
-     * 根据信息点获取表号功能号
-     * 格式为Hash格式 "INFO_CODE_METER_FUC:projectId":{"信息点":"表号-功能号"}
+     * 报警条件
      */
-    public static final String INFO_CODE_METER_FUC_KEY = "INFO_CODE_METER_FUC";
+    public static final String ALARM_CONDITIONS = "ALARM_CONDITIONS";
 
     /**
-     * 根据表号功能号获取信息点
-     * 格式为Hash格式
+     * 设备与报警条件关联关系
      */
-    public static final String METER_FUC_INFO_CODE_KEY = "METER_FUC_INFO_CODE";
+    public static final String OBJ_CONDITION_REL = "OBJ_CONDITION_REL";
 
     /**
-     * 根据表号功能号获取实时数据
-     * 格式为Hash格式 "METER_FUNC_CURRENT_DATA:projectId" : {"表号-功能号" : 实时数据值}
+     * 采集值锁前缀
      */
-    public static final String METER_FUNC_CURRENT_DATA_KEY = "METER_FUNC_CURRENT_DATA";
+    public static final String alarmDataLockPrefix = "ALARM_DATA_LOCK:";
 
     /**
-     * 根据报警定义id获取报警定义详情
-     * 格式为Hash格式 "ALARM_DEFINITION_DETAIL:projectId" : {"报警定义id" : "报警定义详情"}
+     * 报警记录锁前缀
      */
-    public static final String ALARM_DEFINITION_DETAIL_KEY = "ALARM_DEFINITION_DETAIL";
-
-    /**
-     * 根据表号功能号获取多条报警定义id
-     * 格式为Hash格式 "ALARM_DEFINITION_DETAIL:project_id" : {"表号-功能号" : "['报警定义id1','报警定义id2']"}
-     */
-    public static final String ALARM_METER_FUNC_DEFINITION_KEY = "ALARM_METER_FUNC_DEFINITION";
-
-    /**
-     * 根据报警定义id获取对应的表号功能号
-     * 格式为Hash格式 "ALARM_DEFINITION_METER_FUNC:projectId" : {"报警定义id" : "['表号-功能号1','表号-功能号2']"}
-     */
-    public static final String ALARM_DEFINITION_METER_FUNC_KEY = "ALARM_DEFINITION_METER_FUNC";
-
-    /**
-     * 根据报警定义id获取初步满足报警条件的预报警时间
-     * 格式为Hash格式 "ALARM_DEFINITION_PRE_ALARM:projectId" : {"报警定义id" : "2020-07-20 01:00:00"}
-     */
-    public static final String ALARM_DEFINITION_PRE_ALARM_KEY = "ALARM_DEFINITION_PRE_ALARM";
-
-    /**
-     * 根据报警定义id获取报警详情
-     * 格式为Hash格式 "ALARM_DEFINITION_ALARM:projectId" : {"报警定义id" : "{报警定义详情}"}
-     */
-    public static final String ALARM_DEFINITION_ALARM_INFO_KEY = "ALARM_DEFINITION_ALARM_INFO";
-
-    /**
-     * itemCode-objId拼接,用于根据defineId获取报警定义状态
-     */
-    public static final String DEFINE_ID = "ALARM_CONDITION_STATE";
-
-    /**
-     * 报警条件Id REDIS hash 的 id
-     */
-    public static final String ALARMCONDITION_ID = "ALARM_CONDITION";
-
-
-    /**
-     * 设备仪表ID
-     */
-    public static final String METER_ID = "OBJ_CONDITION_REL";
-
+    public static final String alarmRecordLockPrefix = "ALARM_RECORD_LOCK:";
 }

+ 0 - 62
src/main/java/com/persagy/controller/HelloWorld.java

@@ -1,62 +0,0 @@
-package com.persagy.controller;
-
-
-import com.alibaba.fastjson.JSONObject;
-import com.googlecode.aviator.AviatorEvaluator;
-import com.googlecode.aviator.Expression;
-import com.persagy.cache.AlarmInfoCache;
-import com.persagy.client.GroupNettyClient;
-import com.persagy.entity.NettyMessage;
-import com.persagy.enumeration.NettyMsgTypeEnum;
-import com.persagy.repository.AlarmRecordRepository;
-import com.persagy.service.AlarmHandleService;
-import io.swagger.annotations.Api;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
-
-import java.util.Arrays;
-import java.util.HashMap;
-
-@Api(tags = "hello")
-@RestController
-@RequestMapping("hello")
-@Slf4j
-public class HelloWorld {
-    @Autowired
-    AlarmHandleService alarmHandleService;
-
-    @Autowired
-    private GroupNettyClient groupNettyClient;
-    @Autowired
-    AlarmInfoCache alarmInfoCache;
-    @Autowired
-    AlarmRecordRepository alarmRecordRepository;
-
-    @PostMapping("/")
-    public String world(@RequestBody JSONObject jsonObject) throws Exception {
-        return "hello world!";
-    }
-
-
-    private static boolean checkExpression(String expression, HashMap<String, Object> paramMap) {
-        Expression triggerExp = AviatorEvaluator.compile(expression, false);
-        // 执行报警条件公式,得到触发和恢复结果
-        return (Boolean) triggerExp.execute(paramMap);
-    }
-
-    public static void main(String[] args) {
-        String expression = "((ACATAH_FreshAirH!=NaN && ACATAH_RunStatus!=NaN) || (ACATAH_FreshAirH!=NaN && ACATAH_FreshAirH!=NaN)) && (((ACATAH_FreshAirH==NaN || ACATAH_RunStatus==NaN)?false:((ACATAH_FreshAirH>=1) && (ACATAH_RunStatus==0))) || ((ACATAH_FreshAirH==NaN || ACATAH_FreshAirH==NaN)?false:((ACATAH_FreshAirH>=ACATAH_FreshAirH+1))))";
-
-        HashMap<String, Object> paramMap = new HashMap<>();
-        //		paramMap.put("ACATAH_FreshAirH", 1.0);
-        paramMap.put("ACATAH_RunStatus", 0.0);
-        //		paramMap.put("EnergyData", 6.125);
-        //		paramMap.put("Pclose", 5.0);
-        System.out.println(checkExpression(expression, paramMap));
-    }
-}
-

+ 0 - 48
src/main/java/com/persagy/controller/TestController.java

@@ -1,48 +0,0 @@
-//package com.persagy.controller;
-//
-//import com.alibaba.fastjson.JSONObject;
-//import com.persagy.commons.netty.client.GroupNettyClient;
-//import com.persagy.commons.netty.client.TerminalClient;
-//import com.persagy.service.SimpleSchedule;
-//import org.quartz.SchedulerException;
-//import org.springframework.beans.factory.annotation.Autowired;
-//import org.springframework.web.bind.annotation.RequestBody;
-//import org.springframework.web.bind.annotation.RequestMapping;
-//import org.springframework.web.bind.annotation.RestController;
-//
-//@RestController
-//public class TestController {
-//
-//
-//	@Autowired
-//	private GroupNettyClient groupNettyClient;
-//	@Autowired
-//	SimpleSchedule simpleSchedule;
-//	@Autowired
-//	private TerminalClient terminalClient;
-//
-//	@RequestMapping("/test1")
-//	public String test1 () throws SchedulerException {
-//		for (int i = 0; i < 10; i++) {
-//			simpleSchedule.init();
-//		}
-//		return "su";
-//	}
-//
-//	@RequestMapping("/test2")
-//	public String test2 (@RequestBody JSONObject msg) {
-////		JSONObject a = new JSONObject();
-////		a.put("type","request");
-////		a.put("function","ddd");
-//
-//		groupNettyClient.sendMessage(msg.getString("msg"));
-//		return null;
-//	}
-//
-//	@RequestMapping("/test3")
-//	public String test3 (@RequestBody JSONObject msg) throws Exception {
-//		terminalClient.sendMessage(msg.getString("msg"));
-//		return "success";
-//	}
-//
-//}

+ 0 - 140
src/main/java/com/persagy/demo/netty/websocket/client/WebSocketClient.java

@@ -1,140 +0,0 @@
-/*
- * Copyright 2014 The Netty Project
- *
- * The Netty Project licenses this file to you under the Apache License,
- * version 2.0 (the "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at:
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations
- * under the License.
- */
-package com.persagy.demo.netty.websocket.client;
-
-import io.netty.bootstrap.Bootstrap;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelPipeline;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
-import io.netty.handler.codec.http.DefaultHttpHeaders;
-import io.netty.handler.codec.http.HttpClientCodec;
-import io.netty.handler.codec.http.HttpObjectAggregator;
-import io.netty.handler.codec.http.websocketx.*;
-import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler;
-import io.netty.handler.ssl.SslContext;
-import io.netty.handler.ssl.SslContextBuilder;
-import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
-import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.web.socket.WebSocketHandler;
-import org.springframework.web.socket.WebSocketHttpHeaders;
-import org.springframework.web.socket.WebSocketSession;
-
-import java.net.URI;
-
-/**
- * This is an example of a WebSocket client.
- * <p>
- * In order to run this example you need a compatible WebSocket server.
- * Therefore you can either start the WebSocket server from the examples
- * by running {@link io.netty.example.http.websocketx.server}
- * or connect to an existing WebSocket server such as
- * <a href="http://www.websocket.org/echo.html">ws://echo.websocket.org</a>.
- * <p>
- * The client will attempt to connect to the URI passed to it as the first argument.
- * You don't have to specify any arguments if you want to connect to the example WebSocket server,
- * as this is the default.
- * 对于客户端不是netty创建的websocket,连接时候回报错Invalid challenge,暂时没有找到解决办法
- */
-public final class WebSocketClient implements org.springframework.web.socket.client.WebSocketClient {
-
-//    static final String URL = System.getProperty("url", "ws://127.0.0.1:9981/websocket");
-    static final String URL = System.getProperty("url", "ws://121.40.165.18:8800");
-
-    public static void main(String[] args) throws Exception {
-        URI uri = new URI(URL);
-        String scheme = uri.getScheme() == null? "ws" : uri.getScheme();
-        final String host = uri.getHost() == null? "127.0.0.1" : uri.getHost();
-        final int port;
-        if (uri.getPort() == -1) {
-            if ("ws".equalsIgnoreCase(scheme)) {
-                port = 80;
-            } else if ("wss".equalsIgnoreCase(scheme)) {
-                port = 443;
-            } else {
-                port = -1;
-            }
-        } else {
-            port = uri.getPort();
-        }
-
-        if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
-            System.err.println("Only WS(S) is supported.");
-            return;
-        }
-
-        final boolean ssl = "wss".equalsIgnoreCase(scheme);
-        final SslContext sslCtx;
-        if (ssl) {
-            sslCtx = SslContextBuilder.forClient()
-                .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
-        } else {
-            sslCtx = null;
-        }
-
-        EventLoopGroup group = new NioEventLoopGroup();
-        try {
-            // Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
-            // If you change it to V00, ping is not supported and remember to change
-            // HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
-            final WebSocketClientHandler handler =
-                    new WebSocketClientHandler(
-                            WebSocketClientHandshakerFactory.newHandshaker(
-                                    uri, WebSocketVersion.V13, null, true, new DefaultHttpHeaders()));
-
-
-            Bootstrap b = new Bootstrap();
-            b.group(group)
-             .channel(NioSocketChannel.class)
-             .handler(new ChannelInitializer<SocketChannel>() {
-                 @Override
-                 protected void initChannel(SocketChannel ch) {
-                     ChannelPipeline p = ch.pipeline();
-                     if (sslCtx != null) {
-                         p.addLast(sslCtx.newHandler(ch.alloc(), host, port));
-                     }
-                     p.addLast(
-                             new HttpClientCodec(),
-                             new HttpObjectAggregator(8192),
-                             WebSocketClientCompressionHandler.INSTANCE,
-                             handler);
-                 }
-             });
-
-            Channel ch = b.connect(uri.getHost(), port).sync().channel();
-            handler.handshakeFuture().sync();
-            String msg = "console.readLine!";
-            WebSocketFrame frame = new TextWebSocketFrame(msg);
-            ch.writeAndFlush(msg);
-            ch.closeFuture().sync();
-        } finally {
-            group.shutdownGracefully();
-        }
-    }
-
-    @Override
-    public ListenableFuture<WebSocketSession> doHandshake(WebSocketHandler webSocketHandler, String s, Object... objects) {
-        return null;
-    }
-
-    @Override
-    public ListenableFuture<WebSocketSession> doHandshake(WebSocketHandler webSocketHandler, WebSocketHttpHeaders webSocketHttpHeaders, URI uri) {
-        return null;
-    }
-}

+ 0 - 147
src/main/java/com/persagy/demo/netty/websocket/client/WebSocketClientHandler.java

@@ -1,147 +0,0 @@
-/*
- * Copyright 2012 The Netty Project
- *
- * The Netty Project licenses this file to you under the Apache License,
- * version 2.0 (the "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at:
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations
- * under the License.
- */
-//The MIT License
-//
-//Copyright (c) 2009 Carl Bystršm
-//
-//Permission is hereby granted, free of charge, to any person obtaining a copy
-//of this software and associated documentation files (the "Software"), to deal
-//in the Software without restriction, including without limitation the rights
-//to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-//copies of the Software, and to permit persons to whom the Software is
-//furnished to do so, subject to the following conditions:
-//
-//The above copyright notice and this permission notice shall be included in
-//all copies or substantial portions of the Software.
-//
-//THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-//IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-//FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-//AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-//LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-//OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-//THE SOFTWARE.
-
-package com.persagy.demo.netty.websocket.client;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.*;
-import io.netty.handler.codec.http.FullHttpResponse;
-import io.netty.handler.codec.http.websocketx.*;
-import io.netty.util.CharsetUtil;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class WebSocketClientHandler extends SimpleChannelInboundHandler<Object> {
-
-    private final WebSocketClientHandshaker handshaker;
-    private ChannelPromise handshakeFuture;
-
-    public WebSocketClientHandler(WebSocketClientHandshaker handshaker) {
-        this.handshaker = handshaker;
-    }
-
-    public ChannelFuture handshakeFuture() {
-        return handshakeFuture;
-    }
-
-    @Override
-    public void handlerAdded(ChannelHandlerContext ctx) {
-        handshakeFuture = ctx.newPromise();
-    }
-
-    @Override
-    public void channelActive(ChannelHandlerContext ctx) {
-        handshaker.handshake(ctx.channel());
-    }
-
-    @Override
-    public void channelInactive(ChannelHandlerContext ctx) {
-        System.out.println("WebSocket Client disconnected!");
-    }
-
-    @Override
-    public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
-        Channel ch = ctx.channel();
-
-        try {
-            readMsg(msg, ctx);
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-    }
-
-    private void readMsg(Object msg, ChannelHandlerContext ctx) {
-        Channel ch = ctx.channel();
-//        if ((msg instanceof FullHttpResponse)) {
-//            ctx.fireChannelRead(msg);
-//            return;
-//        }
-//        FullHttpResponse x = (FullHttpResponse) msg;
-//        System.out.println("---x---------:" + x);
-//        ByteBuf buf = (ByteBuf) msg;
-//        //创建目标大小的数组
-//        byte[] barray = new byte[buf.readableBytes()];
-//        //把数据从bytebuf转移到byte[]
-//        buf.getBytes(0, barray);
-//        //将byte[]转成字符串用于打印
-//        String str = new String(barray);
-//        System.out.println("------------:" + str);
-//        if (true) {
-//            return;
-//        }
-        if (!handshaker.isHandshakeComplete()) {
-            try {
-                handshaker.finishHandshake(ch, (FullHttpResponse) msg);
-                System.out.println("WebSocket Client connected!");
-                handshakeFuture.setSuccess();
-            } catch (WebSocketHandshakeException e) {
-                log.error("WebSocket Client failed to connect",e);
-                handshakeFuture.setFailure(e);
-            }
-            return;
-        }
-
-        if (msg instanceof FullHttpResponse) {
-            FullHttpResponse response = (FullHttpResponse) msg;
-            throw new IllegalStateException(
-                    "Unexpected FullHttpResponse (getStatus=" + response.status() +
-                            ", content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
-        }
-
-        WebSocketFrame frame = (WebSocketFrame) msg;
-//        String clientMsg = "console.readLine!";
-//        WebSocketFrame clientFrame = new TextWebSocketFrame(clientMsg);
-        if (frame instanceof TextWebSocketFrame) {
-            TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
-            System.out.println("WebSocket Client received message: " + textFrame.text());
-        } else if (frame instanceof PongWebSocketFrame) {
-            System.out.println("WebSocket Client received pong");
-        } else if (frame instanceof CloseWebSocketFrame) {
-            System.out.println("WebSocket Client received closing");
-            ch.close();
-        }
-    }
-
-    @Override
-    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-        cause.printStackTrace();
-        if (!handshakeFuture.isDone()) {
-            handshakeFuture.setFailure(cause);
-        }
-        ctx.close();
-    }
-}

+ 0 - 17
src/main/java/com/persagy/demo/netty/websocket/server/MyChannelHandlerPool.java

@@ -1,17 +0,0 @@
-package com.persagy.demo.netty.websocket.server;
-
-import io.netty.channel.group.ChannelGroup;
-import io.netty.channel.group.DefaultChannelGroup;
-import io.netty.util.concurrent.GlobalEventExecutor;
-
-/**
- * MyChannelHandlerPool
- * 通道组池,管理所有websocket连接
- */
-public class MyChannelHandlerPool {
-
-    public MyChannelHandlerPool(){}
-
-    public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
-
-}

+ 0 - 87
src/main/java/com/persagy/demo/netty/websocket/server/MyWebSocketHandler.java

@@ -1,87 +0,0 @@
-package com.persagy.demo.netty.websocket.server;
-
-import com.alibaba.fastjson.JSON;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.SimpleChannelInboundHandler;
-import io.netty.handler.codec.http.FullHttpRequest;
-import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * MyWebSocketHandler
- * WebSocket处理器,处理websocket连接相关
- */
-public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
-
-    @Override
-    public void channelActive(ChannelHandlerContext ctx) throws Exception {
-        System.out.println("与客户端建立连接,通道开启!");
-
-        //添加到channelGroup通道组
-        MyChannelHandlerPool.channelGroup.add(ctx.channel());
-    }
-
-    @Override
-    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-        System.out.println("与客户端断开连接,通道关闭!");
-        //添加到channelGroup 通道组
-        MyChannelHandlerPool.channelGroup.remove(ctx.channel());
-    }
-
-    @Override
-    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        //首次连接是FullHttpRequest,处理参数 by zhengkai.blog.csdn.net
-        if (null != msg && msg instanceof FullHttpRequest) {
-            FullHttpRequest request = (FullHttpRequest) msg;
-            String uri = request.uri();
-
-            Map paramMap=getUrlParams(uri);
-            System.out.println("接收到的参数是:"+JSON.toJSONString(paramMap));
-            //如果url包含参数,需要处理
-            if(uri.contains("?")){
-                String newUri=uri.substring(0,uri.indexOf("?"));
-                System.out.println(newUri);
-                request.setUri(newUri);
-            }
-
-        }else if(msg instanceof TextWebSocketFrame){
-            //正常的TEXT消息类型
-            TextWebSocketFrame frame=(TextWebSocketFrame)msg;
-            System.out.println("客户端收到服务器数据:" +frame.text());
-            sendAllMessage(frame.text());
-        }
-        super.channelRead(ctx, msg);
-    }
-
-    @Override
-    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {
-
-    }
-
-    private void sendAllMessage(String message){
-        //收到信息后,群发给所有channel
-        MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
-    }
-
-    private static Map getUrlParams(String url){
-        Map<String,String> map = new HashMap<>();
-        url = url.replace("?",";");
-        if (!url.contains(";")){
-            return map;
-        }
-        if (url.split(";").length > 0){
-            String[] arr = url.split(";")[1].split("&");
-            for (String s : arr){
-                String key = s.split("=")[0];
-                String value = s.split("=")[1];
-                map.put(key,value);
-            }
-            return  map;
-
-        }else{
-            return map;
-        }
-    }
-}

+ 0 - 59
src/main/java/com/persagy/demo/netty/websocket/server/NettyServer.java

@@ -1,59 +0,0 @@
-package com.persagy.demo.netty.websocket.server;
-
-import io.netty.bootstrap.ServerBootstrap;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.handler.codec.http.HttpObjectAggregator;
-import io.netty.handler.codec.http.HttpServerCodec;
-import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
-import io.netty.handler.stream.ChunkedWriteHandler;
-
-
-/**
- * NettyServer Netty服务器配置
- * https://github.com/moshowgame/spring-cloud-study.git
- */
-public class NettyServer {
-    private final int port;
-
-    public NettyServer(int port) {
-        this.port = port;
-    }
-
-    public void start() throws Exception {
-        EventLoopGroup bossGroup = new NioEventLoopGroup();
-
-        EventLoopGroup group = new NioEventLoopGroup();
-        try {
-            ServerBootstrap sb = new ServerBootstrap();
-            sb.option(ChannelOption.SO_BACKLOG, 1024);
-            sb.group(group, bossGroup) // 绑定线程池
-                    .channel(NioServerSocketChannel.class) // 指定使用的channel
-                    .localAddress(this.port)// 绑定监听端口
-                    .childHandler(new ChannelInitializer<SocketChannel>() { // 绑定客户端连接时候触发操作
-                        @Override
-                        protected void initChannel(SocketChannel ch) throws Exception {
-                            System.out.println("收到新连接");
-                            //websocket协议本身是基于http协议的,所以这边也要使用http解编码器
-                            ch.pipeline().addLast(new HttpServerCodec());
-                            //以块的方式来写的处理器
-                            ch.pipeline().addLast(new ChunkedWriteHandler());
-                            ch.pipeline().addLast(new HttpObjectAggregator(8192));
-                            ch.pipeline().addLast(new MyWebSocketHandler());
-                            ch.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket", null, true, 65536 * 10));
-                        }
-                    });
-            ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
-            System.out.println(NettyServer.class + " 启动正在监听: " + cf.channel().localAddress());
-            cf.channel().closeFuture().sync(); // 关闭服务器通道
-        } finally {
-            group.shutdownGracefully().sync(); // 释放线程池资源
-            bossGroup.shutdownGracefully().sync();
-        }
-    }
-}

+ 49 - 0
src/main/java/com/persagy/enumeration/AlarmRecordMsgEnum.java

@@ -0,0 +1,49 @@
+package com.persagy.enumeration;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * 报警记录
+ */
+@AllArgsConstructor
+public enum AlarmRecordMsgEnum {
+    /**
+     * 字典类型
+     */
+    NEW_ALARM_RECORD("new_alarm_record", "新增报警记录"),
+    UPDATE_ALARM_RECORD("update_alarm_record", "更新报警记录"),
+    CONTINUE_ALARM("continue_alarm", "报警持续处理");
+
+    @Setter
+    @Getter
+    private String type;
+    @Setter
+    @Getter
+    private String desc;
+
+    /**
+     * 根据type获取到枚举对象
+     *
+     * @param type 报告状态编码
+     * @return 报告状态枚举对象
+     * @author lixing
+     * @version V1.0 2021/5/31 4:06 下午
+     */
+    public static AlarmRecordMsgEnum getByType(String type) {
+        if (StringUtils.isBlank(type)) {
+            return null;
+        }
+        AlarmRecordMsgEnum[] enums = AlarmRecordMsgEnum.values();
+        for (AlarmRecordMsgEnum item : enums) {
+            if (type.equals(item.getType())) {
+                return item;
+            }
+        }
+        return null;
+    }
+}
+
+

+ 0 - 13
src/main/java/com/persagy/init/InitRunner.java

@@ -20,25 +20,12 @@ import lombok.extern.slf4j.Slf4j;
 @Slf4j
 public class InitRunner implements CommandLineRunner {
 	@Autowired
-	private GroupNettyClient groupNettyClient;
-	@Autowired
 	AlarmQuartzService alarmQuartzService;
-	@Autowired
-	CreatedAlarmIdsCache createdAlarmIdsCache;
-	@Autowired
-	AlarmLastTimeCache alarmLastTimeCache;
-
 
 	@Override
 	public void run(String... args) throws Exception {
-		// 已创建的报警id缓存初始化
-		createdAlarmIdsCache.init();
-		// 报警持续时间缓存初始化
-		alarmLastTimeCache.init();
 		//5.0 开始引入了 LRU 缓存,可指定缓存的表达式个数,比如设置为最大 1 万个缓存结果:
 		AviatorEvaluator.getInstance().useLRUExpressionCache(10000);
-		//启动netty客户端,接受云端数据
-		groupNettyClient.start();
 		alarmQuartzService.initAlarmMessagehandler();
 	}
 }

+ 6 - 38
src/main/java/com/persagy/job/AlarmExpireJob.java

@@ -10,13 +10,11 @@ import com.persagy.cache.CreatedAlarmIdsCache;
 import com.persagy.client.GroupNettyClient;
 import com.persagy.entity.AlarmConditionState;
 import com.persagy.entity.AlarmRecord;
-import com.persagy.entity.NettyMessage;
 import com.persagy.entity.ZktAlarmRecordDO;
 import com.persagy.entity.v2.DmpMessage;
 import com.persagy.enumeration.AlarmDefineStateEnum;
-import com.persagy.enumeration.NettyMsgTypeEnum;
+import com.persagy.enumeration.AlarmRecordMsgEnum;
 import com.persagy.repository.AlarmRecordRepository;
-import com.persagy.service.AlarmHandleService;
 import com.persagy.service.impl.AlarmHandleServiceImpl;
 import com.persagy.utils.DateUtils;
 import com.persagy.utils.StringUtil;
@@ -27,7 +25,6 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.quartz.QuartzJobBean;
 
-import java.util.Collections;
 import java.util.Date;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicLong;
@@ -48,7 +45,7 @@ public class AlarmExpireJob extends QuartzJobBean {
     @Autowired
     CreatedAlarmIdsCache createdAlarmIdsCache;
     @Autowired
-    AlarmHandleService alarmHandleService;
+    AlarmHandleServiceImpl alarmHandleService;
     @Autowired
     AlarmRedisCache alarmRedisCache;
     /**
@@ -155,6 +152,7 @@ public class AlarmExpireJob extends QuartzJobBean {
                         .id(alarmId)
                         .state(StringUtil.getInt(state))
                         .groupCode(groupCode)
+                        .itemCode(zktAlarmRecordDO.getItemCode())
                         .projectId(zktAlarmRecordDO.getProjectId())
                         .build();
                 // 恢复任务发送消息时,带上恢复信息
@@ -171,8 +169,9 @@ public class AlarmExpireJob extends QuartzJobBean {
                 dmpMessage.setExts(JSONObject.parseObject(JSONObject.toJSONString(message)));
                 //{"id","123", "state":1, "groupCode":"wd", "projectId":"Pj123"}
                 log.info("定时任务执行完成,向云端推送{}消息", jobType);
-                // TODO: 2021/11/17 向mq推送消息
-                alarmHandleService.sendToMQ(JSONObject.parseObject(JSONObject.toJSONString(dmpMessage)),alarmId,zktAlarmRecordDO.getProjectId());
+                alarmHandleService.sendToMQ(
+                        JSONObject.parseObject(JSONObject.toJSONString(dmpMessage)),
+                        alarmId,zktAlarmRecordDO.getProjectId(), AlarmRecordMsgEnum.UPDATE_ALARM_RECORD.getType());
                 // 删除缓存中报警id
                 //createdAlarmIdsCache.remove(alarmId);
 
@@ -230,35 +229,4 @@ public class AlarmExpireJob extends QuartzJobBean {
         }
     }
 
-    public String getAlarmRecord() {
-        return alarmRecord;
-    }
-
-    public void setAlarmRecord(String alarmRecord) {
-        this.alarmRecord = alarmRecord;
-    }
-
-    public String getRefire() {
-        return refire;
-    }
-
-    public void setRefire(String refire) {
-        this.refire = refire;
-    }
-
-    public String getExpireTime() {
-        return expireTime;
-    }
-
-    public void setExpireTime(String expireTime) {
-        this.expireTime = expireTime;
-    }
-
-    public String getDefineId() {
-        return defineId;
-    }
-
-    public void setDefineId(String defineId) {
-        this.defineId = defineId;
-    }
 }

+ 0 - 60
src/main/java/com/persagy/job/SpringSchedule.java

@@ -1,60 +0,0 @@
-package com.persagy.job;
-
-import cn.hutool.core.date.DateUtil;
-import cn.hutool.core.date.TimeInterval;
-import com.persagy.client.GroupNettyClient;
-import com.persagy.entity.NettyMessage;
-import com.persagy.enumeration.NettyMsgTypeEnum;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.scheduling.annotation.Scheduled;
-import org.springframework.stereotype.Service;
-
-import java.util.Objects;
-
-/**
- * @author LuoGuangyi
- * @title: SpringSchedule
- * @projectName spring-boot-examples
- * @description:
- * @date 2019/09/17 16:54
- */
-
-@Service
-@Slf4j
-public class SpringSchedule {
-    @Autowired
-    GroupNettyClient groupNettyClient;
-
-//    @Scheduled(cron = "${alarm.get.all.alarmdefine.cron}")
-    public void allResetCron() throws InterruptedException {
-        NettyMessage message = new NettyMessage(NettyMsgTypeEnum.REQUEST_ALL_CONFIGS, groupNettyClient.projectId);
-        groupNettyClient.sendMessage(message.toString());
-    }
-
-
-    @Scheduled(initialDelay = 1000, fixedRate = 600000)
-    public void connectAnalizeCron() {
-        if (Objects.isNull(GroupNettyClient.channelGroup)) {
-            log.warn("NettyClient is not init");
-            return;
-        }
-        log.info("NettyClient State:isActive[{}],isOpen[{}],isRegistered[{}],isWritable[{}]", GroupNettyClient.channelGroup.isActive(), GroupNettyClient.channelGroup.isOpen(), GroupNettyClient.channelGroup.isRegistered(), GroupNettyClient.channelGroup.isWritable());
-        sengAlarmMessage();
-    }
-
-    private void sengAlarmMessage() {
-        try {
-            log.info("--sengAlarmMessage--");
-            TimeInterval timer = DateUtil.timer();
-            while (NettyMessageQueue.getNettyMessageQueue().size() > 0 && timer.interval() < 100000) {
-                String msg = NettyMessageQueue.getNettyMessageQueue().consume();
-                log.info("剩余报警消息令总数:{}", NettyMessageQueue.getNettyMessageQueue().size());
-                groupNettyClient.sendMessage(msg);
-            }
-        } catch (Exception e) {
-            log.error("发送报警消息失败", e);
-        }
-    }
-
-}

+ 0 - 28
src/main/java/com/persagy/service/AlarmHandleService.java

@@ -1,28 +0,0 @@
-package com.persagy.service;
-
-import com.alibaba.fastjson.JSONObject;
-import org.quartz.SchedulerException;
-
-/**
- * @description: 报警处理接口:包含报警产生和报警恢复
- * @author:LuoGuangyi
- * @company:PersagyTechnologyCo.,Ltd
- * @since:2020/10/15 11:03
- * @version:V1.0
- **/
-public interface AlarmHandleService {
-
-    /**
-     * 处理接收到的iot消息
-     *
-     * @param msg iot消息
-     * @exception Exception throw when 拼装报警消息,netty发送消息等异常
-     * @author lixing
-     * @company Persagy Technology Co.,Ltd
-     * @since 2021/2/6 10:25 上午
-     * @version V1.0
-     */
-//    void handleIotMsg(String msg) throws Exception;
-
-    void sendToMQ(JSONObject parseObject, String alarmId, String projectId);
-}

File diff suppressed because it is too large
+ 665 - 673
src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java