Spring整合websocket整合应用示例（下）

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 : 站内消息监听器实现* @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();//任务调度private SchedulerFactory sf = new StdSchedulerFactory();@Autowiredprivate PlatNewsService platNewsService;@Overridepublic void afterPersist(PlatNewsVo platNewsVo) {logger.info("监听到有新消息添加。。。");logger.info("新消息为:"+gson.toJson(platNewsVo));//启动线程if(null != platNewsVo && !StringUtils.isBlank(platNewsVo.getCurrentoperatoremail())){//如果是定时消息if(platNewsVo.getNewsType() == PlatNewsCategoryType.TIMING_TIME.getCategoryId()){startTimingTask(platNewsVo); //定时推送}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 ： 如果新添加了定时消息，启动定时消息任务* @param platNewsVo*/private void startTimingTask(PlatNewsVo platNewsVo){logger.info("开始定时推送消息任务。。。");Date timingTime = platNewsVo.getTimingTime();if(null == timingTime){logger.info("定时消息时间为null。");return;}logger.info("定时推送任务时间为："+DateUtil.date2String(timingTime));JobDetail jobDetail= JobBuilder.newJob(TimingJob.class).withIdentity(platNewsVo.getCurrentoperatoremail()+"定时消息"+platNewsVo.getId(), "站内消息").build();//传递参数jobDetail.getJobDataMap().put("platNewsService",platNewsService);jobDetail.getJobDataMap().put("userEmail",platNewsVo.getCurrentoperatoremail());Trigger trigger= TriggerBuilder.newTrigger().withIdentity("定时消息触发"+platNewsVo.getId(), "站内消息").startAt(timingTime).withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(0) //时间间隔.withRepeatCount(0) //重复次数).build();//启动定时任务try {Scheduler sched = sf.getScheduler();sched.scheduleJob(jobDetail,trigger);if(!sched.isShutdown()){sched.start();}} catch (SchedulerException e) {logger.info(e.toString());}logger.info("完成开启定时推送消息任务。。。");}/*** @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);//更新推送时间userSocketVo.setLastSendTime(DateUtil.getNowDate());logger.info("完成推送新消息给用户:"+userSocketVo.getUserEmail()+"。。。");}} catch (IOException e) {logger.error(e.toString());logger.info("站内消息推送失败。。。"+e.toString());}}}logger.info("结束推送消息给"+email+"。。。");}}}

userSocketVo.getWebSocketSession().sendMessage(reMessage);