一、前言
websocket和socket.io区别？
websocket
- 一种让客户端和服务器之间能进行双向实时通信的技术
- 使用时，虽然主流浏览器都已经支持，但仍然可能有不兼容的情况
- 适合用于client和基于node搭建的服务端使用
socket.io
- 将WebSocket、AJAX和其它的通信方式全部封装成了统一的通信接口
- 使用时，不用担心兼容问题，底层会自动选用最佳的通信方式
- 适合进行服务端和客户端双向数据通信
w3cschool上对socket.io的描述如下：
本文将实现
- 基于springboot2.1.8.RELEASE集成netty-socketio ： 仿node.js实现的socket.io服务端
- 集成socket.io-client： socket.io客户端
- 实现服务端与客户端之间的通信
二、Java集成socket.io服务端
1、pom.xml中引入所需依赖
温馨小提示：这里为了方便将后面需要的客户端socket.io-client依赖一起直接引入了哦~
<code><dependency> <groupid>com.corundumstudio.socketio/<groupid> <artifactid>netty-socketio/<artifactid> <version>1.7.7/<version>/<dependency><dependency> <groupid>io.socket/<groupid> <artifactid>socket.io-client/<artifactid> <version>1.0.0/<version>/<dependency>/<code>
2、application.yml中配置socket.io服务端
<code># netty-socketio 配置socketio: host: 127.0.0.1 port: 8888 # 设置最大每帧处理数据的长度，防止他人利用大数据来攻击服务器 maxFramePayloadLength: 1048576 # 设置http交互最大内容长度 maxHttpContentLength: 1048576 # socket连接数大小（如只监听一个端口boss线程组为1即可） bossCount: 1 workCount: 100 allowCustomRequests: true # 协议升级超时时间（毫秒），默认10秒。HTTP握手升级为ws协议超时时间 upgradeTimeout: 1000000 # Ping消息超时时间（毫秒），默认60秒，这个时间间隔内没有接收到心跳消息就会发送超时事件 pingTimeout: 6000000 # Ping消息间隔（毫秒），默认25秒。客户端向服务器发送一条心跳消息间隔 pingInterval: 25000/<code>
3、socket.io服务端配置类
<code>@Configurationpublic class SocketIOConfig { @Value("${socketio.host}") private String host; @Value("${socketio.port}") private Integer port; @Value("${socketio.bossCount}") private int bossCount; @Value("${socketio.workCount}") private int workCount; @Value("${socketio.allowCustomRequests}") private boolean allowCustomRequests; @Value("${socketio.upgradeTimeout}") private int upgradeTimeout; @Value("${socketio.pingTimeout}") private int pingTimeout; @Value("${socketio.pingInterval}") 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); }}/<code>4、socket.io服务端服务层
服务类
<code>public interface ISocketIOService { /** * 启动服务 */ void start(); /** * 停止服务 */ void stop(); /** * 推送信息给指定客户端 * * @param userId: 客户端唯一标识 * @param msgContent: 消息内容 */ void pushMessageToUser(String userId, String msgContent);}/<code>服务实现类：
<code>@Slf4j@Service(value = "socketIOService")public class SocketIOServiceImpl implements ISocketIOService { /** * 存放已连接的客户端 */ private static Map<string> clientMap = new ConcurrentHashMap<>(); /** * 自定义事件`push_data_event`,用于服务端与客户端通信 */ private static final String PUSH_DATA_EVENT = "push_data_event"; @Autowired private SocketIOServer socketIOServer; /** * Spring IoC容器创建之后，在加载SocketIOServiceImpl Bean之后启动 */ @PostConstruct private void autoStartup() { start(); } /** * Spring IoC容器在销毁SocketIOServiceImpl Bean之前关闭,避免重启项目服务端口占用问题 */ @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); } } /** * 获取客户端url中的userId参数（这里根据个人需求和客户端对应修改即可） * * @param client: 客户端 * @return: java.lang.String */ private String getParamsByClient(SocketIOClient client) { // 获取客户端url参数（这里的userId是唯一标识） Map<string>> params = client.getHandshakeData().getUrlParams(); List<string> userIdList = params.get("userId"); if (!CollectionUtils.isEmpty(userIdList)) { return userIdList.get(0); } return null; } /** * 获取连接的客户端ip地址 * * @param client: 客户端 * @return: java.lang.String */ private String getIpByClient(SocketIOClient client) { String sa = client.getRemoteAddress().toString(); String clientIp = sa.substring(1, sa.indexOf(":")); return clientIp; }}/<string>/<string>/<string>/<code> 三、Java开发socket.io客户端
- socket.emit：发送数据到服务端事件
- socket.on： 监听服务端事件
<code>@Slf4jpublic 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(); } }}/<code>四、运行测试
当客户端上线后，会通过自定义事件push_data_event每隔3秒向服务端发送消息，日志如下
而服务端中跑了一个广播消息(自定义事件myBroadcast) 每隔3秒也会返回给客户端
广播事件: 向所有的socket连接进行广播发送消息数据
<code>socketIOServer.getBroadcastOperations().sendEvent("myBroadcast", "广播消息 " + DateUtil.now());/<code>日志如下：
编写服务端主动发送消息给客户端接口
<code>@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(); }}/<code> 调用接口测试发送helloworld...
五、总结
socket.io通信，服务端：
1.socketIOServer.addConnectListener：监听客户端连接
- socketIOServer.addDisconnectListener：监听客户端断开连接
- socketIOServer.addEventListener：监听客户端传输的消息
- client.sendEvent("自定义事件名称", "消息内容")：服务端向指定的clien客户端发送消息
- socketIOServer.getBroadcastOperations().sendEvent("自定义事件名称", "消息内容")：服务端发送广播消息给所有客户端
socket.io通信，客户端：
- IO.socket(url)：与指定的socket.io服务端建立连接
- socket.emit：发送数据到服务端事件
- socket.on： 监听服务端事件
本文案例demo源码
https://gitee.com/zhengqingya/java-workspace
分享到:
閱讀更多 鄭清 的文章
