From 4bf3959c33853e223aa0a6bfec2f91ba4db30343 Mon Sep 17 00:00:00 2001 From: NetRiceCake Date: Mon, 8 Dec 2025 01:50:26 +0900 Subject: [PATCH] Virtual thread related Using asynchronous methods for logic such as checking if room information exists may lead to unexpected issues. --- .../github/netricecake/kakao/TalkClient.java | 7 ++++--- .../github/netricecake/loco/LocoSocket.java | 21 ++++++++++++------- .../netricecake/loco/codec/LocoCodec.java | 8 +++++-- 3 files changed, 23 insertions(+), 13 deletions(-) diff --git a/src/main/java/com/github/netricecake/kakao/TalkClient.java b/src/main/java/com/github/netricecake/kakao/TalkClient.java index 1a817dd..0aa8958 100644 --- a/src/main/java/com/github/netricecake/kakao/TalkClient.java +++ b/src/main/java/com/github/netricecake/kakao/TalkClient.java @@ -32,6 +32,7 @@ import java.security.MessageDigest; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; public class TalkClient { @@ -91,7 +92,7 @@ public class TalkClient { public void onError(Exception e) { e.printStackTrace(); } - }); + }, Executors.newFixedThreadPool(1)); byte[] body = new CheckInOut(loginData.userId).toBson(); checkInSocket.connect(); LocoPacket checkinResponse = checkInSocket.writeAndRead(new LocoPacket(1000, "CHECKIN", body)); @@ -111,7 +112,7 @@ public class TalkClient { rp = ByteUtil.hexStringToByteArray("0100ffff0100"); // 이게 도대체 뭐임 } - socket = new LocoSocket(checkInData.getHost(), checkInData.getPort(), new LocoSocketHandlerImpl(this)); + socket = new LocoSocket(checkInData.getHost(), checkInData.getPort(), new LocoSocketHandlerImpl(this), Executors.newFixedThreadPool(1)); socket.connect(); LoginListOut req = new LoginListOut(); req.setDuuid(deviceUuid); @@ -182,7 +183,7 @@ public class TalkClient { int status = jsonObject.get("status").getAsInt(); future.complete(status); } - }); + }, Executors.newFixedThreadPool(1)); postSocket.connect(); PostOut po = new PostOut(); diff --git a/src/main/java/com/github/netricecake/loco/LocoSocket.java b/src/main/java/com/github/netricecake/loco/LocoSocket.java index 3c6a726..68f8673 100644 --- a/src/main/java/com/github/netricecake/loco/LocoSocket.java +++ b/src/main/java/com/github/netricecake/loco/LocoSocket.java @@ -36,14 +36,17 @@ public class LocoSocket { private final LocoSocketHandler locoSocketHandler; + private final ExecutorService handlerLoop; + private final Map> waitList = new HashMap<>(); private int packetIdCounter = 1000; - public LocoSocket(String ip, int port, LocoSocketHandler locoSocketHandler) { + public LocoSocket(String ip, int port, LocoSocketHandler locoSocketHandler, ExecutorService handlerLoop) { this.ip = ip; this.port = port; this.locoSocketHandler = locoSocketHandler; + this.handlerLoop = handlerLoop; } public void connect() throws IOException { @@ -66,10 +69,10 @@ public class LocoSocket { alive = true; channel.writeAndFlush(cryptoManager.generateHandshakeMessage()).sync(); channel.pipeline().addLast(new SecureLayerCodec(cryptoManager)); - channel.pipeline().addLast(new LocoCodec(locoSocketHandler, waitList)); - Thread.ofVirtual().start(locoSocketHandler::onConnect); + channel.pipeline().addLast(new LocoCodec(locoSocketHandler, waitList, handlerLoop)); + handlerLoop.execute(locoSocketHandler::onConnect); - final CompletableFuture closeFuture = new CompletableFuture<>(); // 네티 퓨처 sync함수가 virtual thread에서 제대로 작동하지 않습니다.(쓰레드 양보를 안함) 그래서 이렇게 해야됨 + final CompletableFuture closeFuture = new CompletableFuture<>(); // 네티 퓨처 sync함수가 virtual thread에서 제대로 작동하지 않습니다.(쓰레드 양보를 안함) channel.closeFuture().addListener(future -> { closeFuture.complete(null); }); @@ -77,7 +80,7 @@ public class LocoSocket { try { closeFuture.get(); eventLoopGroup.shutdownGracefully(); - locoSocketHandler.onDisconnect(); + handlerLoop.execute(locoSocketHandler::onDisconnect); alive = false; } catch (Exception e) { locoSocketHandler.onError(e); @@ -85,7 +88,7 @@ public class LocoSocket { }); } catch (InterruptedException e) { - Thread.ofVirtual().start(() -> { + handlerLoop.execute(() -> { locoSocketHandler.onError(e); }); } @@ -114,7 +117,9 @@ public class LocoSocket { waitList.remove(packetId); return result; } catch (Exception e) { - locoSocketHandler.onError(e); + handlerLoop.execute(() -> { + locoSocketHandler.onError(e); + }); } return null; } @@ -123,7 +128,7 @@ public class LocoSocket { if (!alive) return; eventLoopGroup.shutdownGracefully(); channel.close(); - Thread.ofVirtual().start(locoSocketHandler::onDisconnect); + handlerLoop.execute(locoSocketHandler::onDisconnect); alive = false; } diff --git a/src/main/java/com/github/netricecake/loco/codec/LocoCodec.java b/src/main/java/com/github/netricecake/loco/codec/LocoCodec.java index 253569f..f842037 100644 --- a/src/main/java/com/github/netricecake/loco/codec/LocoCodec.java +++ b/src/main/java/com/github/netricecake/loco/codec/LocoCodec.java @@ -10,6 +10,7 @@ import io.netty.handler.codec.MessageToMessageCodec; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; public class LocoCodec extends MessageToMessageCodec { @@ -21,9 +22,12 @@ public class LocoCodec extends MessageToMessageCodec { private final LocoSocketHandler locoSocketHandler; - public LocoCodec(LocoSocketHandler locoSocketHandler, Map> waitList) { + private final ExecutorService handlerLoop; + + public LocoCodec(LocoSocketHandler locoSocketHandler, Map> waitList, ExecutorService handlerLoop) { this.locoSocketHandler = locoSocketHandler; this.waitList = waitList; + this.handlerLoop = handlerLoop; } @Override @@ -72,7 +76,7 @@ public class LocoCodec extends MessageToMessageCodec { ((CompletableFuture) waitList.get(currentLocoPacket.getPacketId())).complete(currentLocoPacket); } else { final LocoPacket p = currentLocoPacket; - Thread.ofVirtual().start(() -> { + handlerLoop.execute(() -> { locoSocketHandler.onPacket(p); }); }