SpringBoot(23) 集成socket.io服务端和客户端实现通信

服务实现类：

@Slf4j
@Service(value = "socketIOService")
public class SocketIOServiceImpl implements ISocketIOService {

    
    private static Map clientMap = new ConcurrentHashMap<>();

    
    private static final String PUSH_DATA_EVENT = "push_data_event";

    @Autowired
    private SocketIOServer socketIOServer;

    
    @PostConstruct
    private void autoStartup() {
 start();
    }

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

    @Override
    public void start() {
 // 监听客户端连接
 socketIOServer.addConnectListener(client -> {
     log.debug("************ 客户端： " + getIpByClient(client) + " 已连接 ************");
     // 自定义事件`connected` -> 与客户端通信  （也可以使用内置事件，如：Socket.EVENT_CONNECT）
     client.sendEvent("connected", "你成功连接上了哦...");
     String userId = getParamsByClient(client);
     if (userId != null) {
  clientMap.put(userId, client);
     }
 });

 // 监听客户端断开连接
 socketIOServer.addDisconnectListener(client -> {
     String clientIp = getIpByClient(client);
     log.debug(clientIp + " *********************** " + "客户端已断开连接");
     String userId = getParamsByClient(client);
     if (userId != null) {
  clientMap.remove(userId);
  client.disconnect();
     }
 });

 // 自定义事件`client_info_event` -> 监听客户端消息
 socketIOServer.addEventListener(PUSH_DATA_EVENT, String.class, (client, data, ackSender) -> {
     // 客户端推送`client_info_event`事件时，onData接受数据，这里是string类型的json数据，还可以为Byte[],object其他类型
     String clientIp = getIpByClient(client);
     log.debug(clientIp + " ************ 客户端：" + data);
 });

 // 启动服务
 socketIOServer.start();

 // broadcast: 默认是向所有的socket连接进行广播，但是不包括发送者自身，如果自己也打算接收消息的话，需要给自己单独发送。
 new Thread(() -> {
     int i = 0;
     while (true) {
  try {
      // 每3秒发送一次广播消息
      Thread.sleep(3000);
      socketIOServer.getBroadcastOperations().sendEvent("myBroadcast", "广播消息 " + DateUtil.now());
  } catch (InterruptedException e) {
      e.printStackTrace();
  }
     }
 }).start();
    }

    @Override
    public void stop() {
 if (socketIOServer != null) {
     socketIOServer.stop();
     socketIOServer = null;
 }
    }

    @Override
    public void pushMessageToUser(String userId, String msgContent) {
 SocketIOClient client = clientMap.get(userId);
 if (client != null) {
     client.sendEvent(PUSH_DATA_EVENT, msgContent);
 }
    }

    
    private String getParamsByClient(SocketIOClient client) {
 // 获取客户端url参数（这里的userId是唯一标识）
 Map> params = client.getHandshakeData().getUrlParams();
 List userIdList = params.get("userId");
 if (!CollectionUtils.isEmpty(userIdList)) {
     return userIdList.get(0);
 }
 return null;
    }

    
    private String getIpByClient(SocketIOClient client) {
 String sa = client.getRemoteAddress().toString();
 String clientIp = sa.substring(1, sa.indexOf(":"));
 return clientIp;
    }

}

@Slf4j
public class SocketIOClientLaunch {

    public static void main(String[] args) {
 // 服务端socket.io连接通信地址
 String url = "http://127.0.0.1:8888";
 try {
     IO.Options options = new IO.Options();
     options.transports = new String[]{"websocket"};
     options.reconnectionAttempts = 2;
     // 失败重连的时间间隔
     options.reconnectionDelay = 1000;
     // 连接超时时间(ms)
     options.timeout = 500;
     // userId: 唯一标识 传给服务端存储
     final Socket socket = IO.socket(url + "?userId=1", options);

     socket.on(Socket.EVENT_CONNECT, args1 -> socket.send("hello..."));

     // 自定义事件`connected` -> 接收服务端成功连接消息
     socket.on("connected", objects -> log.debug("服务端:" + objects[0].toString()));

     // 自定义事件`push_data_event` -> 接收服务端消息
     socket.on("push_data_event", objects -> log.debug("服务端:" + objects[0].toString()));

     // 自定义事件`myBroadcast` -> 接收服务端广播消息
     socket.on("myBroadcast", objects -> log.debug("服务端：" + objects[0].toString()));

     socket.connect();

     while (true) {
  Thread.sleep(3000);
  // 自定义事件`push_data_event` -> 向服务端发送消息
  socket.emit("push_data_event", "发送数据 " + DateUtil.now());
     }
 } catch (Exception e) {
     e.printStackTrace();
 }
    }

}

socketIOServer.getBroadcastOperations().sendEvent("myBroadcast", "广播消息 " + DateUtil.now());

@RestController
@RequestMapping("/api/socket.io")
@Api(tags = "SocketIO测试-接口")
public class SocketIOController {

    @Autowired
    private ISocketIOService socketIOService;

    @PostMapping(value = "/pushMessageToUser", produces = Constants.CONTENT_TYPE)
    @ApiOperation(value = "推送信息给指定客户端", httpMethod = "POST", response = ApiResult.class)
    public ApiResult pushMessageToUser(@RequestParam String userId, @RequestParam String msgContent) {
 socketIOService.pushMessageToUser(userId, msgContent);
 return ApiResult.ok();
    }

}