package cn.bridgeli.websocket;import com.google.gson.Gson;import com.google.gson.GsonBuilder;import com.lagou.common.base.util.date.DateUtil;import com.lagou.platform.news.api.enumeration.PlatNewsCategoryType;import com.lagou.platform.news.web.dao.ext.model.PlatNewsVo;import com.lagou.platform.news.web.dao.ext.model.SearchCondition;import com.lagou.platform.news.web.quartz.impl.TimingJob;import com.lagou.platform.news.web.service.PlatNewsService;import org.apache.commons.lang.StringUtils;import org.json.simple.JSONArray;import org.json.simple.JSONObject;import org.quartz.*;import org.quartz.impl.StdSchedulerFactory;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import org.springframework.web.socket.TextMessage;import java.io.IOException;import java.util.Date;import java.util.List;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/*** @Description : 站內(nèi)消息監(jiān)聽器實(shí)現(xiàn)* @Date : 16-3-7*/@Componentpublic class NewsListenerImpl implements NewsListener{private static final Logger logger = LoggerFactory.getLogger(NewsListenerImpl.class);Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();//線程池private ExecutorService executorService = Executors.newCachedThreadPool();//任務(wù)調(diào)度private SchedulerFactory sf = new StdSchedulerFactory();@Autowiredprivate PlatNewsService platNewsService;@Overridepublic void afterPersist(PlatNewsVo platNewsVo) {logger.info("監(jiān)聽到有新消息添加。。。");logger.info("新消息為:"+gson.toJson(platNewsVo));//啟動(dòng)線程if(null != platNewsVo && !StringUtils.isBlank(platNewsVo.getCurrentoperatoremail())){//如果是定時(shí)消息if(platNewsVo.getNewsType() == PlatNewsCategoryType.TIMING_TIME.getCategoryId()){startTimingTask(platNewsVo); //定時(shí)推送}else{//立即推送executorService.execute(new AfterConnectionEstablishedTask(platNewsVo.getCurrentoperatoremail()));}}}@Overridepublic void afterConnectionEstablished(String email) {logger.info("建立websocket連接后推送新消息。。。");if(!StringUtils.isBlank(email)){executorService.execute(new AfterConnectionEstablishedTask(email));}}/*** @Description ： 如果新添加了定時(shí)消息，啟動(dòng)定時(shí)消息任務(wù)* @param platNewsVo*/private void startTimingTask(PlatNewsVo platNewsVo){logger.info("開始定時(shí)推送消息任務(wù)。。。");Date timingTime = platNewsVo.getTimingTime();if(null == timingTime){logger.info("定時(shí)消息時(shí)間為null。");return;}logger.info("定時(shí)推送任務(wù)時(shí)間為："+DateUtil.date2String(timingTime));JobDetail jobDetail= JobBuilder.newJob(TimingJob.class).withIdentity(platNewsVo.getCurrentoperatoremail()+"定時(shí)消息"+platNewsVo.getId(), "站內(nèi)消息").build();//傳遞參數(shù)jobDetail.getJobDataMap().put("platNewsService",platNewsService);jobDetail.getJobDataMap().put("userEmail",platNewsVo.getCurrentoperatoremail());Trigger trigger= TriggerBuilder.newTrigger().withIdentity("定時(shí)消息觸發(fā)"+platNewsVo.getId(), "站內(nèi)消息").startAt(timingTime).withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(0) //時(shí)間間隔.withRepeatCount(0) //重復(fù)次數(shù)).build();//啟動(dòng)定時(shí)任務(wù)try {Scheduler sched = sf.getScheduler();sched.scheduleJob(jobDetail,trigger);if(!sched.isShutdown()){sched.start();}} catch (SchedulerException e) {logger.info(e.toString());}logger.info("完成開啟定時(shí)推送消息任務(wù)。。。");}/*** @Description : 建立websocket鏈接后的推送線程*/class AfterConnectionEstablishedTask implements Runnable{String email ;public AfterConnectionEstablishedTask(String email){this.email = email;}@Overridepublic void run() {logger.info("開始推送消息給用戶:"+email+"。。。");if(!StringUtils.isBlank(email)){SearchCondition searchCondition = new SearchCondition();searchCondition.setOperatorEmail(email);JSONArray jsonArray = new JSONArray();for(PlatNewsCategoryType type : PlatNewsCategoryType.values()){searchCondition.setTypeId(type.getCategoryId());int count = platNewsService.countPlatNewsByExample(searchCondition);JSONObject object = new JSONObject();object.put("name",type.name());object.put("description",type.getDescription());object.put("count",count);jsonArray.add(object);}if(null != jsonArray && jsonArray.size()>0){UserSocketVo userSocketVo = WSSessionLocalCache.get(email);TextMessage reMessage = new TextMessage(gson.toJson(jsonArray));try {if(null != userSocketVo){//推送消息userSocketVo.getWebSocketSession().sendMessage(reMessage);//更新推送時(shí)間userSocketVo.setLastSendTime(DateUtil.getNowDate());logger.info("完成推送新消息給用戶:"+userSocketVo.getUserEmail()+"。。。");}} catch (IOException e) {logger.error(e.toString());logger.info("站內(nèi)消息推送失敗。。。"+e.toString());}}}logger.info("結(jié)束推送消息給"+email+"。。。");}}}

userSocketVo.getWebSocketSession().sendMessage(reMessage);