提示：长连接，采用了WebSocket，Redis，Quartz，SpringBoot等技术

文章目录

• 前言
• 一、搭建环境
• • 1.1 引入依赖和配置环境
  • 1.2 html测试
• 二、开发
• • 2.1 启动类
  • 2.2 WebSocket
  • • 2.2.1 WebSocket配置
    • 2.2.2 WebSocket使用
    • 2.2.3 测试结果
  • 2.3 Redis
  • • 2.3.1 Redis配置
    • 2.3.2 Redis封装成工具类
    • 2.3.3 Redis常量记录
  • 2.4 Quartz
  • • 2.4.1 Quartz配置
    • 2.4.2 Job
    • 2.4.2 其他
  • 2.5 实体类
  • • 2.5.1返回的vo
    • 2.5.2 R
  • 2.6 其他
  • • 2.6.1 Controller
    • 2.6.2 工具类
• 三、总结

前言

在web开发中，服务端和客户端需要一条长连接，用于消息的推送下发。
使用场景：

1. Android上服务型 APP，需要一条长连接的消息链路，当手机端或其他端触发了数据的更改，需要实时通知APP，比如微信APP有个单独的长连接进程，弹出一个通知窗口；

• 优点：有一个稳定的长连接，便于消息的下发；进程活着，可操作的空间增大，比如拉活其他APP。
• 缺点：对于Android，需要考虑一下几个点：可能需要单独开一个push进程，占用资源增加；确保APP的保活，不被系统杀死；设置心跳；断开重连等。

3. 聊天室：一个发送，一个接受。
4. 轮询：采用轮询的方式，每隔n秒钟主动请求服务端接口，获取更新的内容。

• 优点：实现简单，间隔时间http请求
• 缺点：控制间隔参数是一个经验值；当机器增多时，不断的请求调用同一个接口，增加了服务端的压力。随着用户数量和设备的增加，轮询接口调用的百分比数量急剧增加。

5. …

一、搭建环境

1.1 引入依赖和配置环境

在pom.xml中引入依赖

4.0.0org.ympushserver1.0-SNAPSHOT882.7.2org.springframework.bootspring-boot-starter-weborg.projectlomboklombokorg.springframework.bootspring-boot-starter-websocketorg.springframework.bootspring-boot-starter-quartzorg.springframework.bootspring-boot-devtoolsorg.springframework.bootspring-boot-starter-data-redisorg.apache.commonscommons-pool2cn.hutoolhutool-all5.8.9com.google.code.gsongsonorg.springframework.bootspring-boot-dependencies${spring-boot.version}pomimportsrc/main/java**/*.xml**/*.yml**/*.htmltruesrc/main/resources**/*.xml**/*.yml**/*.vm**/*.txt**/*.html/static/true

application.yml中配置：

server:port: 8081spring:application:name: PushServerredis:host: localhostport: 6379lettuce:pool:max-idle: 16max-active: 32min-idle: 8devtools:restart:enabled: true

1.2 html测试

测试路径：http://localhost:8081/push_client.html
链接socket url: ws://127.0.0.1:8081/websocket/"+userId

 WebSocket


Please input text in the fields.



User Id: Send Message: Receive Message: 
开启Socket发送消息关闭Socket

二、开发

2.1 启动类

代码如下：

package com.ym.pushserver;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;/*** @Author: Yangmiao* @Date: 2023/3/9 19:09* @Desc:*/
@SpringBootApplication
@EnableScheduling
public class PushServerApplication {public static void main(String[] args) {SpringApplication.run(PushServerApplication.class,args);}
}

2.2 WebSocket

2.2.1 WebSocket配置

package com.ym.pushserver.config;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;/*** @Author: Yangmiao* @Date: 2023/3/9 19:20* @Desc:*/
@Configuration
public class WebSocketConfig {@Beanpublic ServerEndpointExporter serverEndpointExporter(){return new ServerEndpointExporter();}
}

2.2.2 WebSocket使用

