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

在spring整合websocket整合应用示例（上）文章中，我们已经实现了websocket，但还有一个核心的业务实现类没有实现，这里我们就实现这个业务核心类，因为老夫参与的这个系统使用websocket发送消息，所以其实现就是如何发送消息了。

7. newslistenerimpl的实现

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
*/
@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的核心业务的实现，其具体肯定和业务相关，由于业务的不同，实现肯定不同，因为老夫参与的系统是发送消息，所以里面最核心的一句就是：

usersocketvo.getwebsocketsession().sendmessage(remessage);

通过websocketsession的sendmessage方法把我们的消息发送出去。另外，这主要是后端的实现，至于前端的实现，因为老夫是后端程序猿比较关注后端，所以前端就不多做介绍了，大家可以自己去网上查资料。最后需要说明的是，老夫之前搜一些学习资料的时候，发现老夫该同事的写法和有一篇文章几乎一样，我想该同事应该是参考了这篇文章，所以列在下面，算作参考资料。