//這里忽略報(bào)警系統(tǒng)代碼，報(bào)警系統(tǒng)推送報(bào)警消息是通過(guò)RocketMQ實(shí)現(xiàn)
topic: alarm-camera

@Configuration
public class RocketMqConsumer {
    private static Logger logger = LogManager.getLogger(RocketMqConsumer.class);


    public void init() {
        pullAlarm();
        logger.warn("rocketmq拉取告警數(shù)據(jù)成功！");
    }

    /**
     * pullAlarm:拉取告警源數(shù)據(jù)。
     * @author liaokh
     * @since JDK 1.8
     */
    public static void pullAlarm() {
        new Thread() {
            public void run() {
                logger.warn("---------開(kāi)始消費(fèi)報(bào)警broker---------");
                try {
                    // 聲明并初始化一個(gè)consumer
                    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocketmq-consumer-dev-camera" + "-alarm");

                    // 同樣也要設(shè)置NameServer地址
                    consumer.setNamesrvAddr("我的RocketMQ服務(wù)器地址");

                    // 廣播模式 當(dāng) Consumer 使用廣播模式時(shí)，每條消息都會(huì)被 Consumer 集群內(nèi)所有的 Consumer 實(shí)例消費(fèi)一次。
                    consumer.setMessageModel(MessageModel.BROADCASTING);

                    // 這里設(shè)置的是一個(gè)consumer的消費(fèi)策略
                    // CONSUME_FROM_LAST_OFFSET 默認(rèn)策略，從該隊(duì)列最尾開(kāi)始消費(fèi)，即跳過(guò)歷史消息
                    // CONSUME_FROM_FIRST_OFFSET 從隊(duì)列最開(kāi)始開(kāi)始消費(fèi)，即歷史消息（還儲(chǔ)存在broker的）全部消費(fèi)一遍
                    // CONSUME_FROM_TIMESTAMP 從某個(gè)時(shí)間點(diǎn)開(kāi)始消費(fèi)，和setConsumeTimestamp()配合使用，默認(rèn)是半個(gè)小時(shí)以前
                    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

                    // 設(shè)置consumer所訂閱的Topic和Tag，*代表全部的Tag
                    consumer.subscribe("alarm-camera", "*");

                    // 設(shè)置一個(gè)Listener，主要進(jìn)行消息的邏輯處理
                    consumer.registerMessageListener(new MessageListenerConcurrently() {

                        @Override
                        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                        ConsumeConcurrentlyContext context) {

                            for (MessageExt msg : msgs) {
                                try {
                                    String tag = msg.getTags();
                                    String alarmJson = new String(msg.getBody());
                                    logger.warn("收到alarm-camera數(shù)據(jù)：tag:" + tag + " alarmJson:" + alarmJson);
                                    CameraAlarmResp resultAlarm = new CameraAlarmResp();
                                    AlarmMQResp alarm = JSON.parseObject(alarmJson, AlarmMQResp.class);
                                    //查看當(dāng)前告警類型是否在該用戶配置的列表中
                                    //根據(jù)攝像頭設(shè)備號(hào)獲取用戶信息
                                    Camera cameraEntity = Utils.getCameraById(alarm.getCameraId());  //這種核心數(shù)據(jù)也可以加載到緩存中
                                    UserAlarm userAlarm = Utils.getUserAlarm(cameraEntity.getUserId());
                                    if (userAlarm == null || StringUtils.isBlank(userAlarm.getAlarmIds())){
                                        logger.error("設(shè)備號(hào)" + alarm.getId() + "的用戶未配置需要推送的告警類型");
                                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                                    }
                                    boolean isReturn = true;
                                    //獲取該用戶告警列表過(guò)濾
                                    String[] userAlarmArr = userAlarm.getAlarmIds().split(",");
                                    for (String s : userAlarmArr) {
                                        if (alarm.getAlarmType().equals(s)){  //說(shuō)明需要推送
                                            isReturn = false;
                                        }
                                    }
                                    if (isReturn){
                                        //匹配則該告警不需要推送，直接消費(fèi)成功
                                        logger.warn("該設(shè)備號(hào)的用戶未配置需要推送的告警類型");
                                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                                    }

                                    WebSocket webSocket = SpringUtil.getBean(WebSocket.class);
                                    //創(chuàng)建業(yè)務(wù)消息信息
                                    JSONObject obj = new JSONObject();
                                    obj.put("cmd", "alarm");//業(yè)務(wù)類型
                                    obj.put("msgId", msg.getMsgId());//消息id
                                    obj.put("msgTxt", JSON.toJSONString(alarm));//消息內(nèi)容
                                    //單個(gè)用戶發(fā)送
                                    webSocket.sendOneMessage(alarm.getUserId(), obj.toJSONString());
                                } catch (Exception e) {
                                    logger.error("請(qǐng)求異常", e);
                                }
                            }
                            // 返回消費(fèi)狀態(tài)，消費(fèi)成功
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                    });

                    // 調(diào)用start()方法啟動(dòng)consumer
                    consumer.start();
                    logger.warn("rocketmq消費(fèi)者創(chuàng)建成功");
                } catch (Exception e) {
                    logger.error("請(qǐng)求異常", e);
                }
            }
        }.start();
    }
}

@Component
public class Utils {
	private static final Logger logger = LoggerFactory.getLogger(GuavaCacheUtils.class);
	
