• 目录

  前言

  一、SocketIO和WebSocket是什么？

  二、使用步骤

  1.SocketIO

  2.WebSocket

  总结

---

前言

由于公司业务需要，需要使用WebSocket发送消息提示，但一开始的技术选型是使用SocketIO，但由于SocketIO独占一个端口，所以就放弃了，但是换原生WebSocket之前已经把SocketIO全部测试ok了，所以今天就把SocketIO和WebSocket一起过一遍。使用的过程也都参考其他文章，这里主要记录一下自己的使用思路以及方法。

---

下面开始我们今天的学习吧！

一、SocketIO和WebSocket是什么？

WebSocket实现了服务端和客户端之间的实时事件以及消息传递，通常使用于聊天室之类的实时在线发送消息，底层走的是TCP协议，保证了连接的稳定性而且性能也要比HTTP协议更好。其他区别网上也有一大堆，我就不一一列举了。

SocketIO提供了更多功能，封装了WebSocket，支持WebSocket，以轮训方式的Http来兼容旧版浏览器，，它比WebSocket使用更加方便简单。但是有个缺点是它需要自己独占一个端口，有一个解决办法就是使用nginx的反向代理，通过请求协议来转发到不同的端口号上。

二、使用步骤

1.SocketIO

前端：

首先安装依赖

pnpm add socket.io-client

之后创建一个Ts文件来创建全局SocketIO对象

import { io } from 'socket.io-client';

let socket;

export function socketInstance(token, userId) {
  socket = io(import.meta.env.VITE_SOCKET_API_URL, {
    transports: ['websocket'],//默认是HTTP轮训，设置这个就是为ws
    upgrade: false,//关闭自动升级ws，开启的话监听的通过消息会收不到
    query: { //通过参数的形式传参
      token: token,
      userId: userId,
    },
  });
  return socket;
}

这样我们就创建好了SocketIO对象，但是我们想在其它地方也使用这个对象，所有我们在暴露一个只返回socket对象的方法

export function getSocket() {
  return socket;
}

只需在使用的地方引用这个方法就可以啦。

创建全局对象的时机是在登录之后，因为登录之后才可以获取到Token和UserId，具体的地方我就不展示了根据启动顺序找到一个合适的文件在该文件中引用这个Ts文件使用暴露socketInstance
方法创建即可。

import { socketInstance } from '@/utils/http/ws/index';
const socket = socketInstance(getToken());

在需要监听的页面上使用getSocket方法获取对象来创建监听方法

import { getSocket} from '@/utils/http/ws/index';
const socket = getSocket();

socket.on('connect', () => { // 默认通道 connect是通道名称
  console.log('连接成功');
});
socket.on('menu_prompt_add', (data) => { // 自定义通道 前后端约定
  console.log('监听到增加消息：');
  console.log(data);
  setMenuPromptSize(menusRef.value, data, false, 'add');
});
socket.on('menu_prompt_sub', (data) => { // 自定义通道
  console.log('监听到减少消息：');
  console.log(data);
  setMenuPromptSize(menusRef.value, data, true, 'sub');
});
socket.on('disconnect', () => { // 默认通道
   console.log('断开连接');
});

后端：

首先引入依赖

    <!-- SocketIO -->
    <dependency>
      <groupId>com.corundumstudio.socketio</groupId>
      <artifactId>netty-socketio</artifactId>
      <version>2.0.0</version>
    </dependency>

yml配置文件添加

socketio:
  host: "localhost" 
  port: 9555
  bossCount: 1 #socket连接数大小
  workCount: 100
  allowCustomRequests: true
  upgradeTimeout: 1000000 #协议升级超时时间（毫秒）
  pingTimeout: 6000000 #Ping消息超时时间（毫秒）
  pingInterval: 25000 #Ping消息间隔（毫秒）心跳验证

创建SocketIO配置类

@Data
@Configuration
@ConfigurationProperties(prefix = "socketio")
public class SocketIOConfig {

    private String host;
    private Integer port;
    private int bossCount;
    private int workCount;
    private boolean allowCustomRequests;
    private int upgradeTimeout;
    private int pingTimeout;
    private int pingInterval;

    @Bean
    public SocketIOServer socketIOServer() {
        SocketConfig socketConfig = new SocketConfig();
        socketConfig.setTcpNoDelay(true);
        socketConfig.setSoLinger(0);
        com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration();
        config.setSocketConfig(socketConfig);
        config.setHostname(host);
        config.setPort(port);
        config.setBossThreads(bossCount);
        config.setWorkerThreads(workCount);
        config.setAllowCustomRequests(allowCustomRequests);
        config.setUpgradeTimeout(upgradeTimeout);
        config.setPingTimeout(pingTimeout);
        config.setPingInterval(pingInterval);
        return new SocketIOServer(config);
    }

