Quellcode durchsuchen

修复rabbitMQ注册问题

lixing vor 3 Jahren
Ursprung
Commit
c3d9e1645c

+ 0 - 146
src/main/java/com/persagy/client/GroupNettyClient.java

@@ -1,146 +0,0 @@
-package com.persagy.client;
-
-import com.persagy.cache.CreatedAlarmIdsCache;
-import com.persagy.job.NettyMessageQueue;
-import com.persagy.service.impl.NettyMsgHandler;
-import io.netty.bootstrap.Bootstrap;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
-import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
-import io.netty.handler.codec.LengthFieldPrepender;
-import io.netty.handler.codec.string.StringDecoder;
-import io.netty.handler.codec.string.StringEncoder;
-import io.netty.handler.timeout.IdleStateHandler;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Service;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * @author lqshi
- * @ClassName: NettyClient
- * @Description: netty启动、发送消息等
- * @date 2020年8月31日 上午10:34:45
- */
-@Service
-@Slf4j
-public class GroupNettyClient {
-    @Autowired
-    CreatedAlarmIdsCache createdAlarmIdsCache;
-    @Autowired
-    private NettyMsgHandler nettyMsgHandler;
-
-    static Bootstrap groupBootstrap = new Bootstrap();
-    public static Channel channelGroup;
-    @Value("${group.alarm.host:127.0.0.1}")
-    public String host;
-    @Value("${group.alarm.port}")
-    public int port;
-    /**
-     * 项目名称
-     */
-    @Value("${project.id}")
-    public String projectId;
-    /**
-     * 集团编码
-     */
-    @Value("${group.code}")
-    public String groupCode;
-
-    //会话对象组
-    public static Map<String, Channel> channelsMap = new ConcurrentHashMap<>();
-
-
-    /**
-     * @param
-     * @description:启动netty客户端
-     * @exception:
-     * @author: LuoGuangyi
-     * @company: Persagy Technology Co.,Ltd
-     * @return: void
-     * @since: 2020/10/20 15:54
-     * @version: V1.0
-     */
-    public void start() {
-
-        GroupNettyClient groupNettyClient = this;
-        try {
-            // 启动辅助对象类
-            groupBootstrap.group(new NioEventLoopGroup(5))
-                    // 建立通道
-                    .channel(NioSocketChannel.class)
-                    // 初始化通道及进行配置
-                    .handler(new ChannelInitializer<SocketChannel>() {
-                        @Override
-                        protected void initChannel(SocketChannel ch) throws Exception {
-                            ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
-                            ch.pipeline().addLast(new LengthFieldPrepender(4));
-                            // 将分隔之后的字节数据转换为字符串
-                            ch.pipeline().addLast(new StringDecoder());
-                            ch.pipeline().addLast(new StringEncoder());
-                            ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(60, 30, 0));
-                            // pipeline可以理解为所有handler的初始化容器
-                            ch.pipeline().addLast(new GroupNettyClientHandler(
-                                    groupNettyClient,
-                                    nettyMsgHandler,
-                                    createdAlarmIdsCache
-                            ));// 添加自定义handler
-                        }
-                    });
-            // 连接远程节点,等待连接完成
-            // channel = b.connect().sync();//
-            // sync()代表同步等待连接,然后在f.channel().closeFuture().sync();,如果不加就不会等待,直接关闭
-            channelGroup = groupBootstrap.connect(host, port).channel();
-            channelsMap.put("group", channelGroup);
-        } catch (Exception e) {
-            e.printStackTrace();
-        } finally {
-            // 关闭主线程组
-            //回收group后下面的自动重连connect()无法直接使用
-            //group.shutdownGracefully();
-
-        }
-    }
-
-    public void connect(GroupNettyClient groupNettyClient, Channel channel) {
-        // 加入断线后自动重连监听器
-        log.info("try connect!");
-        for (Map.Entry<String, Channel> entry : channelsMap.entrySet()) {
-            if (entry.getValue() == channel && !channel.isActive()) {
-                channelGroup = groupBootstrap.connect(groupNettyClient.host, groupNettyClient.port).addListener(new ChannelFutureListener() {
-                    @Override
-                    public void operationComplete(ChannelFuture future) throws Exception {
-                        if (future.cause() != null) {
-                            log.info("Failed to connect: " + future.cause());
-                        }
-                    }
-                }).channel();
-                channelsMap.put(entry.getKey(), channelGroup);
-            }
-        }
-    }
-
-    /**
-     * @param msg
-     * @Title: sendMessage
-     * @Description: 发送消息
-     */
-    public void sendMessage(String msg) throws InterruptedException {
-        log.info("给云端发送数据: {}", msg);
-        if (channelGroup.isWritable()) {
-            channelGroup.writeAndFlush(msg);
-        } else {
-            log.warn("云端netty不可写,放入缓冲队列中[{}]", msg);
-            NettyMessageQueue.getNettyMessageQueue().produce(msg);
-        }
-    }
-
-}

