长连接

在上篇中，我们已经实现了websocket，但还有一个核心的业务实现类没有实现，这里我们就实现这个业务核心类，因为老夫参与的这个系统使用websocket发送消息，所以其实现就是如何发送消息了。 NewsListenerImpl的实现 package cn.bridgeli.websocket; import com.google.gson.Gson; import com.google.gson.GsonBuilder; import cn.bridgeli.DateUtil; import cn.bridgeli.enumeration.PlatNewsCategoryType; import cn.bridgeli.model.PlatNewsVo; import cn.bridgeli.model.SearchCondition; import cn.bridgeli.quartz.impl.TimingJob; import cn.bridgeli.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 */ @Component public 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(); @Autowired private PlatNewsService platNewsService; @Override public 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())); } } } @Override public 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; } @Override public 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+"。。。"); } } } 这个类就是websocket的核心业务的实现，其具体肯定和业务相关，由于业务的不同，实现肯定不同，因为老夫参与的系统是发送消息，所以里面最核心的一句就是： ...

嗯，这次真的仅仅是一个入门教程，因为老夫表示自己也不会。近期老夫参与开发公司的一个CRM系统，系统中有很多消息的推送，由一个同事负责，其用到了websocket技术，老夫比较感兴趣，删繁就简，整理了一个教程，留作自己笔记，因很多原理老夫也是不甚了了，以备将来用到了有资料可查。 maven依赖 <dependency> <groupId>javax.servlet</groupId> <artifactId>javax.servlet-api</artifactId> <version>3.1.0</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-core</artifactId> <version>2.3.0</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.3.0</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-websocket</artifactId> <version>4.0.1.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-messaging</artifactId> <version>4.0.1.RELEASE</version> </dependency> spring-servlet的配置 <?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:mvc="http://www.springframework.org/schema/mvc" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:websocket="http://www.springframework.org/schema/websocket" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-3.1.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.1.xsd http://www.springframework.org/schema/websocket http://www.springframework.org/schema/websocket/spring-websocket.xsd"> &#8230;&#8230; <!&#8211; websocket &#8211;> <bean id="websocket" class="cn.bridgeli.websocket.WebsocketEndPoint"/> <websocket:handlers> <websocket:mapping path="/websocket" handler="websocket"/> <websocket:handshake-interceptors> <bean class="cn.bridgeli.websocket.HandshakeInterceptor"/> </websocket:handshake-interceptors> </websocket:handlers> </beans> 其中，path对应的路径就是前段通过ws协议调的接口路径 HandshakeInterceptor的实现 package cn.bridgeli.websocket; import cn.bridgeli.utils.UserManager; import cn.bridgeli.util.DateUtil; import cn.bridgeli.sharesession.UserInfo; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; import org.springframework.web.context.request.RequestContextHolder; import org.springframework.web.context.request.ServletRequestAttributes; import org.springframework.web.socket.WebSocketHandler; import org.springframework.web.socket.server.support.HttpSessionHandshakeInterceptor; import java.util.Date; import java.util.Map; /** * @Description :创建握手（handshake）接口 * @Date : 16-3-3 */ public class HandshakeInterceptor extends HttpSessionHandshakeInterceptor{ private static final Logger logger = LoggerFactory.getLogger(HandshakeInterceptor.class); @Override public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception { logger.info("建立握手前&#8230;"); ServletRequestAttributes attrs = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes(); UserInfo currUser = UserManager.getSessionUser(attrs.getRequest()); UserSocketVo userSocketVo = new UserSocketVo(); String email= ""; if(null != currUser){ email = currUser.getEmail(); } if(StringUtils.isBlank(email)){ email = DateUtil.date2String(new Date()); } userSocketVo.setUserEmail(email); attributes.put("SESSION_USER", userSocketVo); return super.beforeHandshake(request, response, wsHandler, attributes); } @Override public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception ex) { logger.info("建立握手后&#8230;"); super.afterHandshake(request, response, wsHandler, ex); } } 因为老夫不是很懂，所以最大限度的保留原代码，这其实就是从单点登录中取出当前登录用户，转成UserSocketVo对象，放到Map中。所以接下来我们看看UserSocketVo对象的定义 ...