优化threadlocal

jetcache-core/src/main/java/com/alicp/jetcache/ObjectPool.java

@@ -0,0 +1,52 @@
+package com.alicp.jetcache;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * @Description
+ * @author: zhangtong
+ * @create: 2023/10/6 3:27 PM
+ */
+public class ObjectPool<T> {
+        private final ArrayBlockingQueue<T> queue;
+        private final int size;
+        private final ObjectFactory<T> factory;
+        private static final Logger logger = LoggerFactory.getLogger(ObjectPool.class);
+
+        public ObjectPool(int size, ObjectFactory<T> factory) {
+            this.size = size;
+            this.factory = factory;
+            queue = new ArrayBlockingQueue<>(size);
+            for (int i = 0; i < size; i++) {
+                queue.add(factory.create());
+            }
+            logger.debug("Init the object pool with size {}", size);
+        }
+
+        public T borrowObject() {
+            T t = queue.poll();
+            if(t == null) {
+                logger.debug("The pool is not enough, create a new object");
+                return factory.create();
+            }
+            return t;
+        }
+
+        public void returnObject(T obj) {
+            if (obj == null) {
+                return;
+            }
+            factory.reset(obj);
+            queue.offer(obj);
+        }
+
+        public interface ObjectFactory<T> {
+            T create();
+            void reset(T obj);
+        }
+}

jetcache-core/src/main/java/com/alicp/jetcache/support/Kryo5ValueDecoder.java

@@ -27,16 +27,24 @@ public Object doApply(byte[] buffer) {
             in = new ByteArrayInputStream(buffer);
         }
         Input input = new Input(in);
-        Kryo kryo = (Kryo) Kryo5ValueEncoder.kryoThreadLocal.get()[0];
-        ClassLoader classLoader = Kryo5ValueDecoder.class.getClassLoader();
-        Thread t = Thread.currentThread();
-        if (t != null) {
-            ClassLoader ctxClassLoader = t.getContextClassLoader();
-            if (ctxClassLoader != null) {
-                classLoader = ctxClassLoader;
+        Kryo5ValueEncoder.Kryo5Cache kryoCache = null;
+        try {
+            kryoCache = Kryo5ValueEncoder.kryoCacheObjectPool.borrowObject();
+            Kryo kryo = kryoCache.getKryo();
+            ClassLoader classLoader = Kryo5ValueDecoder.class.getClassLoader();
+            Thread t = Thread.currentThread();
+            if (t != null) {
+                ClassLoader ctxClassLoader = t.getContextClassLoader();
+                if (ctxClassLoader != null) {
+                    classLoader = ctxClassLoader;
+                }
+            }
+            kryo.setClassLoader(classLoader);
+            return kryo.readClassAndObject(input);
+        }finally {
+            if(kryoCache != null){
+                Kryo5ValueEncoder.kryoCacheObjectPool.returnObject(kryoCache);
             }
         }
-        kryo.setClassLoader(classLoader);
-        return kryo.readClassAndObject(input);
     }
 }

jetcache-core/src/main/java/com/alicp/jetcache/support/Kryo5ValueEncoder.java

@@ -1,12 +1,11 @@
 package com.alicp.jetcache.support;
 
+import com.alicp.jetcache.ObjectPool;
 import com.alicp.jetcache.anno.SerialPolicy;
 import com.esotericsoftware.kryo.kryo5.Kryo;
 import com.esotericsoftware.kryo.kryo5.io.Output;
 import com.esotericsoftware.kryo.kryo5.serializers.CompatibleFieldSerializer;
 
