增加netty版本

2023-08-20 22:19:23 +08:00
parent a53475fef8
commit eaccee2a9b
41 changed files with 2660 additions and 15 deletions
--- a/netty-client/src/main/java/com/sv/netty/ClientThread.java
+++ b/netty-client/src/main/java/com/sv/netty/ClientThread.java
@@ -0,0 +1,121 @@
+package com.sv.netty;
+
+import com.sv.netty.config.Constant;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import java.net.InetSocketAddress;
+import java.util.concurrent.TimeUnit;
+
+
+/**
+ * 客户端通讯
+ *
+ * @author peakren
+ * @date 07/12/2017 10:12 PM
+ */
+public class ClientThread extends Thread{
+
+    private static ClientThread instance;
+
+    private volatile EventLoopGroup workerGroup;
+    private volatile Bootstrap bootstrap;
+    private volatile boolean closed = false;
+    private String remoteHost;
+    private int remotePort;
+
+    private ChannelFuture future;
+
+
+    public static ClientThread getInstance() {
+        if (instance == null) {
+            synchronized (ClientThread.class) {
+                if (instance == null) {
+                    instance = new ClientThread("127.0.0.1", 56792);
+                }
+            }
+        }
+        return instance;
+    }
+
+    private ClientThread(String remoteHost, int remotePort) {
+        this.remoteHost = remoteHost;
+        this.remotePort = remotePort;
+    }
+
+    public void run() {
+
+        closed = false;
+        workerGroup = new NioEventLoopGroup();
+        bootstrap = new Bootstrap();
+        bootstrap.group(workerGroup);
+        bootstrap.channel(NioSocketChannel.class);
+        bootstrap.option(ChannelOption.TCP_NODELAY, true);
+        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
+
+        ReConnectHandler reConnectHandler = new ReConnectHandler();
+        ClientHandler dmClientHandler = new ClientHandler();
+        ClientInitializer channelInitializer = new ClientInitializer(reConnectHandler, dmClientHandler);
+
+        bootstrap.handler(channelInitializer);
+        doConnect();
+    }
+
+    public void clearFuture(){
+        future = null;
+    }
+
+    public void doConnect() {
+        System.out.println("现在开始链接了");
+        if (closed) {
+            return;
+        }
+        System.out.println("连接 = " + remoteHost + "   " + remotePort);
+        future = bootstrap.connect(new InetSocketAddress(remoteHost, remotePort));
+        future.addListener(new ChannelFutureListener() {
+            public void operationComplete(ChannelFuture f) throws Exception {
+                f.channel().eventLoop().schedule(new Runnable() {
+                    @Override
+                    public void run() {
+                        if (!f.isSuccess()) {
+                            doConnect();
+                            System.out.println("等待连接");
+                        } else {
+                            System.out.println("已连接");
+                        }
+                    }
+                }, 2, TimeUnit.SECONDS);
+
+
+            }
+        });
+
+    }
+
+    public void close() {
+        closed = true;
+        workerGroup.shutdownGracefully();
+    }
+
+    public void restart() {
+        close();
+        run();
+    }
+
+    @ChannelHandler.Sharable
+    public class ReConnectHandler extends ChannelInboundHandlerAdapter {
+        @Override
+        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+            super.channelInactive(ctx);
+            System.out.println(ctx.toString() + "======inactive");
+            ctx.channel().eventLoop().schedule(new Runnable() {
+                @Override
+                public void run() {
+                    doConnect();
+                }
+            }, 1, TimeUnit.SECONDS);
+        }
+    }
+
+}