    @Bean
    public SpringAnnotationScanner springAnnotationScanner() {
        return new SpringAnnotationScanner(socketIOServer());
    }
}

创建WebSocket连接处理类，当用户连接时校验Token，校验通过存到Map中否则断开连接

@Slf4j
@Component
public class SocketIOHandler {

    public static Map<String, SocketIOClient> clientMap = new ConcurrentHashMap<>();
    @Autowired
    private SocketIOServer socketIoServer;

    @PostConstruct
    private void autoStartup() {
        try {
            socketIoServer.start();
        } catch (Exception ex) {
            ex.printStackTrace();
            log.error("SocketIOServer启动失败");
        }
    }

    @PreDestroy
    private void autoStop() {
        socketIoServer.stop();
    }

    @OnConnect
    public void onConnect(SocketIOClient client) {
        String token = client.getHandshakeData().getSingleUrlParam("token");
        try {
            // 校验Token
            if (Token不通过) {
                client.disconnect();
                return;
            }
            String userId = loginUser.getId();
            if (userId.equals(client.getHandshakeData().getSingleUrlParam("userId"))) {
                client.joinRoom(userId);
                clientMap.put(userId, client);

            log.info("客户端:" + client.getRemoteAddress() + "  sessionId:" + client.getSessionId() + " userId: " + loginUser.getUsername() + "已连接");
            log.info(clientMap.toString());
            System.out.println("onConnect" + socketIoServer.getAllClients().size());
            }
        } catch (Exception e) {
            e.printStackTrace();
            client.disconnect();
            System.out.println("onConnect" + socketIoServer.getAllClients().size());
        }
    }

    @OnDisconnect
    public void onDisconnect(SocketIOClient client) {
        String userId = client.getHandshakeData().getSingleUrlParam("userId");
        clientMap.remove(userId);
        client.disconnect();
        log.info("客户端:" + client.getSessionId() + "断开连接");
        log.info(clientMap.toString());
        System.out.println("onDisconnect" + socketIoServer.getAllClients().size());
    }

    @OnEvent(value = "heartbeat")
    public void onEvent(SocketIOClient client, AckRequest ackRequest, MenuPromptMessageDTO data) {
        ackRequest.sendAckData(true);
        client.sendEvent("heartbeatResponse", "ok");
        log.info("服务端收到消息：{}，{}", data.toString(), ackRequest.isAckRequested());
    }

    /**
     * 广播消息 函数可在其他类中调用
     */
    public static void sendBroadcast(byte[] data) {
        //向已连接的所有客户端发送数据,map实现客户端的存储
        for (SocketIOClient client : clientMap.values()) {
            if (client.isChannelOpen()) {
                client.sendEvent("message_event", data);
            }
        }
    }

}

创建公共发送Socket消息类，messageTopic消息通道，userIds用户列表，MenuPromptMessageDTO消息体

@Slf4j
@Service
public class SocketIOService {

    @Autowired
    private SocketIOServer socketIoServer;
   /**
    * 发送消息
    *
    * @param messageTopic Socket监听通道
    * @param userIds 用户列表
    * @param dto 消息体
    * @return void
    */
    public void sendMessage(String messageTopic, List<String> userIds, MenuPromptMessageDTO dto){
        if (CollectionUtils.isNotEmpty(userIds)) {
            for (String userId : userIds) {
                // 这个是给加入房间的所有用户发消息
                BroadcastOperations roomOperations = socketIoServer.getRoomOperations(userId);
                if (roomOperations != null) {
                    roomOperations.sendEvent(messageTopic,dto);
                }
                // 根据用户发消息 和上面的二选一
//                SocketIOClient socketIOClient = SocketIOHandler.clientMap.get(userId);
//                if (socketIOClient != null){
//                    socketIOClient.sendEvent(messageTopic,dto);
//                }else {
//                    log.info("发送失败，用户：{}，未连接",userId);
//                }
            }
        }

    }

}

2.WebSocket

前端：

Vue3核心包中已经写好了WebSocket，我们需要写一个Ts文件来全局创建一个WebSocket，这个文件参考Jeecg的开源框架修改而来的

import { useWebSocket, UseWebSocketReturn } from '@vueuse/core';
import { getToken } from '/@/utils/auth';

let wsInstance: UseWebSocketReturn<any>;
const listeners = new Map();

/**
 * 开启 WebSocket 链接，全局只需执行一次
 * @param url
 */
