socket-demo的实现 (4)

下面以server端发送数据为例,改代码中加入了重试机制：

public void println(String message) { int count = 0; PrintWriter writer; do { try { writer = new PrintWriter(new OutputStreamWriter(socket.getOutputStream()), true); writer.println(message); break; } catch (IOException e) { count++; if (count >= RETRY_COUNT) { //重试多次失败，说明client端socket异常 this.connectionThread.stopRunning(); } } try { Thread.sleep(2 * 1000); } catch (InterruptedException e1) { log.error("Connection.println.IOException interrupt,userId:{}", userId); } } while (count < 3); }

上述调用的this.connectionThread.stopRunning();代码如下：

public void stopRunning() { //设置线程对象状态，便于线程清理 isRunning = false; try { //异常情况需要将该socket资源释放 socket.close(); } catch (IOException e) { log.error("ConnectionThread.stopRunning failed.exception:{}", e); } }

上述代码中设置了线程对象的状态，下述代码在监测线程中执行，将没有运行的线程给清理掉

/** * 存储只要有socket处理的线程 */ private List<ConnectionThread> existConnectionThreadList = Collections.synchronizedList(new ArrayList<>()); /** * 中间list，用于遍历的时候删除 */ private List<ConnectionThread> noConnectionThreadList = Collections.synchronizedList(new ArrayList<>()); //... //删除list中没有用的thread引用 existConnectionThreadList.forEach(connectionThread -> { if (!connectionThread.isRunning()) { noConnectionThreadList.add(connectionThread); } }); noConnectionThreadList.forEach(connectionThread -> { existConnectionThreadList.remove(connectionThread); if (connectionThread.getConnection().isLogin()) { //说明用户已经身份验证成功了，需要删除map this.existSocketMap.remove(connectionThread.getConnection().getUserId()); } }); noConnectionThreadList.clear(); 项目结构

由于使用了springboot框架来实现该demo，所以项目结构如下：

socket工具包目录如下：

pom文件主要添加了springboot的相关依赖，以及json工具和lombok工具等，依赖如下：

<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.0.3.RELEASE</version> <relativePath/> </parent> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.36</version> </dependency> </dependencies>

自己写的socket工具包的使用方式如下：

@Configuration @Slf4j public class SocketServerConfig { @Bean public SocketServer socketServer() { SocketServer socketServer = new SocketServer(60000); socketServer.setLoginHandler(userId -> { log.info("处理socket用户身份验证,userId:{}", userId); //用户名中包含了dingxu则允许登陆 return userId.contains("dingxu"); }); socketServer.setMessageHandler((connection, receiveDto) -> log .info("处理socket消息,userId:{},receiveDto:{}", connection.getUserId(), JSONObject.toJSONString(receiveDto))); socketServer.start(); return socketServer; } }

该demo中主要提供了以下几个接口进行测试：

服务端：获得当前用户列表，发送一个消息

客户端：开始一个socket客户端，发送一个消息，关闭一个socket客户端，查看已开启的客户端

具体的postman文件也放已在项目中，具体可点此链接获得

demo中还提供了一个简单压测函数，如下：

@Slf4j public class SocketClientTest { public static void main(String[] args) { ExecutorService clientService = Executors.newCachedThreadPool(); String userId = "dingxu"; for (int i = 0; i < 1000; i++) { int index = i; clientService.execute(() -> { try { SocketClient client; client = new SocketClient(InetAddress.getByName("127.0.0.1"), 60000); //登陆 ClientSendDto dto = new ClientSendDto(); dto.setFunctionCode(FunctionCodeEnum.LOGIN.getValue()); dto.setUserId(userId + index); client.println(JSONObject.toJSONString(dto)); ScheduledExecutorService clientHeartExecutor = Executors.newSingleThreadScheduledExecutor( r -> new Thread(r, "socket_client+heart_" + r.hashCode())); clientHeartExecutor.scheduleWithFixedDelay(() -> { try { ClientSendDto heartDto = new ClientSendDto(); heartDto.setFunctionCode(FunctionCodeEnum.HEART.getValue()); client.println(JSONObject.toJSONString(heartDto)); } catch (Exception e) { log.error("客户端异常,userId:{},exception：{}", userId, e.getMessage()); client.close(); } }, 0, 5, TimeUnit.SECONDS); while (true){ } } catch (Exception e) { log.error(e.getMessage()); } }); } } } 参考

系统间通信技术专栏

JAVA Socket编程学习10--解决TCP粘包分包问题

Socket之心跳包实现思路