From cd8431ec95af934f5ba8ce4c6cadc9e0ae2d5f7e Mon Sep 17 00:00:00 2001 From: sun Date: Wed, 20 Sep 2023 17:36:31 +0800 Subject: [PATCH] feat[virtual thread]: support virtual thread in net --- .../com/zfoo/net/enhance/EnhanceUtils.java | 24 ++++++++++++++----- .../com/zfoo/net/enhance/IPacketReceiver.java | 2 ++ .../net/enhance/PacketReceiverDefinition.java | 6 +++++ .../main/java/com/zfoo/net/router/Router.java | 16 +++++++++---- 4 files changed, 38 insertions(+), 10 deletions(-) diff --git a/net/src/main/java/com/zfoo/net/enhance/EnhanceUtils.java b/net/src/main/java/com/zfoo/net/enhance/EnhanceUtils.java index 80c2404f..bba40de3 100644 --- a/net/src/main/java/com/zfoo/net/enhance/EnhanceUtils.java +++ b/net/src/main/java/com/zfoo/net/enhance/EnhanceUtils.java @@ -74,12 +74,24 @@ public abstract class EnhanceUtils { } enhanceClazz.addMethod(invokeMethod); - // 定义类实现的接口方法bus - CtMethod busMethod = new CtMethod(classPool.get(Task.class.getCanonicalName()), "task", null, enhanceClazz); - busMethod.setModifiers(Modifier.PUBLIC + Modifier.FINAL); - String busMethodBody = StringUtils.format("{ return {}.{}; }", Task.class.getCanonicalName(), definition.getTask()); - busMethod.setBody(busMethodBody); - enhanceClazz.addMethod(busMethod); + // 定义类实现的接口方法task + CtMethod taskMethod = new CtMethod(classPool.get(Task.class.getCanonicalName()), "task", null, enhanceClazz); + taskMethod.setModifiers(Modifier.PUBLIC + Modifier.FINAL); + String taskMethodBody = StringUtils.format("{ return {}.{}; }", Task.class.getCanonicalName(), definition.getTask()); + taskMethod.setBody(taskMethodBody); + enhanceClazz.addMethod(taskMethod); + + // 定义类实现的接口方法attachment + CtMethod attachmentMethod = new CtMethod(classPool.get(Class.class.getCanonicalName()), "attachment", null, enhanceClazz); + attachmentMethod.setModifiers(Modifier.PUBLIC + Modifier.FINAL); + if (attachmentClazz == null) { + String attachmentMethodBody = "{ return null; }"; + attachmentMethod.setBody(attachmentMethodBody); + } else { + String attachmentMethodBody = StringUtils.format("{ return {}.class; }", attachmentClazz.getName()); + attachmentMethod.setBody(attachmentMethodBody); + } + enhanceClazz.addMethod(attachmentMethod); enhanceClazz.detach(); diff --git a/net/src/main/java/com/zfoo/net/enhance/IPacketReceiver.java b/net/src/main/java/com/zfoo/net/enhance/IPacketReceiver.java index 6b5552a0..c26f7e9c 100644 --- a/net/src/main/java/com/zfoo/net/enhance/IPacketReceiver.java +++ b/net/src/main/java/com/zfoo/net/enhance/IPacketReceiver.java @@ -23,6 +23,8 @@ public interface IPacketReceiver { Task task(); + Class attachment(); + void invoke(Session session, Object packet, Object attachment); } diff --git a/net/src/main/java/com/zfoo/net/enhance/PacketReceiverDefinition.java b/net/src/main/java/com/zfoo/net/enhance/PacketReceiverDefinition.java index e34c965b..f9cf0ff5 100644 --- a/net/src/main/java/com/zfoo/net/enhance/PacketReceiverDefinition.java +++ b/net/src/main/java/com/zfoo/net/enhance/PacketReceiverDefinition.java @@ -66,6 +66,12 @@ public class PacketReceiverDefinition implements IPacketReceiver { return task; } + @Override + public Class attachment() { + return attachmentClazz; + } + + @Override public void invoke(Session session, Object packet, Object attachment) { if (attachmentClazz == null) { diff --git a/net/src/main/java/com/zfoo/net/router/Router.java b/net/src/main/java/com/zfoo/net/router/Router.java index b11099ce..0a6a3dc3 100644 --- a/net/src/main/java/com/zfoo/net/router/Router.java +++ b/net/src/main/java/com/zfoo/net/router/Router.java @@ -16,6 +16,7 @@ package com.zfoo.net.router; import com.zfoo.event.manager.EventBus; import com.zfoo.net.NetContext; import com.zfoo.net.anno.PacketReceiver; +import com.zfoo.net.anno.Task; import com.zfoo.net.core.event.ServerExceptionEvent; import com.zfoo.net.core.gateway.model.AuthUidToGatewayCheck; import com.zfoo.net.core.gateway.model.AuthUidToGatewayConfirm; @@ -370,14 +371,21 @@ public class Router implements IRouter { var session = packetReceiverTask.getSession(); var packet = packetReceiverTask.getPacket(); var attachment = packetReceiverTask.getAttachment(); + + // The routing of the message + var receiver = receiverMap.get(ProtocolManager.protocolId(packet.getClass())); + var threadLocalAttachment = attachment != null && receiver.task() != Task.VirtualThread; try { + // 接收者（服务器）同步和异步消息的接收 - if (attachment != null) { + if (threadLocalAttachment) { serverReceiverAttachmentThreadLocal.set(attachment); } - // The routing of the message - var receiver = receiverMap.get(ProtocolManager.protocolId(packet.getClass())); + if (receiver.task() == Task.VirtualThread && receiver.attachment() == null) { + logger.warn("virtual thread task can not set Attachment, may cause some sync and async timeout, please use attachment in receiver method signature"); + } + receiver.invoke(session, packet, attachment); } catch (Exception e) { EventBus.post(ServerExceptionEvent.valueOf(session, packet, attachment, e)); @@ -386,7 +394,7 @@ public class Router implements IRouter { logger.error(StringUtils.format("e[uid:{}][sid:{}] unknown error", session.getUid(), session.getSid(), t.getMessage()), t); } finally { // 如果有服务器在处理同步或者异步消息的时候由于错误没有返回给客户端消息，则可能会残留serverAttachment，所以先移除 - if (attachment != null) { + if (threadLocalAttachment) { serverReceiverAttachmentThreadLocal.set(null); } }