export function connectWebSocket(url: string) {
  const token = (getToken() || '') as string;
  wsInstance = useWebSocket(url, {
    heartbeat: {
      interval: 60000,
    },
    autoReconnect: {
      retries: 5,
      delay: 5000,
    },
    protocols: [token],
    onError,
    onMessage,
  });
}

function onError(ws, e) {
  console.log('[WebSocket] 连接发生错误: ', e);
}

function onMessage(ws, e) {
  try {
    const data = JSON.parse(e.data);
    for (const callback of listeners.keys()) {
      try {
        callback(data);
      } catch (err) {
        console.error(err);
      }
    }
  } catch (err) {
    console.error('[WebSocket] data解析失败：', err);
  }
}

/**
 * 添加 WebSocket 消息监听
 * @param callback
 */
export function onWebSocket(callback: (data: object) => any) {
  if (!listeners.has(callback)) {
    if (typeof callback === 'function') {
      listeners.set(callback, null);
    } else {
      console.log('[WebSocket] 添加 WebSocket 消息监听失败：传入的参数不是一个方法');
    }
  }
}

/**
 * 解除 WebSocket 消息监听
 *
 * @param callback
 */
export function offWebSocket(callback: (data: object) => any) {
  listeners.delete(callback);
}

export function useMyWebSocket() {
  return wsInstance;
}

这个主要就是在这里监听消息，需要我们自定义通道标识，比如topic：message，body：Object

在需要监听的地方引用onWebSocket传入一个方法具体如下

import { onWebSocket, offWebSocket } from '/@/hooks/web/useWebSocket';

onMounted(async () => {
  onWebSocket(socketLoadNotKs);
});
  
onUnmounted(() => {
  offWebSocket(socketOnFun); //组件关闭时删除这个方法
});
const socketOnFun = (data) => {
    if (data.topic=== 'message') {
      // 业务代码
      const body = data.body;// 消息体
    }
};

后端：

主要方法依然参考的是Jeecg框架，它使用的是Redis订阅者模式，兼容集群部署，每个服务都订阅这个频道来消费这里面的数据，但是有一点就是如果订阅的服务宕机了，消息就丢失了，因为Redis不提供持久化。那这里为啥不用消息队列呢？因为项目小，数据也不是特别重要，消息只是为了提醒用户，所以用了Redis来兼容以后的扩展成多个服务，当然现在还是一个服务。

实现步骤如下

引入依赖

    <!-- websocket -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-websocket</artifactId>
    </dependency>

创建WebSocket配置类

@Configuration
public class WebSocketConfig {

·   // 这个是为使用@ServerEndpoint，必须创建，否则不生效
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

    @Bean
    public FilterRegistrationBean<WebSocketFilter> getFilterRegistrationBean(){
        FilterRegistrationBean<WebSocketFilter> bean = new FilterRegistrationBean<>();
        bean.setFilter(new WebSocketFilter());
        bean.addUrlPatterns("/websocket/*");
        return bean;
    }

}

创建WebSocket连接过滤器

@Slf4j
public class WebSocketFilter implements Filter {

    private static final String TOKEN_KEY = "Sec-WebSocket-Protocol";

    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {

        HttpServletRequest httpServletRequest = (HttpServletRequest) request;
        String token = httpServletRequest.getHeader(TOKEN_KEY);
        log.debug("Websocket连接 Token安全校验，Path = {}，token:{}", httpServletRequest.getRequestURI(), token);

        try {
            // Token校验
        } catch (AuthenticationException e) {
            log.error(e.getMessage(), e);
            log.error("Websocket连接 Token安全校验失败，IP:{}, Token:{}, Path = {}，异常：{}", IPUtil.getIpAddr(httpServletRequest), token, httpServletRequest.getRequestURI(), e.getMessage());
            return;
        }

        HttpServletResponse httpServletResponse = (HttpServletResponse) response;
        httpServletResponse.setHeader(TOKEN_KEY, token);
        chain.doFilter(request, response);
    }

}

创建WebSocket连接处理类

@Slf4j
@Component
@ServerEndpoint("/websocket/{userId}")
public class WebSocketServer {

    private static final ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();

    @OnOpen
    public void onOpen(Session session, @PathParam(value = "userId") String userId) {
        sessionPool.put(userId, session);
        log.debug("【websocket】有新的连接，总数为:" + sessionPool.size());
    }

    @OnClose
    public void onClose(@PathParam("userId") String userId) {
        sessionPool.remove(userId);
        log.debug("【websocket】连接断开，总数为:" + sessionPool.size());
    }

    @OnError
    public void onError(Session session, Throwable t) {
        log.error("【websocket】出现错误");
        log.error(t.getMessage(), t);
    }

