feat: fix netty 4.2 upgrade (#461)

Author: caojiajun

camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/netty/GlobalRedisProxyEnv.java

@@ -19,6 +19,7 @@
 import io.netty.channel.uring.IoUring;
 import io.netty.channel.uring.IoUringIoHandler;
 import io.netty.channel.uring.IoUringServerSocketChannel;
+import io.netty.channel.uring.IoUringServerDomainSocketChannel;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -124,9 +125,12 @@ public static void init(CamelliaServerProperties serverProperties, CamelliaTrans
             } else if (nettyTransportMode == NettyTransportMode.io_uring && isIOUringAvailable()) {
                 bossThread = serverProperties.getBossThread();
                 bossGroup = new MultiThreadIoEventLoopGroup(bossThread, new DefaultThreadFactory(BOSS_GROUP_NAME), IoUringIoHandler.newFactory());
+                udsBossGroup = bossGroup;
                 workThread = serverProperties.getWorkThread();
                 workGroup = new MultiThreadIoEventLoopGroup(workThread, new DefaultThreadFactory(WORK_GROUP_NAME), IoUringIoHandler.newFactory());
+                udsWorkGroup = workGroup;
                 serverChannelClass = IoUringServerSocketChannel.class;
+                serverUdsChannelClass = IoUringServerDomainSocketChannel.class;
                 nettyTransportMode = NettyTransportMode.io_uring;
             } else {
                 bossThread = serverProperties.getBossThread();
@@ -145,6 +149,10 @@ public static void init(CamelliaServerProperties serverProperties, CamelliaTrans
                     udsBossGroup = new MultiThreadIoEventLoopGroup(bossThread, new DefaultThreadFactory(UDS_BOSS_GROUP_NAME), KQueueIoHandler.newFactory());
                     udsWorkGroup = new MultiThreadIoEventLoopGroup(workThread, new DefaultThreadFactory(UDS_WORK_GROUP_NAME), KQueueIoHandler.newFactory());
                     serverUdsChannelClass = KQueueServerDomainSocketChannel.class;
+                } else if (isIOUringAvailable()) {
+                    udsBossGroup = new MultiThreadIoEventLoopGroup(bossThread, new DefaultThreadFactory(UDS_BOSS_GROUP_NAME), IoUringIoHandler.newFactory());
+                    udsWorkGroup = new MultiThreadIoEventLoopGroup(workThread, new DefaultThreadFactory(UDS_WORK_GROUP_NAME), IoUringIoHandler.newFactory());
+                    serverUdsChannelClass = IoUringServerDomainSocketChannel.class;
                 }
             }
             queueFactory = (QueueFactory) serverProperties.getProxyBeanFactory()

camellia-redis-proxy/camellia-redis-proxy-core/src/main/java/com/netease/nim/camellia/redis/proxy/upstream/connection/RedisConnectionHub.java

@@ -21,9 +21,7 @@
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.MultiThreadIoEventLoopGroup;
-import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.epoll.EpollIoHandler;
-import io.netty.channel.kqueue.KQueueEventLoopGroup;
 import io.netty.channel.kqueue.KQueueIoHandler;
 import io.netty.channel.nio.NioIoHandler;
 import io.netty.channel.uring.IoUringIoHandler;
@@ -107,6 +105,7 @@ public void init(CamelliaTranspondProperties properties, ProxyBeanFactory proxyB
             this.udsEventLoopGroup = eventLoopGroup;
         } else if (nettyTransportMode == NettyTransportMode.io_uring) {
             this.eventLoopGroup = new MultiThreadIoEventLoopGroup(redisConf.getDefaultTranspondWorkThread(), new DefaultThreadFactory("camellia-redis-connection"), IoUringIoHandler.newFactory());
+            this.udsEventLoopGroup = eventLoopGroup;
         } else {
             this.eventLoopGroup = new MultiThreadIoEventLoopGroup(redisConf.getDefaultTranspondWorkThread(), new DefaultThreadFactory("camellia-redis-connection"), NioIoHandler.newFactory());
         }
@@ -115,6 +114,8 @@ public void init(CamelliaTranspondProperties properties, ProxyBeanFactory proxyB
                 this.udsEventLoopGroup = new MultiThreadIoEventLoopGroup(redisConf.getDefaultTranspondWorkThread(), new DefaultThreadFactory("camellia-redis-connection-uds"), EpollIoHandler.newFactory());
             } else if (GlobalRedisProxyEnv.isKQueueAvailable()) {
                 this.udsEventLoopGroup = new MultiThreadIoEventLoopGroup(redisConf.getDefaultTranspondWorkThread(), new DefaultThreadFactory("camellia-redis-connection-uds"), KQueueIoHandler.newFactory());
+            } else if (GlobalRedisProxyEnv.isIOUringAvailable()) {
+                this.udsEventLoopGroup = new MultiThreadIoEventLoopGroup(redisConf.getDefaultTranspondWorkThread(), new DefaultThreadFactory("camellia-redis-connection-uds"), IoUringIoHandler.newFactory());
             }
         }
 
@@ -442,7 +443,11 @@ private boolean eventLoopMatch(RedisConnectionAddr addr, EventLoop eventLoop) {
             return true;
         } else if (channelType == ChannelType.uds) {
             EventLoopGroup parent = eventLoop.parent();
-            return (parent instanceof EpollEventLoopGroup) || (parent instanceof KQueueEventLoopGroup);
+            if (parent instanceof MultiThreadIoEventLoopGroup ioEventLoopGroup) {
+                return ioEventLoopGroup.isIoType(EpollIoHandler.class) || ioEventLoopGroup.isIoType(KQueueIoHandler.class)
+                        || ioEventLoopGroup.isIoType(IoUringIoHandler.class);
+            }
+            return false;
         }
         return false;
     }
@@ -461,11 +466,12 @@ private EventLoop selectEventLoop(RedisConnectionAddr addr) {
                 return loop;
             } else if (channelType == ChannelType.uds) {
                 EventLoopGroup parent = loop.parent();
-                if (parent instanceof EpollEventLoopGroup) {
-                    return loop;
-                }
-                if (parent instanceof KQueueEventLoopGroup) {
-                    return loop;
+                if (parent instanceof MultiThreadIoEventLoopGroup ioEventLoopGroup) {
+                    boolean match = ioEventLoopGroup.isIoType(EpollIoHandler.class) || ioEventLoopGroup.isIoType(KQueueIoHandler.class)
+                            || ioEventLoopGroup.isIoType(IoUringIoHandler.class);
+                    if (match) {
+                        return loop;
+                    }
                 }
                 return udsEventLoopGroup.next();
             }