	/**
	* 獲取用戶告警配置信息
	*/
    public static UserAlarm getUserAlarm(String userId){
		if(StringUtils.isBlank(userId)){
			return null;
		}
		UserAlarm userAlarm = null;
		try {
			userAlarm = GuavaCacheUtils.userAlarmCache.get(userId).orNull();
			if(null == userAlarm){
				GuavaCacheUtils.userAlarmCache.invalidate(userId);  //清除Guava的緩存
                //嘗試從Redis中獲取
                String userAlarmJson = RedisUtils.hget("alarm_camera", userId);
				userAlarm = JSON.parseObject(userAlarmJson, UserAlarm.class);
			}
		} catch (ExecutionException e) {
			logger.error("獲取用戶配置緩存異常",e);
		}
		return userAlarm;
	}

	/**
	* 獲取攝像頭信息
	*/
	public static Camera getCameraById(String cameraId){
		Camera camera= null;
		try {
			camera= GuavaCacheUtils.cameraCache.get(cameraId).orNull();
		} catch (ExecutionException e) {
			logger.error("獲取設(shè)備數(shù)據(jù)異常異常",e);
		}
		return device;
	}
}

/**
 * ClassName:GuavaCacheUtils <br/>
 * @version
 * @since JDK 1.8
 * @see java（jvm）緩存存儲(chǔ)
 */
@Component
public class GuavaCacheUtils {
   private static final Logger logger = LoggerFactory.getLogger(GuavaCacheUtils.class);
	 /**
	 * 用戶告警推送列表緩存
	 *
	 * expireAfterWrite：10分鐘內(nèi)沒(méi)有更新將被回收重新獲取
	 *
	 * load：獲取緩存為空時(shí)執(zhí)行（去數(shù)據(jù)庫(kù)查詢并將結(jié)果放入緩存）
	 */
	public static LoadingCache<String, Optional<UserAlarm>> userAlarmCache = CacheBuilder.newBuilder()
			.expireAfterAccess(10, TimeUnit.MINUTES).build(new CacheLoader<String, Optional<UserAlarm>>() {
				@Override
				public Optional<UserAlarm> load(String userId) throws Exception {
                    UserAlarm userAlarm = SpringUtil.getBean(UserAlarmService.class)
                        .getOne(new LambdaQueryWrapper<UserAlarm>()
									.eq(UserAlarm::getUserId,userId));
					return Optional.fromNullable(userAlarm);
				}
			});

	/**
	 * 攝像頭設(shè)備信息緩存
	 */
	public static LoadingCache<String, Optional<Camera>> cameraCache = CacheBuilder.newBuilder()
			.expireAfterAccess(10, TimeUnit.MINUTES)
			.build(new CacheLoader<String, Optional<Camera>>() {
				@Override
				public Optional<Camera> load(String cameraId) throws Exception {
					String cameraJson = RedisUtils.hget("camera", cameraId);
					Cameracamera= JSON.parseObject(cameraJson, Camera.class);
					return Optional.fromNullable(camera);
				}
			});
}

@Service
public class UserAlarmServiceImpl extends ServiceImpl<UserAlarmMapper, UserAlarm> implements UserAlarmService{
    //新增用戶告警配置
    @Override
    public String insert(UserAlarm userAlarm){
        try{
            this.save(userAlarm);
            //隨即存入Redis
            RedisUtil.hset("alarm_camera",userAlarm.getUserId,userAlarm);
        } catch (Exception e) {
            return "失敗啦";
        }
        return "成功咯";
    }
    
    //修改用戶告警配置
    @Override
    public String update(UserAlarm userAlarm){
        try{
            UpdateWrapper<UserAlarm> wrapper = new UpdateWrapper();
            wrapper.set("alarmType",userAlarm.getAlarmType());
                .eq("user_id",userAlarm.getUserId);
            this.save(userAlarm);
            //隨即更新Redis
            RedisUtil.hset("alarm_camera",userAlarm.getUserId,userAlarm);
        } catch (Exception e) {
            return "失敗啦";
        }
        return "成功咯";
    }
}

/**
 * ClassName:GuavaCacheUtils <br/>
 * @version
 * @since JDK 1.8
 * @see java（jvm）緩存存儲(chǔ)
 */
@Component
public class GuavaCacheUtils {
   private static final Logger logger = LoggerFactory.getLogger(GuavaCacheUtils.class);

   /**
    * LoadingCache登錄緩存
    * 鏈?zhǔn)秸{(diào)用
    * removalListener：設(shè)置緩存被移除后的監(jiān)聽(tīng)任務(wù)
    * build：構(gòu)建對(duì)象
    */
   public static LoadingCache<String, Optional<User>> loginCache = CacheBuilder.newBuilder()
         .expireAfterAccess(720, TimeUnit.MINUTES).removalListener(new MyRemovalListener())
         .build(new CacheLoader<String, Optional<User>>() {
            @Override
            public Optional<User> load(String token) throws Exception {
               User user = null;
               try {
                   //到redis中匹配
                  String loginJson = RedisUtils.get(token);
                  user = JSON.parseObject(loginJson, User.class);
               } catch (Exception e) {
                  logger.error("登錄緩存查詢異常", e);
               }
               return Optional.fromNullable(user);
            }
         });

    /**
    * MyRemovalListener自定義緩存移除監(jiān)聽(tīng)器，需要實(shí)現(xiàn)RemovalListener接口并實(shí)現(xiàn)RemovalListener<K,V>接口，K，V為key和value的泛型
    * Optional：主要用于解決空指針異常，簡(jiǎn)潔判空
    * notification.getCause()：監(jiān)聽(tīng)到的緩存失效原因
    */
   private static class MyRemovalListener implements RemovalListener<String, Optional<User>> {
      @Override
      public void onRemoval(RemovalNotification<String, Optional<User>> notification) {
         if (notification.getCause().toString().equals("EXPIRED")) {
            String token = notification.getKey();
            RedisUtils.del(0,token);
         }
      }
   }
}