    @OnMessage
    public void onMessage(String message, @PathParam(value = "userId") String userId) {
        log.debug("【websocket】收到客户端消息:" + message);
        this.pushMessage(userId, message);
    }

    /**
     * 推送消息
     *
     * @param message 消息内容
     */
    public void pushMessage(String message) {
        log.debug("【websocket】推送广播消息:" + message);
        for (Map.Entry<String, Session> item : sessionPool.entrySet()) {
            try {
                if (item.getValue().isOpen()) {
                    item.getValue().getAsyncRemote().sendText(message);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    /**
     * 推送消息
     *
     * @param userId  发送用户
     * @param message 消息内容
     */
    public void pushMessage(String userId, String message) {
        log.debug("【websocket】推送单人消息:" + message);
        for (Map.Entry<String, Session> item : sessionPool.entrySet()) {
            try {
                if (item.getKey().startsWith(userId)) {
                    item.getValue().getBasicRemote().sendText(message);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

}

创建自定义Redis监听处理类

@Slf4j
@Component
@RequiredArgsConstructor
public class WebSocketRedisListener {

    private final WebSocketServer webSocketServer;

    public void handleMessage(Map<String, Object> message) {
        String userId = message.get("userId").toString();
        String wsData = JSON.toJSONString(message.get("waData"));
        if (StringUtils.isNotBlank(userId)) {
            webSocketServer.pushMessage(userId, wsData);
        } else {
            webSocketServer.pushMessage(wsData);
        }
    }

}

创建Redis消息监听配置类

@Configuration
public class RedisListenerConfigure {

    @Bean
    public RedisMessageListenerContainer listenerContainer(RedisConnectionFactory factory, MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        // 监听的Redis频道
        container.addMessageListener(listenerAdapter, new ChannelTopic("redis_channel_topic"));
        return container;
    }

    @Bean
    public MessageListenerAdapter listenerAdapter(WebSocketRedisListener webSocketRedisListener) {
        // 自定义消息处理类webSocketRedisListener
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(webSocketRedisListener);
        messageListenerAdapter.setSerializer(jacksonSerializer());
        return messageListenerAdapter;
    }

    // 序列化
    private Jackson2JsonRedisSerializer<Object> jacksonSerializer() {
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
        ObjectMapper mapper = new ObjectMapper();
        mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        mapper.activateDefaultTyping(mapper.getPolymorphicTypeValidator(), ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(mapper);
        return jackson2JsonRedisSerializer;
    }

}

创建公共发消息到Redis的处理类

@Slf4j
@Component
public class WebSocketHelper {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 发送消息到redis
     *
     * @param channel     消息通道
     * @param messageBody 消息体
     */
    public void sendMessage(WebSocketChannelEnum channel, Object messageBody) {
        sendMessage(channel, "", messageBody);
    }

    /**
     * 发送消息到redis
     *
     * @param channel     消息通道
     * @param userIds     发送用户
     * @param messageBody 消息体
     */
    public void sendMessage(WebSocketChannelEnum channel, List<String> userIds, Object messageBody) {
        if (CollectionUtils.isNotEmpty(userIds)) {
            for (String userId : userIds) {
                sendMessage(channel, userId, messageBody);
            }
        }
    }

    /**
     * 发送消息到redis
     * WebSocketChannelEnum 中的枚举格式：NOTICE("notice", "通知公告"),
     * notice是value，前后端约定好即可
     *
     * @param channel     消息通道
     * @param userId      发送用户
     * @param messageBody 消息体
     */
    public void sendMessage(WebSocketChannelEnum channel, String userId, Object messageBody) {
        // 消息体
        JSONObject wsData = new JSONObject();
        wsData.put("topic", channel.getValue());
        wsData.put("body", messageBody);

        Map<String, Object> message = new HashMap<>();
        message.put("userId", userId);
        message.put("wsData", wsData);
        redisTemplate.convertAndSend("redis_channel_topic", message);
    }

}

在业务代码中注入WebSocketHelper 类就可实现消息推送

1. 调用WebSocketHelper 的sendMessage方法后将消息发送到Redis

2. 通过WebSocketRedisListener 监听类来处理消息

3. 每个服务收到消息后都会执行一遍pushMessage

---

总结

这些我也是看着其他帖子实现的，大体思路差不多，细节不太一样，实现还是没问题的。

下一期我想准备UniApp的学习，看情况吧（不确定有没有时间）主要以公司项目为主，现在阶段的任务还没有太难的技术点，有点难我会记录下来并发布到（PCode进阶）公众号CSDN同步更新上。

记得一键三连，嘻嘻！