+ 0 - 229
src/main/java/com/persagy/client/GroupNettyClientHandler.java

@@ -1,229 +0,0 @@
-package com.persagy.client;
-
-import cn.hutool.core.date.DateUtil;
-import cn.hutool.core.date.TimeInterval;
-import com.alibaba.fastjson.JSONObject;
-import com.persagy.cache.CreatedAlarmIdsCache;
-import com.persagy.entity.NettyMessage;
-import com.persagy.enumeration.NettyMsgTypeEnum;
-import com.persagy.job.NettyMessageQueue;
-import com.persagy.service.impl.NettyMsgHandler;
-import com.persagy.utils.StringUtil;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.handler.timeout.IdleState;
-import io.netty.handler.timeout.IdleStateEvent;
-import lombok.extern.slf4j.Slf4j;
-
-import java.util.Arrays;
-import java.util.concurrent.TimeUnit;
-
-/**
- * @author lqshi
- * @ClassName: EchoClientHandler
- * @Description: 客户端处理类(处理云端数据)
- * @date 2020年8月29日 下午8:31:59
- */
-// 注意:SimpleChannelInboundHandler<ByteBuf>的<>中是什么,channelRead0第二参数是什么
-@Slf4j
-public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
-    // Sleep 5 seconds before a reconnection attempt.
-    static final int RECONNECT_DELAY = Integer.parseInt(System.getProperty("reconnectDelay", "5"));
-    // Reconnect when the server sends nothing for 10 seconds.
-    private static final int READ_TIMEOUT = Integer.parseInt(System.getProperty("readTimeout", "10"));
-
-    private GroupNettyClient groupNettyClient;
-    private CreatedAlarmIdsCache createdAlarmIdsCache;
-    private NettyMsgHandler nettyMsgHandler;
-
-
-    public GroupNettyClientHandler(
-            GroupNettyClient groupNettyClient,
-            NettyMsgHandler nettyMsgHandler,
-            CreatedAlarmIdsCache createdAlarmIdsCache
-    ) {
-        this.groupNettyClient = groupNettyClient;
-        this.createdAlarmIdsCache = createdAlarmIdsCache;
-        this.nettyMsgHandler = nettyMsgHandler;
-    }
-
-    @Override
-    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
-
-        if (!(evt instanceof IdleStateEvent)) {
-            return;
-        }
-        IdleStateEvent e = (IdleStateEvent) evt;
-        if (e.state() == IdleState.READER_IDLE) {
-            // 如果一直未读取到数据,则关闭通道,触发通道的重新连接
-            ctx.close();
-        } else if (e.state() == IdleState.WRITER_IDLE) {
-            // 如果客户端一直闲置(没有写操作),则发送一个心跳包。服务端会返回一个心跳包,避免通道关闭
-            NettyMessage heartBeat = new NettyMessage(groupNettyClient.projectId);
-            heartBeat.setOpCode(NettyMsgTypeEnum.HEART_BEAT);
-            ctx.writeAndFlush(heartBeat.toString());
-        }
-    }
-
-    /**
-     * 在到服务器的连接已经建立之后将被调用
-     *
-     * @param ctx
-     * @throws Exception
-     */
-    @Override
-    public void channelActive(ChannelHandlerContext ctx) throws Exception {
-        log.info("Connected to: " + ctx.channel().remoteAddress());
-        ctx.channel().writeAndFlush(new NettyMessage<>(
-                NettyMsgTypeEnum.CONNECT, groupNettyClient.projectId).toString());
-        //启动的时候发送消息,获取全部报警定义
-        //{"groupCode":"wd", "projectId":"Pj123"}
-        NettyMessage nettyMessage = new NettyMessage(
-                NettyMsgTypeEnum.REQUEST_ALL_CONFIGS, groupNettyClient.projectId);
-        nettyMessage.setRemark("连接已经建立;");
-        JSONObject content = new JSONObject();
-        content.put("groupCode", groupNettyClient.groupCode);
-        content.put("projectId", groupNettyClient.projectId);
-        nettyMessage.setContent(Arrays.asList(content));
-        log.info(nettyMessage.toString());
-        ctx.channel().writeAndFlush(nettyMessage.toString());
-        resumeSendUndeliveredMsg(ctx);
-    }
-
-    private void resumeSendUndeliveredMsg(ChannelHandlerContext ctx) {
-        try {
-            TimeInterval timer = DateUtil.timer();
-            while (NettyMessageQueue.getNettyMessageQueue().size() > 0 && timer.interval() < 10000) {
-                String msg = NettyMessageQueue.getNettyMessageQueue().consume();
-                log.info("剩余未发送消息总数:{}", NettyMessageQueue.getNettyMessageQueue().size());
-                ctx.writeAndFlush(msg);
-            }
-        } catch (Exception e) {
-            log.error("发送消息失败", e);
-        }
-    }
-
-    /**
-     * 当从服务器接收到一个消息时被调用
-     *
-     * @param ctx
-     * @param msg
-     * @throws Exception
-     */
-    @Override
-    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        if (StringUtil.isJSONObject((String) msg)) {
-            NettyMessage message = StringUtil.tranferItemToDTO((String) msg, NettyMessage.class);
-            if (message.getOpCode() == NettyMsgTypeEnum.ALL_CONDITIONS) {
-                log.info("获取到全量的报警条件, 共[{}]条", message.getContent().size());
-            } else if (message.getOpCode() == NettyMsgTypeEnum.ALL_OBJ_CONDITION_REL) {
-                log.info("获取到设备与报警条件的关联关系, 共[{}]条", message.getContent().size());
-            } else {
-                log.info("接收到netty消息: {}", msg);
-            }
-        }
-
-        try {
-            TimeInterval timer = DateUtil.timer();
-            handlerMsg(ctx, msg);
-            log.debug("处理消息时间[{}]", timer.interval());
-        } catch (Exception e) {
-            log.error("channelRead", e);
-        }
-    }
-
-    public void handlerMsg(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
-        if (StringUtil.isJSONObject((String) msg)) {
-            NettyMessage message = StringUtil.tranferItemToDTO((String) msg, NettyMessage.class);
-            NettyMsgTypeEnum opCode = message.getOpCode();
-            // 通过增加synchronized关键字控制同一类消息顺序执行,
-            // 避免出现同时更新导致的异常。例如:先接到删除数据的信息,又接到新增数据的信息
-            // 如果不保证执行顺序,很可能无法得到想要的结果
-            switch (opCode) {
-                case ALL_CONDITIONS:
-                    synchronized ("sync_condition") {
-                        nettyMsgHandler.cacheAllAlarmConditions(msg);
-                    }
-                    break;
-                case NEW_CONDITION:
-                    synchronized ("sync_condition") {
-                        nettyMsgHandler.cacheNewCondition(msg);
-                    }
-                    break;
-                case UPDATE_CONDITION:
-                    synchronized ("sync_condition") {
-                        nettyMsgHandler.cacheUpdatedCondition(msg);
-                    }
-                    break;
-                case DELETE_CONDITION:
-                    synchronized ("sync_condition") {
-                        nettyMsgHandler.removeCachedCondition(msg);
-                    }
-                    break;
-                case ALL_OBJ_CONDITION_REL:
-                    synchronized ("sync_obj_condition_rel") {
-                        nettyMsgHandler.cacheAllObjConditionRel(msg);
-                    }
-                    break;
-                case NEW_OBJ_CONDITION_REL:
-                    synchronized ("sync_obj_condition_rel") {
-                        nettyMsgHandler.cacheNewObjConditionRel(msg);
-                    }
-                    break;
-                case DELETE_OBJ_CONDITION_REL:
-                    synchronized ("sync_obj_condition_rel") {
-                        nettyMsgHandler.removeCachedObjConditionRel(msg);
-                    }
-                    break;
-                case RECORD_ID:
-                    nettyMsgHandler.cacheRecordId(message);
-                    break;
-                default:
-                    break;
-            }
-            nettyMsgHandler.logCacheInfo();
-            // 发送已接收回执
-            nettyMsgHandler.acceptedReply(channelHandlerContext);
-        }
-    }
-
-    @Override
-    public void channelReadComplete(ChannelHandlerContext ctx) {
-        ctx.flush();
-    }
-
-    /**
-     * 在处理过程中引发异常时被调用
-     *
-     * @param ctx
-     * @param cause
-     * @throws Exception
-     */
-    @Override
-    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-        cause.printStackTrace();
-        ctx.close();
-    }
-
-    /**
-     * 通道处于非活跃状态动作,该方法只会在失效时调用一次
-     */
-    @Override
-    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-        //客户端自己不正常情况下自己在重连一次
-        log.info("Disconnected from: " + ctx.channel().remoteAddress());
-
-    }
-
-    @Override
-    public void channelUnregistered(final ChannelHandlerContext ctx) throws Exception {
-        log.info("Sleeping for: " + RECONNECT_DELAY + "s,Reconnecting to: " + groupNettyClient.host + ':' + groupNettyClient.port);
-        ctx.channel().eventLoop().schedule(new Runnable() {
-            @Override
-            public void run() {
-                log.info("Reconnecting to: " + groupNettyClient.host + ':' + groupNettyClient.port);
-                groupNettyClient.connect(groupNettyClient, ctx.channel());
-            }
-        }, RECONNECT_DELAY, TimeUnit.SECONDS);
-    }
-}

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