-import java.lang.ref.WeakReference;
-
 /**
  * Created on 2016/10/4.
  *
@@ -16,53 +15,63 @@ public class Kryo5ValueEncoder extends AbstractValueEncoder {
 
     public static final Kryo5ValueEncoder INSTANCE = new Kryo5ValueEncoder(true);
 
-    private static int INIT_BUFFER_SIZE = 256;
-
-    static ThreadLocal<Object[]> kryoThreadLocal = ThreadLocal.withInitial(() -> {
-        Kryo kryo = new Kryo();
-        kryo.setDefaultSerializer(CompatibleFieldSerializer.class);
-        kryo.setRegistrationRequired(false);
+    private static final int INIT_BUFFER_SIZE = 256;
 
-        Output output = new Output(INIT_BUFFER_SIZE, -1);
+    //Default size = 1M
+    static ObjectPool<Kryo5Cache> kryoCacheObjectPool = new ObjectPool<>(1024*1024/INIT_BUFFER_SIZE, new ObjectPool.ObjectFactory<Kryo5Cache>() {
+        @Override
+        public Kryo5Cache create() {
+            return new Kryo5Cache();
+        }
 
-        WeakReference<Output> ref = new WeakReference<>(output);
-        return new Object[]{kryo, ref};
+        @Override
+        public void reset(Kryo5Cache obj) {
+            obj.getKryo().reset();
+            obj.getOutput().reset();
+        }
     });
 
+    public static class Kryo5Cache {
+        final Output output;
+        final Kryo kryo;
+        public Kryo5Cache(){
+            kryo = new Kryo();
+            kryo.setDefaultSerializer(CompatibleFieldSerializer.class);
+            kryo.setRegistrationRequired(false);
+            output = new Output(INIT_BUFFER_SIZE, -1);
+        }
+
+        public Output getOutput(){
+            return output;
+        }
+
+        public Kryo getKryo(){
+            return kryo;
+        }
+
+    }
+
     public Kryo5ValueEncoder(boolean useIdentityNumber) {
         super(useIdentityNumber);
     }
 
     @Override
     public byte[] apply(Object value) {
+        Kryo5Cache kryoCache = null;
         try {
-            Object[] kryoAndBuffer = kryoThreadLocal.get();
-            Kryo kryo = (Kryo) kryoAndBuffer[0];
-            WeakReference<Output> ref = (WeakReference<Output>) kryoAndBuffer[1];
-            Output output = ref.get();
-            if (output == null) {
-                output = new Output(INIT_BUFFER_SIZE, -1);
-            }
-
-            try {
-                if (useIdentityNumber) {
-                    writeInt(output, SerialPolicy.IDENTITY_NUMBER_KRYO5);
-                }
-                kryo.reset();
-                kryo.writeClassAndObject(output, value);
-                return output.toBytes();
-            } finally {
-                //reuse buffer if possible
-                output.reset();
-                if (ref.get() == null) {
-                    ref = new WeakReference<>(output);
-                    kryoAndBuffer[1] = ref;
-                }
+            kryoCache = kryoCacheObjectPool.borrowObject();
+            if (useIdentityNumber) {
+                writeInt(kryoCache.getOutput(), SerialPolicy.IDENTITY_NUMBER_KRYO5);
             }
+            kryoCache.getKryo().writeClassAndObject(kryoCache.getOutput(), value);
+            return kryoCache.getOutput().toBytes();
         } catch (Exception e) {
             StringBuilder sb = new StringBuilder("Kryo Encode error. ");
             sb.append("msg=").append(e.getMessage());
             throw new CacheEncodeException(sb.toString(), e);
+        }finally {
+            if(kryoCache != null)
+                kryoCacheObjectPool.returnObject(kryoCache);
         }
     }
 

jetcache-core/src/main/java/com/alicp/jetcache/support/KryoValueDecoder.java

@@ -27,16 +27,24 @@ public Object doApply(byte[] buffer) {
             in = new ByteArrayInputStream(buffer);
         }
         Input input = new Input(in);
-        Kryo kryo = (Kryo) KryoValueEncoder.kryoThreadLocal.get()[0];
-        ClassLoader classLoader = KryoValueDecoder.class.getClassLoader();
-        Thread t = Thread.currentThread();
-        if (t != null) {
-            ClassLoader ctxClassLoader = t.getContextClassLoader();
-            if (ctxClassLoader != null) {
-                classLoader = ctxClassLoader;
+        KryoValueEncoder.KryoCache kryoCache = null;
+        try {
+            kryoCache =  KryoValueEncoder.kryoCacheObjectPool.borrowObject();
+            Kryo kryo = kryoCache.getKryo();
+            ClassLoader classLoader = KryoValueDecoder.class.getClassLoader();
+            Thread t = Thread.currentThread();
+            if (t != null) {
+                ClassLoader ctxClassLoader = t.getContextClassLoader();
+                if (ctxClassLoader != null) {
+                    classLoader = ctxClassLoader;
+                }
+            }
+            kryo.setClassLoader(classLoader);
+            return kryo.readClassAndObject(input);
+        }finally {
+            if(kryoCache != null){
+                KryoValueEncoder.kryoCacheObjectPool.returnObject(kryoCache);
             }
         }
-        kryo.setClassLoader(classLoader);
-        return kryo.readClassAndObject(input);
     }
 }

jetcache-core/src/main/java/com/alicp/jetcache/support/KryoValueEncoder.java

@@ -1,11 +1,13 @@
 package com.alicp.jetcache.support;
 
+import com.alicp.jetcache.ObjectPool;
 import com.alicp.jetcache.anno.SerialPolicy;
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.io.Output;
 import com.esotericsoftware.kryo.serializers.CompatibleFieldSerializer;
 
 import java.lang.ref.WeakReference;
+import java.util.Arrays;
 
 /**
  * Created on 2016/10/4.
@@ -16,53 +18,60 @@ public class KryoValueEncoder extends AbstractValueEncoder {
 
     public static final KryoValueEncoder INSTANCE = new KryoValueEncoder(true);
 
-    private static int INIT_BUFFER_SIZE = 256;
+    private static final int INIT_BUFFER_SIZE = 256;
 
-    static ThreadLocal<Object[]> kryoThreadLocal = ThreadLocal.withInitial(() -> {
-        Kryo kryo = new Kryo();
-        kryo.setDefaultSerializer(CompatibleFieldSerializer.class);
-//        kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
-//        kryo.setInstantiatorStrategy(new Kryo.DefaultInstantiatorStrategy(new StdInstantiatorStrategy()));
-
-        byte[] buffer = new byte[INIT_BUFFER_SIZE];
+    //Default size = 1M
+    static ObjectPool<KryoCache> kryoCacheObjectPool = new ObjectPool<>(1024*1024/INIT_BUFFER_SIZE, new ObjectPool.ObjectFactory<KryoCache>() {
+        @Override
+        public KryoCache create() {
+            return new KryoCache();
+        }
 
-        WeakReference<byte[]> ref = new WeakReference<>(buffer);
-        return new Object[]{kryo, ref};
+        @Override
+        public void reset(KryoCache obj) {
+            obj.getKryo().reset();
+            Arrays.fill(obj.buffer, (byte) 0);
+        }
     });
 
+    public static class KryoCache {
+        final byte[] buffer;
+        final Kryo kryo;
+        public KryoCache(){
+            kryo = new Kryo();
+            kryo.setDefaultSerializer(CompatibleFieldSerializer.class);
+            buffer = new byte[INIT_BUFFER_SIZE];
+        }
+        public byte[] getBuffer(){
+            return buffer;
+        }
+        public Kryo getKryo(){
+            return kryo;
+        }
+    }
+
     public KryoValueEncoder(boolean useIdentityNumber) {
         super(useIdentityNumber);
     }
 
     @Override
     public byte[] apply(Object value) {
+        KryoCache kryoCache = null;
         try {
-            Object[] kryoAndBuffer = kryoThreadLocal.get();
-            Kryo kryo = (Kryo) kryoAndBuffer[0];
-            WeakReference<byte[]> ref = (WeakReference<byte[]>) kryoAndBuffer[1];
-            byte[] buffer = ref.get();
-            if (buffer == null) {
-                buffer = new byte[INIT_BUFFER_SIZE];
-            }
-            Output output = new Output(buffer, -1);
-
-            try {
-                if (useIdentityNumber) {
-                    writeInt(output, SerialPolicy.IDENTITY_NUMBER_KRYO4);
-                }
-                kryo.writeClassAndObject(output, value);
-                return output.toBytes();
-            } finally {
-                //reuse buffer if possible
-                if (ref.get() == null || buffer != output.getBuffer()) {
-                    ref = new WeakReference<>(output.getBuffer());
-                    kryoAndBuffer[1] = ref;
-                }
+            kryoCache = kryoCacheObjectPool.borrowObject();
+            Output output = new Output(kryoCache.getBuffer(), -1);
+            if (useIdentityNumber) {
+                writeInt(output, SerialPolicy.IDENTITY_NUMBER_KRYO4);
             }
+            kryoCache.getKryo().writeClassAndObject(output, value);
+            return output.toBytes();
         } catch (Exception e) {
             StringBuilder sb = new StringBuilder("Kryo Encode error. ");
             sb.append("msg=").append(e.getMessage());
             throw new CacheEncodeException(sb.toString(), e);
+        } finally {
+            if(kryoCache != null)
+                kryoCacheObjectPool.returnObject(kryoCache);
         }
     }
 

jetcache-test/src/test/java/com/alicp/jetcache/VirtualThreadUtil.java

@@ -0,0 +1,28 @@
+package com.alicp.jetcache;
+
+import org.springframework.cglib.core.ReflectUtils;
+
+import java.lang.reflect.Method;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @Description
+ * @author: zhangtong
+ * @create: 2023/10/5 11:44 AM
+ */
+public class VirtualThreadUtil {
+    public static ExecutorService createExecuteor(){
+        ExecutorService executorService = null;
+        try {
+            Method method = ReflectUtils.findDeclaredMethod(java.util.concurrent.Executors.class, "newVirtualThreadPerTaskExecutor", new Class[]{});
+            if (method != null) {
+                System.out.println("use newVirtualThreadPerTaskExecutor");
+                executorService = (ExecutorService) method.invoke(null);
+            }
+        }catch (Exception e){
+            return null;
+        }
+        return executorService;
+    }
+}