使用websocket建立连接，并且监听接口，引入Redis
调用路径：/websocket/{userId}

package com.ym.pushserver.service;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.util.StrUtil;
import com.ym.pushserver.entity.PushCommonVo;
import com.ym.pushserver.entity.R;
import com.ym.pushserver.redis.RedisConstant;
import com.ym.pushserver.redis.RedisUtil;
import com.ym.pushserver.utils.JsonUtil;
import com.ym.pushserver.utils.StrHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;/*** @Author: Yangmiao* @Date: 2023/3/9 19:23* @Desc: 发送消息*/
@Component
@ServerEndpoint("/websocket/{userId}")
public class WebSocketServer {@Autowiredprivate RedisUtil redisUtil;private static Logger log = LoggerFactory.getLogger(WebSocketServer.class);/*** 统计连接数量*/private static AtomicInteger atomicInteger = new AtomicInteger();/*** 根据userid记录对应的socket，TODO 后期加入设备id,防止统一设备一直重试*/private static Map map= new ConcurrentHashMap<>();private Session session;private String userId = "";/*** 建立连接* @param session* @param userId 用户id*/@OnOpenpublic void onOpen(Session session, @PathParam(value = "userId")String userId){try {this.session = session;this.userId = userId;map.put(userId,this);atomicInteger.getAndIncrement();log.info("onOpen, connect a new Socket {}",userId);}catch (Exception e){log.error("onOpen error {}",e.getMessage());e.printStackTrace();}}/*** 关闭连接*/@OnClosepublic void onClose(){map.remove(this.userId);atomicInteger.getAndDecrement();removeRedisKey();log.info("onClose, disconnect a socket {}",this.userId);}@OnErrorpublic void onError(Throwable throwable){map.remove(this.userId);atomicInteger.getAndDecrement();removeRedisKey();log.error("onError, error! {}",throwable.getMessage());}@OnMessagepublic void onMessage(Session session,String msg){log.info("receive msg from client: {}, session: {}",msg,session);}/*** 获取链接数量* @return*/public int getConNum(){return atomicInteger.intValue();}/*** 发送消息* @param msg*/public void sendMessage(String msg){this.session.getAsyncRemote().sendText(msg);}/*** 向指定用户发送消息* @param userId* @param msg*/public void sendMessageToUser(String userId,String msg){if (map.containsKey(userId)){// 生成key==pushIdString key = StrHelper.getFormatStr(RedisConstant.KEY_PUSH_ID,userId);redisUtil.set(key, StrHelper.getRandNum());// 将消息封装到消息体中PushCommonVo pushCommonVo = PushCommonVo.builder().msg(msg).pushId(key).updateTime(DateTime.now()).build();String jsonStr = JsonUtil.toStr(R.ok(pushCommonVo));map.get(userId).sendMessage(jsonStr);}}/*** 向指定用户群发消息* @param userIds* @param msg*/public void sendBatchMessage(List userIds,String msg){userIds.forEach(userId->{sendMessageToUser(userId,msg);});}/*** 向所有用户发送消息* @param msg*/public void sendMessageToAll(String msg){map.forEach((k,v)->{sendMessageToUser(k,msg);});}/*** 删除redis key* @return*/public boolean removeRedisKey(){if (StrUtil.isNotEmpty(this.userId)) {String key = StrHelper.getFormatStr(RedisConstant.KEY_PUSH_ID, this.userId);log.info("removeRedisKey: "+key);if (redisUtil.hasKey(key)) {redisUtil.deleteByKey(key);return true;}}return false;}/*** 服务端向客户端发送ping*/public static void pingMessage(String userId){byte[] pingByte = new byte[1];pingByte[0]='p';ByteBuffer byteBuffer = ByteBuffer.wrap(pingByte);if (map.containsKey(userId)) {try {map.get(userId).session.getBasicRemote().sendPing(byteBuffer);} catch (IOException e) {log.error("ping error: {}",e.getMessage());e.printStackTrace();}}}
}

2.2.3 测试结果

2.3 Redis

引入Redis，用于记录下发消息结构体重的pushId。

1. 推送一条信息，服务端缓存消息体中的pushId，并设置过期时间
2. 当客户端接受到消息后，http调用并确认
3. 收到客户端的确认后，删除redis缓存

2.3.1 Redis配置

package com.ym.pushserver.config;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;/*** @Author: Yangmiao* @Date: 2023/3/11 12:18* @Desc: Redis 配置*/
@Configuration
public class RedisConfig {@Autowiredprivate RedisConnectionFactory redisConnectionFactory;@Beanpublic RedisTemplate redisTemplate(){RedisTemplate redisTemplate = new RedisTemplate<>();redisTemplate.setKeySerializer(new StringRedisSerializer());redisTemplate.setHashKeySerializer(new StringRedisSerializer());redisTemplate.setHashValueSerializer(new StringRedisSerializer());redisTemplate.setValueSerializer(new StringRedisSerializer());redisTemplate.setConnectionFactory(redisConnectionFactory);return redisTemplate;}}

2.3.2 Redis封装成工具类

package com.ym.pushserver.redis;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;/*** @Author: Yangmiao* @Date: 2023/3/11 12:22* @Desc:*/
@Component
public class RedisUtil {@Autowiredprivate RedisTemplate redisTemplate;/*** 给一个指定的 key 值附加过期时间** @param key* @param time* @return*/public boolean expire(String key, long time) {return redisTemplate.expire(key, time, TimeUnit.SECONDS);}/*** 根据key 获取过期时间** @param key* @return*/public long getTime(String key) {return redisTemplate.getExpire(key, TimeUnit.SECONDS);}/*** 根据key 获取过期时间** @param key* @return*/public boolean hasKey(String key) {return redisTemplate.hasKey(key);}/*** 移除指定key 的过期时间** @param key* @return*/public boolean persist(String key) {return redisTemplate.boundValueOps(key).persist();}//- - - - - - - - - - - - - - - - - - - - -  String类型 - - - - - - - - - - - - - - - - - - - -/*** 根据key获取值** @param key 键* @return 值*/public Object get(String key) {return key == null ? null : redisTemplate.opsForValue().get(key);}/*** 将值放入缓存** @param key   键* @param value 值* @return true成功 false 失败*/public void set(String key, String value) {redisTemplate.opsForValue().set(key, value);}/*** 将值放入缓存并设置时间** @param key   键* @param value 值* @param time  时间(秒) -1为无期限* @return true成功 false 失败*/public void set(String key, String value, long time) {if (time > 0) {redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);} else {redisTemplate.opsForValue().set(key, value);}}/*** 批量添加 key (重复的键会覆盖)** @param keyAndValue*/public void batchSet(Map keyAndValue) {redisTemplate.opsForValue().multiSet(keyAndValue);}/*** 批量添加 key-value 只有在键不存在时,才添加* map 中只要有一个key存在,则全部不添加** @param keyAndValue*/public void batchSetIfAbsent(Map keyAndValue) {redisTemplate.opsForValue().multiSetIfAbsent(keyAndValue);}/*** 对一个 key-value 的值进行加减操作,* 如果该 key 不存在 将创建一个key 并赋值该 number* 如果 key 存在,但 value 不是长整型 ,将报错** @param key* @param number*/public Long increment(String key, long number) {return redisTemplate.opsForValue().increment(key, number);}/*** 对一个 key-value 的值进行加减操作,* 如果该 key 不存在 将创建一个key 并赋值该 number* 如果 key 存在,但 value 不是 纯数字 ,将报错** @param key* @param number*/public Double increment(String key, double number) {return redisTemplate.opsForValue().increment(key, number);}//- - - - - - - - - - - - - - - - - - - - -  set类型 - - - - - - - - - - - - - - - - - - - -/*** 将数据放入set缓存** @param key 键* @return*/public void sSet(String key, String value) {redisTemplate.opsForSet().add(key, value);}/*** 获取变量中的值** @param key 键* @return*/public Set