@@ -7,9 +7,6 @@ import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Service;
 
 import com.googlecode.aviator.AviatorEvaluator;
-import com.persagy.cache.AlarmLastTimeCache;
-import com.persagy.cache.CreatedAlarmIdsCache;
-import com.persagy.client.GroupNettyClient;
 import com.persagy.service.AlarmQuartzService;
 
 import lombok.extern.slf4j.Slf4j;

+ 0 - 3
src/main/java/com/persagy/job/AlarmExpireJob.java

@@ -7,7 +7,6 @@ import com.alibaba.fastjson.JSONObject;
 import com.persagy.cache.AlarmInfoCache;
 import com.persagy.cache.AlarmRedisCache;
 import com.persagy.cache.CreatedAlarmIdsCache;
-import com.persagy.client.GroupNettyClient;
 import com.persagy.entity.AlarmConditionState;
 import com.persagy.entity.AlarmRecord;
 import com.persagy.entity.ZktAlarmRecordDO;
@@ -37,8 +36,6 @@ import static org.quartz.SimpleScheduleBuilder.simpleSchedule;
 public class AlarmExpireJob extends QuartzJobBean {
 
     @Autowired
-    private GroupNettyClient groupNettyClient;
-    @Autowired
     AlarmRecordRepository alarmRecordRepository;
     @Autowired
     AlarmInfoCache alarmInfoCache;

+ 11 - 3
src/main/java/com/persagy/mq/RabbitConfig.java

@@ -25,9 +25,17 @@ public class RabbitConfig {
 	private final String alarmDataQueue = "alarm-data-queue";
 
 	@Bean
-	public Binding conditionBindingAlarm() {
-		return BindingBuilder.bind(new Queue(alarmDataQueue, true)).to(new TopicExchange(exchange))
-				.with(alarmMsgRouting);
+	public TopicExchange exchange() {
+		return new TopicExchange(exchange);
 	}
 
+	@Bean
+	public Queue alarmDataQueue() {
+		return new Queue(alarmDataQueue, true);
+	}
+
+	@Bean
+	public Binding alarmObjBinding() {
+		return BindingBuilder.bind(alarmDataQueue()).to(exchange()).with(alarmMsgRouting);
+	}
 }

+ 0 - 3
src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java

@@ -9,7 +9,6 @@ import com.persagy.cache.AlarmInfoCache;
 import com.persagy.cache.AlarmLastTimeCache;
 import com.persagy.cache.AlarmRedisCache;
 import com.persagy.cache.CreatedAlarmIdsCache;
-import com.persagy.client.GroupNettyClient;
 import com.persagy.constant.RedisConstant;
 import com.persagy.entity.*;
 import com.persagy.entity.v2.AlarmCondition;
@@ -62,8 +61,6 @@ public class AlarmHandleServiceImpl {
     @Autowired
     AlarmInfoCache alarmInfoCache;
     @Autowired
-    GroupNettyClient groupNettyClient;
-    @Autowired
     AlarmQuartzService alarmQuartzService;
     @Autowired
     AlarmRecordRepository alarmRecordRepository;

+ 0 - 278
src/main/java/com/persagy/service/impl/NettyMsgHandler.java

@@ -1,278 +0,0 @@
-package com.persagy.service.impl;
-
-import cn.hutool.core.collection.CollectionUtil;
-import com.alibaba.fastjson.JSONObject;
-import com.alibaba.fastjson.TypeReference;
-import com.persagy.cache.AlarmInfoCache;
-import com.persagy.cache.AlarmLastTimeCache;
-import com.persagy.cache.CreatedAlarmIdsCache;
-import com.persagy.client.GroupNettyClient;
-import com.persagy.entity.NettyMessage;
-import com.persagy.entity.v2.AlarmCondition;
-import com.persagy.entity.v2.ObjConditionRel;
-import com.persagy.enumeration.NettyMsgTypeEnum;
-import com.persagy.utils.LockUtil;
-import io.netty.channel.ChannelHandlerContext;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Service;
-import org.springframework.util.CollectionUtils;
-
-import java.util.List;
-
-/**
- * Netty消息处理类
- *
- * @author lixing
- * @version V1.0 2021/10/25 8:34 下午
- **/
-@Service
-@Slf4j
-public class NettyMsgHandler {
-    @Autowired
-    public AlarmConditionServiceImpl alarmConditionService;
-    @Autowired
-    public GroupNettyClient groupNettyClient;
-    @Autowired
-    public CreatedAlarmIdsCache createdAlarmIdsCache;
-    @Autowired
-    public AlarmLastTimeCache alarmLastTimeCache;
-    @Autowired
-    public AlarmInfoCache alarmInfoCache;
-    @Value("${logging.level.com.persagy}")
-    private String logLevel;
-
-    /**
-     * 已接收回执
-     *
-     * @param channelHandlerContext 通道上下文
-     * @author lixing
-     * @version V1.0 2021/10/25 8:32 下午
-     */
-    public void acceptedReply(ChannelHandlerContext channelHandlerContext) {
-        NettyMessage response = new NettyMessage(groupNettyClient.projectId);
-        response.setOpCode(NettyMsgTypeEnum.ACCEPTED);
-        response.setRemark("已经收到消息");
-        channelHandlerContext.write(response.toString());
-    }
-
-    /**
-     * 缓存报警记录id
-     *
-     * @param message netty消息
-     * @author lixing
-     * @version V1.0 2021/10/25 8:27 下午
-     */
-    public void cacheRecordId(NettyMessage message) {
-        log.debug("云端完成报警记录创建");
-        // {"id":"","objId":"","itemCode":""}  id为报警记录ID
-        log.debug("返回报警记录id[{}]", message);
-        List content = message.getContent();
-        if (CollectionUtil.isNotEmpty(content)) {
-            JSONObject parseObject = JSONObject.parseObject(JSONObject.toJSONString(content.get(0)));
-            String alarmId = parseObject.getString("id");
-            // 将alarmId放入缓存中,用于后续判断报警是否完成创建
-            createdAlarmIdsCache.put(alarmId);
-            alarmLastTimeCache.setAlarmHasCreated(alarmId);
-        }
-    }
-
-    /**
-     * 缓存设备与报警条件的关联
-     *
-     * @param msg netty消息
-     * @author lixing
-     * @version V1.0 2021/10/25 8:27 下午
-     */
-    public void cacheAllObjConditionRel(Object msg) {
-        List<ObjConditionRel> content = getObjConditionRelList(msg);
-        if (CollectionUtils.isEmpty(content)) {
-            log.error("接收到的消息中报警条件与设备的关联关系为空");
-            return;
-        }
-        log.info("正在同步报警条件与设备的关联关系 -> 项目id:[{}], 同步条数[{}]",
-                content.get(0).getProjectId(), content.size());
-        try {
-            LockUtil.getInstance().lock.lock();
-            LockUtil.getInstance().setExecute(false);
-            alarmConditionService.cacheObjConditionRelList(content);
-            LockUtil.getInstance().setExecute(true);
-            LockUtil.getInstance().condition.signalAll();
-            log.info("同步报警条件与设备的关联关系完成 -> 项目id:[{}]",
-                    content.get(0).getProjectId());
-        } catch (Exception e) {
-            log.error("同步设备与报警条件关联关系发生异常", e);
-        } finally {
-            LockUtil.getInstance().lock.unlock();
-        }
-    }
-
-    /**
-     * 将消息解析为设备与报警条件关联关系列表
-     *
-     * @param msg netty消息
-     * @return 设备与报警条件关联关系列表
-     * @author lixing
-     * @version V1.0 2021/10/25 9:05 下午
-     */
-    private List<ObjConditionRel> getObjConditionRelList(Object msg) {
-        NettyMessage<ObjConditionRel> objConditionRelNettyMessage = JSONObject.parseObject(
-                String.valueOf(msg),
-                new TypeReference<NettyMessage<ObjConditionRel>>() {
-                });
-
-        return objConditionRelNettyMessage.getContent();
-    }
-
-    /**
-     * 日志记录当前缓存中的数据数量
-     *
-     * @author lixing
-     * @version V1.0 2021/10/28 9:58 上午
-     */
-    public void logCacheInfo() {
-        // 只在debug日志等级下才获取数据数量,避免预期之外的统计计算
-        if (!"debug".equals(logLevel)) {
-            return;
-        }
-        log.debug("当前缓存中报警条件数量:[{}]", alarmInfoCache.getCachedConditionCount());
-        log.debug("当前缓存中设备数量:[{}]", alarmInfoCache.getCachedObjCount());
-        log.debug("当前缓存中设备与报警条件关联关系数量:[{}]", alarmInfoCache.getCachedObjConditionRelCount());
-    }
-
-    /**
-     * 缓存报警条件
-     *
-     * @param msg 全量报警条件消息
-     * @author lixing
-     * @version V1.0 2021/10/22 3:21 下午
-     */
-    public void cacheAllAlarmConditions(Object msg) {
-        log.info("开始全量同步报警条件");
-        List<AlarmCondition> conditionList = getAlarmConditions(msg);
-        if (CollectionUtil.isNotEmpty(conditionList)) {
-            try {
-                LockUtil.getInstance().lock.lock();
-                LockUtil.getInstance().setExecute(false);
-                //加个等待,保证正在执行的逻辑执行成功
-                Thread.sleep(4000);
-                alarmConditionService.cacheAllConditions(conditionList);
-                log.info("全量同步报警条件完成");
-                LockUtil.getInstance().setExecute(true);
-                LockUtil.getInstance().condition.signalAll();
-            } catch (Exception e) {
-                log.error("全量同步报警条件发生异常", e);
-            } finally {
-                LockUtil.getInstance().lock.unlock();
-            }
-        }
-    }
-
-    /**
-     * 将消息解析为报警条件列表
-     *
-     * @param msg netty消息
-     * @return 报警条件列表
-     * @author lixing
-     * @version V1.0 2021/10/25 8:54 下午
-     */
-    private List<AlarmCondition> getAlarmConditions(Object msg) {
-        NettyMessage<AlarmCondition> alarmConditionNettyMessage = JSONObject.parseObject(
-                String.valueOf(msg),
-                new TypeReference<NettyMessage<AlarmCondition>>() {
-                });
-        return alarmConditionNettyMessage.getContent();
-    }
-
-    /**
-     * 将消息解析为报警条件
-     *
-     * @param msg netty消息
-     * @return 报警条件
-     * @author lixing
-     * @version V1.0 2021/10/25 8:54 下午
-     */
-    private AlarmCondition getAlarmCondition(Object msg) {
-        NettyMessage<AlarmCondition> alarmConditionNettyMessage = JSONObject.parseObject(
-                String.valueOf(msg),
-                new TypeReference<NettyMessage<AlarmCondition>>() {
-                });
-        List<AlarmCondition> content = alarmConditionNettyMessage.getContent();
-        if (CollectionUtils.isEmpty(content)) {
-            return null;
-        }
-        return content.get(0);
-    }
-
-
-    /**
-     * 缓存新增的报警条件
-     *
-     * @param msg 新增的报警条件
-     * @author lixing
-     * @version V1.0 2021/10/22 3:21 下午
-     */
-    public void cacheNewCondition(Object msg) {
-        AlarmCondition alarmCondition = getAlarmCondition(msg);
-        alarmConditionService.cacheNewCondition(alarmCondition);
-    }
-
-    /**
-     * 缓存更新的报警条件
-     *
-     * @param msg 更新的报警条件
-     * @author lixing
-     * @version V1.0 2021/10/22 3:21 下午
-     */
-    public void cacheUpdatedCondition(Object msg) {
-        AlarmCondition alarmCondition = getAlarmCondition(msg);
-        alarmConditionService.cacheUpdatedCondition(alarmCondition);
-    }
-
-    /**
-     * 移除缓存中的报警条件
-     *
-     * @param msg 要移除的报警条件
-     * @author lixing
-     * @version V1.0 2021/10/22 3:21 下午
-     */
-    public void removeCachedCondition(Object msg) {
-        AlarmCondition alarmCondition = getAlarmCondition(msg);
-        alarmConditionService.removeCachedCondition(alarmCondition);
-    }
-
-    /**
-     * 缓存新增的设备与报警条件关联关系
-     *
-     * @param msg 新增的设备与报警条件关联关系
-     * @author lixing
-     * @version V1.0 2021/10/22 3:21 下午
-     */
-    public void cacheNewObjConditionRel(Object msg) {
-        List<ObjConditionRel> objConditionRelList = getObjConditionRelList(msg);
-        if (CollectionUtils.isEmpty(objConditionRelList)) {
-            return;
-        }
-        for (ObjConditionRel objConditionRel : objConditionRelList) {
-            alarmConditionService.cacheNewObjConditionRel(objConditionRel);
-        }
-    }
-
-    /**
-     * 移除缓存中的设备与报警条件关联关系
-     *
-     * @param msg 要移除的关联
-     * @author lixing
-     * @version V1.0 2021/10/22 3:21 下午
-     */
-    public void removeCachedObjConditionRel(Object msg) {
-        List<ObjConditionRel> objConditionRelList = getObjConditionRelList(msg);
-        if (CollectionUtils.isEmpty(objConditionRelList)) {
-            return;
-        }
-        for (ObjConditionRel objConditionRel : objConditionRelList) {
-            alarmConditionService.removeCachedObjConditionRel(objConditionRel);
-        }
-    }
-}

+ 14 - 30
src/main/resources/application.yml

@@ -1,35 +1,34 @@
-alarm:
-  subsystem:
-    # 报警子系统ip
-    ip: 192.168.17.55
-#    ip: localhost
-  iotservice:
-    # 报警iot采集服务ip
-    ip: 192.168.17.55
-    port: 8081
-#    ip: localhost
+# 集团编码
+group:
+  code: WD
+# rabbitmq连接信息
 rabbitmq.host: 192.168.100.93
 rabbitmq.port: 9936
 rabbitmq.username: pbsage
 rabbitmq.password: pbsage123
+# redis连接信息
 redis.host: 192.168.100.93
 redis.port: 9944
 redis.database: 14
 redis.password.sentinel: test123
 redis.password.standalone: test123
 redis.sentinel.nodes: 39.102.43.179:9940
-
+# 数据库连接信息
+datasource.url: jdbc:mysql://192.168.100.94:9934/alarm-quartz?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai&autoReconnect=true&failOverReadOnly=false
+datasource.username: root
+datasource.password: persagy@2020
 server:
   port: 8101
+# -------------------------只需要修改分隔线以上的内容-----------------------------------------------
 spring:
   # 应用名称
   application:
     name: alarm-engine
   datasource:
-    url: jdbc:mysql://192.168.100.94:9934/alarm-quartz?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai&autoReconnect=true&failOverReadOnly=false
+    url: ${datasource.url}
     driver-class-name: com.mysql.jdbc.Driver  # mysql8.0以前使用com.mysql.jdbc.Driver
-    username: root
-    password: persagy@2020
+    username: ${datasource.username}
+    password: ${datasource.password}
     type: com.alibaba.druid.pool.DruidDataSource
     initial-size: 5  # 初始化大小
     min-idle: 5  # 最小
@@ -96,22 +95,7 @@ spring:
     publisher-confirms: true
     publisher-returns: true
     send-flag: true
-group:
-  code: WD   #标识哪个集团 比如万达使用WD, 华润使用HR
-  alarm:
-    #    host: 192.168.17.55    #netty IP
-    host: ${alarm.subsystem.ip}    #netty IP
-    port: 9986          #netty 端口9986
-terminal: #边缘端IOT采集程序地址
-  alarm: # 拼接后的地址为ws://host:port/suffix
-    compress: false    #采用的是压缩方式还是不压缩方式  true-压缩 false-不压缩
-      #    host: 192.168.17.55
-    host: ${alarm.iotservice.ip}
-    port: ${alarm.iotservice.port}
-    suffix: websocket/iot   #websocker后缀
-project:
-  iotid:   #iot使用项目ID,不包含PJ
-  id:    #项目ID
+
 system:
   id: system  #默认用户表示 ,用于报警记录的创建人
 #alarm: