WebSocket实时推送消息

一个小型项目，但是中间涉及到一个gps定位推送的需求，找了一些资料，在此做一个小结

1. 需要做到实时后台推送到前端

2.前端自动关闭或者主动关闭都不影响下次正常上线推送

在线客户端：http://www.websocket-test.com/

中间需要推送的数据存储到redis中了，因项目的不同可以做相对于的调整。

3.定时异步操作

后台通信处理，涉及连接客户端后对接的消息处理

gps
type=2&013320397086
type=2&013502892975
type=2&015391619781
type=2&013344952593
type=2&013798501105
013798501105
000000000000
alarmData
type=1&014092603014
type=1&015391619781
type=1&014092603008
type=1&014080602002

业务代码如下： 

package exsun.bigdata.hbasestorage.websocket;
import com.alibaba.dubbo.config.annotation.Service;
import exsun.bigdata.hbasestorage.config.RedisConfig;
import org.java_websocket.WebSocket;
import org.java_websocket.handshake.ClientHandshake;
import org.java_websocket.server.WebSocketServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@Service
public class WsServer extends WebSocketServer {
private static final Logger logger = LoggerFactory.getLogger(WsServer.class);
public static Long DEVICE_ALARM_SLEEP_TIME = 10L;
public static Long GPS_SLEEP_TIME = 5L;
//initialDelay是说系统启动后，需要等待多久才开始执行
public static Long INIT_DELAY = 1L;
public static String DeviceAlarmtableName = "DeviceAlarmHbaseFactory";
public static String GpsDatatableName = "GpsDataStorageFactory";
private static final Map GpsDataMap = new HashMap();
private static final Map DeviceAlarmMap = new HashMap();
private static RedisConfig redisCOnfig= new RedisConfig();
private static int userNum = 0;
public WsServer(int port) {
super(new InetSocketAddress(port));
}
public WsServer(InetSocketAddress address) {
super(address);
}
public static void setRedisConfig(RedisConfig redis) {
redisCOnfig= redis;
}
@Override
public void onOpen(WebSocket conn, ClientHandshake handshake) {
// ws连接的时候触发的代码，onOpen中我们不做任何操作
logger.info(" onOpen start: ");
WsPool.addUser(userNum+"",conn);
userNum++;
}
@Override
public void onClose(WebSocket conn, int code, String reason, boolean remote) {
//断开连接时候触发代码
userLeave(conn);
userNum--;
Future Gpsfuture = GpsDataMap.get(conn);
if (null != conn && null != Gpsfuture) {
Gpsfuture.cancel(true);
}
Future Devicfuture = DeviceAlarmMap.get(conn);
if (null != conn && null != Devicfuture) {
Devicfuture.cancel(true);
}
}
@Override
public void onMessage(WebSocket conn, String message) {
logger.info( " onMessage: " + message);
if(message == null || "".equals(message)){
WsPool.sendMessageToUser(conn, "500");
}
String dvo= message.substring((message.lastIndexOf("&")>0) ? (message.lastIndexOf("&")+1):message.length());
String type=message.substring(0,((message.indexOf("&")>0) ? message.indexOf("&"): message.length()));
if (("type=1").equals(type)) {//告警
//创建一个线程池，它可安排在给定延迟后运行命令或者定期地执行。
//参数：corePoolSize - 池中所保存的线程数，即使线程是空闲的也包括在内。
//返回：新创建的安排线程池
ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
// 从现在开始1秒钟之后，每隔10秒钟执行一次告警推送
Future future = service.scheduleAtFixedRate(new DynamicTiminTasks(conn,dvo,redisConfig,DeviceAlarmtableName), INIT_DELAY, DEVICE_ALARM_SLEEP_TIME, TimeUnit.SECONDS);
DeviceAlarmMap.put(conn,future);
} else if (("type=2").equals(type)) {//gps实时数据
//以固定周期频率执行任务 创建一个线程执行
ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
// 从现在开始1秒钟之后，每隔5秒钟执行一次GPS推送
Future future = service.scheduleAtFixedRate(new DynamicTiminTasks(conn,dvo,redisConfig,GpsDatatableName), INIT_DELAY, GPS_SLEEP_TIME, TimeUnit.SECONDS);
GpsDataMap.put(conn,future);
}
}
@Override
public void onError(WebSocket conn, Exception ex) {
//错误时候触发的代码
logger.error(" onError on error");
ex.printStackTrace();
}
/**
* 去除掉失效的websocket链接
* @param conn
*/
private void userLeave(WebSocket conn){
WsPool.removeUser(conn);
userNum--;
Future Gpsfuture = GpsDataMap.get(conn);
if (null != conn && null != Gpsfuture) {
Gpsfuture.cancel(true);
}
Future Devicfuture = DeviceAlarmMap.get(conn);
if (null != conn && null != Devicfuture) {
Devicfuture.cancel(true);
}
}
/**
* 将websocket加入用户池
* @param conn
* @param userName
*/
private void userJoin(WebSocket conn,String userName){
WsPool.addUser(userName, conn);
userNum++;
}
public static void start(int port){
WsServer s = new WsServer(port);
s.start();
}
//推送类型&推送设备编号
public static void main1111(String[] args) {
WsServer.start(8887);
}
}

实体发送如下：

package exsun.bigdata.hbasestorage.websocket;
import org.java_websocket.WebSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
public class WsPool {
private static final Logger logger = LoggerFactory.getLogger(WsPool.class);
private static final Map wsUserMap = new HashMap();
public static final Integer OK = 0;
public static final Integer ERROR = 1;
/**
* 通过websocket连接获取其对应的用户
*
* @param conn
* @return
*/
public static String getUserByWs(WebSocket conn) {
return wsUserMap.get(conn);
}
/**
* 根据userName获取WebSocket,这是一个list,此处取第一个
* 因为有可能多个websocket对应一个userName（但一般是只有一个，因为在close方法中，我们将失效的websocket连接去除了）
*
* @param
*/
public static WebSocket getWsByUser(String userName) {
Set keySet = wsUserMap.keySet();
synchronized (keySet) {
for (WebSocket conn : keySet) {
String cuser = wsUserMap.get(conn);
if (cuser.equals(userName)) {
return conn;
}
}
}
return null;
}
/**
* 向连接池中添加连接
*
* @param
*/
public static void addUser(String userName, WebSocket conn) {
wsUserMap.put(conn, userName); // 添加连接
}
/**
* 获取所有连接池中的用户，因为set是不允许重复的，所以可以得到无重复的user数组
*
* @return
*/
public static Collection getOnlineUser() {
List setUsers = new ArrayList();
Collection setUser = wsUserMap.values();
for (String u : setUser) {
setUsers.add(u);
logger.error(" getOnlineUser setUser： " + setUser);
}
return setUsers;
}
/**
* 移除连接池中的连接
*
* @param
*/
public static boolean removeUser(WebSocket conn) {
if (wsUserMap.containsKey(conn)) {
wsUserMap.remove(conn); // 移除连接
return true;
} else {
return false;
}
}
/**
* 向特定的用户发送数据
*
* @param
* @param message
*/
public static Integer sendMessageToUser(WebSocket conn, String message) {
if (null != conn && null != wsUserMap.get(conn)) {
logger.info(" sendMessageToUser message： " + message);
conn.send(message);
return OK;
}else {
logger.error(" sendMessageToUser conn is null");
return ERROR;
}
}
/**
* 向所有的用户发送消息
*
* @param message
*/
public static void sendMessageToAll(String message) {
Set keySet = wsUserMap.keySet();
synchronized (keySet) {
for (WebSocket conn : keySet) {
String user = wsUserMap.get(conn);
if (user != null) {
logger.info(" sendMessageToAll message： " + message);
conn.send(message);
}
}
}
}
}


定时任务异步处理如下：

package exsun.bigdata.hbasestorage.websocket;
import exsun.bigdata.bigdatadata.entity.DeviceAlarmData;
import exsun.jt808.data.upload_data.GpsData;
import exsun.bigdata.hbasestorage.config.RedisConfig;
import org.java_websocket.WebSocket;
import org.springframework.data.redis.core.HashOperations;
class DynamicTiminTasks implements Runnable {
private WebSocket conn;
private String dvo;
private RedisConfig redisConfig;
private String sendType;
DynamicTiminTasks(WebSocket conn,String dvo,RedisConfig redisConfig,String sendType) {
this.cOnn= conn;
this.dvo = dvo;
this.redisCOnfig= redisConfig;
this.sendType = sendType;
}
@Override
public void run() {
if (sendType.equals(WsServer.GpsDatatableName)) {
HashOperations hashOperatiOns= redisConfig.redisTemplate().opsForHash();
GpsData gpsData = hashOperations.get(WsServer.GpsDatatableName, dvo);
String gpsdataStr = "GpsData(lat:" + gpsData.getLat() + '\'' + "lng:" + gpsData.getLng() + '\'' + " hgt=" + gpsData.getHgt() + '\'' +
"spd=" + gpsData.getSpd() + '\'' + ", dre=" + gpsData.getDre() + '\'' + ", gte=" + gpsData.getGte() + '\'' + ", gdt=" + gpsData.getGdt() + '\'' +
", mie=" + gpsData.getMie() + '\'' + ", poi=" + gpsData.getPoi() + '\'' + ", alm=" + gpsData.getAlm() + '\'' + ", amm=" + gpsData.getAmm() + '\'' +
", sts=" + gpsData.getSts() + '\'' + ", stm=" + gpsData.getStm() + '\'' + ", pop=" + gpsData.getPop() + '\'' + ", cnt=" + gpsData.getCnt() + '\'' +
", Ons=" + gpsData.getOns() + ", dvo=" + gpsData.getDvo() + ")";
WsPool.sendMessageToUser(conn, gpsdataStr);
}else if (sendType.equals(WsServer.DeviceAlarmtableName)) {
HashOperations hashOperatiOns= redisConfig.redisTemplate().opsForHash();
DeviceAlarmData deviceAlarmData = hashOperations.get(WsServer.DeviceAlarmtableName, dvo);
WsPool.sendMessageToUser(conn, deviceAlarmData.toString());
}
}
}


 代码中带有部分业务属性，有需要的朋友可以结合实际情况进一步简化操作！