From 9e7c4c974891c4d096de1f312a79018faf400c31 Mon Sep 17 00:00:00 2001
From: harryV361 <916858799@qq.com>
Date: Fri, 8 Oct 2021 23:52:59 +0800
Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E4=BA=A4=E7=AC=AC=E4=B8=89=E5=91=A8?=
 =?UTF-8?q?=E4=BD=9C=E4=B8=9A?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 02nio/nio02/README.md                         |  11 +-
 02nio/nio02/pom.xml                           |  12 ++
 .../filter/HeaderHttpRequestFilter.java       |   5 +
 .../gateway/filter/HttpRequestFilter.java     |   2 +
 .../filter/TokenBucketHttpRequestFilter.java  |  25 ++++
 .../gateway/inbound/HttpInboundHandler.java   |  20 +--
 .../okhttp/OkhttpOutboundHandler.java         | 128 ++++++++++++++++++
 .../router/RandHttpEndPointRouter.java        |  30 ++++
 8 files changed, 221 insertions(+), 12 deletions(-)
 create mode 100644 02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/TokenBucketHttpRequestFilter.java
 create mode 100644 02nio/nio02/src/main/java/io/github/kimmking/gateway/router/RandHttpEndPointRouter.java

diff --git a/02nio/nio02/README.md b/02nio/nio02/README.md
index 82214410..e1bfc1fa 100644
--- a/02nio/nio02/README.md
+++ b/02nio/nio02/README.md
@@ -1,6 +1,11 @@
 # netty-gateway
 
-```
-   
+第一题答案：
+OkhttpOutboundHandler
+
+第三题答案：
+TokenBucketHttpRequestFilter
+
+第四题答案：
+RandHttpEndPointRouter
 
-```
\ No newline at end of file
diff --git a/02nio/nio02/pom.xml b/02nio/nio02/pom.xml
index 005de90a..de5c4056 100644
--- a/02nio/nio02/pom.xml
+++ b/02nio/nio02/pom.xml
@@ -57,6 +57,18 @@
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>com.squareup.okhttp3</groupId>
+            <artifactId>okhttp</artifactId>
+            <version>3.12.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>21.0</version>
+        </dependency>
     
         <!--
         <dependency>
diff --git a/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/HeaderHttpRequestFilter.java b/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/HeaderHttpRequestFilter.java
index af5c37fb..2e59e3db 100644
--- a/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/HeaderHttpRequestFilter.java
+++ b/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/HeaderHttpRequestFilter.java
@@ -9,4 +9,9 @@ public class HeaderHttpRequestFilter implements HttpRequestFilter {
     public void filter(FullHttpRequest fullRequest, ChannelHandlerContext ctx) {
         fullRequest.headers().set("mao", "soul");
     }
+
+    @Override
+    public boolean filter2(FullHttpRequest fullRequest, ChannelHandlerContext ctx) {
+        return false;
+    }
 }
diff --git a/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/HttpRequestFilter.java b/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/HttpRequestFilter.java
index 31253b40..ee00cdd2 100644
--- a/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/HttpRequestFilter.java
+++ b/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/HttpRequestFilter.java
@@ -6,5 +6,7 @@
 public interface HttpRequestFilter {
     
     void filter(FullHttpRequest fullRequest, ChannelHandlerContext ctx);
+
+    boolean filter2(FullHttpRequest fullRequest, ChannelHandlerContext ctx);
     
 }
diff --git a/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/TokenBucketHttpRequestFilter.java b/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/TokenBucketHttpRequestFilter.java
new file mode 100644
index 00000000..b9a65e80
--- /dev/null
+++ b/02nio/nio02/src/main/java/io/github/kimmking/gateway/filter/TokenBucketHttpRequestFilter.java
@@ -0,0 +1,25 @@
+package io.github.kimmking.gateway.filter;
+
+import com.google.common.util.concurrent.RateLimiter;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.FullHttpRequest;
+
+/**
+ * 使用guava令牌桶算法限流
+ *
+ * @author lizhe
+ */
+public class TokenBucketHttpRequestFilter implements HttpRequestFilter {
+
+    private static RateLimiter rateLimiter = RateLimiter.create(1.0);
+
+    @Override
+    public void filter(FullHttpRequest fullRequest, ChannelHandlerContext ctx) {
+
+    }
+
+    @Override
+    public boolean filter2(FullHttpRequest fullRequest, ChannelHandlerContext ctx) {
+        return rateLimiter.tryAcquire();
+    }
+}
diff --git a/02nio/nio02/src/main/java/io/github/kimmking/gateway/inbound/HttpInboundHandler.java b/02nio/nio02/src/main/java/io/github/kimmking/gateway/inbound/HttpInboundHandler.java
index 69b40fde..8d8e18b5 100644
--- a/02nio/nio02/src/main/java/io/github/kimmking/gateway/inbound/HttpInboundHandler.java
+++ b/02nio/nio02/src/main/java/io/github/kimmking/gateway/inbound/HttpInboundHandler.java
@@ -2,7 +2,8 @@
 
 import io.github.kimmking.gateway.filter.HeaderHttpRequestFilter;
 import io.github.kimmking.gateway.filter.HttpRequestFilter;
-import io.github.kimmking.gateway.outbound.httpclient4.HttpOutboundHandler;
+import io.github.kimmking.gateway.filter.TokenBucketHttpRequestFilter;
+import io.github.kimmking.gateway.outbound.okhttp.OkhttpOutboundHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.http.FullHttpRequest;
@@ -16,14 +17,15 @@ public class HttpInboundHandler extends ChannelInboundHandlerAdapter {
 
     private static Logger logger = LoggerFactory.getLogger(HttpInboundHandler.class);
     private final List<String> proxyServer;
-    private HttpOutboundHandler handler;
-    private HttpRequestFilter filter = new HeaderHttpRequestFilter();
-    
+    private OkhttpOutboundHandler handler;
+    private HttpRequestFilter filter = new TokenBucketHttpRequestFilter();
+
     public HttpInboundHandler(List<String> proxyServer) {
         this.proxyServer = proxyServer;
-        this.handler = new HttpOutboundHandler(this.proxyServer);
+//        this.handler = new HttpOutboundHandler(this.proxyServer);
+        this.handler = new OkhttpOutboundHandler(this.proxyServer);
     }
-    
+
     @Override
     public void channelReadComplete(ChannelHandlerContext ctx) {
         ctx.flush();
@@ -39,10 +41,10 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
 //            if (uri.contains("/test")) {
 //                handlerTest(fullRequest, ctx);
 //            }
-    
+
             handler.handle(fullRequest, ctx, filter);
-    
-        } catch(Exception e) {
+
+        } catch (Exception e) {
             e.printStackTrace();
         } finally {
             ReferenceCountUtil.release(msg);
diff --git a/02nio/nio02/src/main/java/io/github/kimmking/gateway/outbound/okhttp/OkhttpOutboundHandler.java b/02nio/nio02/src/main/java/io/github/kimmking/gateway/outbound/okhttp/OkhttpOutboundHandler.java
index 5f194588..1a714801 100644
--- a/02nio/nio02/src/main/java/io/github/kimmking/gateway/outbound/okhttp/OkhttpOutboundHandler.java
+++ b/02nio/nio02/src/main/java/io/github/kimmking/gateway/outbound/okhttp/OkhttpOutboundHandler.java
@@ -1,4 +1,132 @@
 package io.github.kimmking.gateway.outbound.okhttp;
 
+import io.github.kimmking.gateway.filter.HttpRequestFilter;
+import io.github.kimmking.gateway.outbound.httpclient4.NamedThreadFactory;
+import io.github.kimmking.gateway.router.HttpEndpointRouter;
+import io.github.kimmking.gateway.router.RandHttpEndPointRouter;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpUtil;
+import okhttp3.*;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.*;
+
+import static io.netty.handler.codec.http.HttpResponseStatus.NO_CONTENT;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+
 public class OkhttpOutboundHandler {
+
+    private OkHttpClient client;
+    private ExecutorService proxyService;
+    private List<String> backendUrls;
+    private HttpEndpointRouter router;
+
+    public OkhttpOutboundHandler(List<String> backendUrls) {
+        this.backendUrls = backendUrls;
+
+        ConnectionPool pool = new ConnectionPool(5, 10, TimeUnit.MINUTES);
+        client = new OkHttpClient().newBuilder().readTimeout(2000, TimeUnit.MILLISECONDS)
+                .connectionPool(pool).retryOnConnectionFailure(false).followRedirects(true).build();
+
+        int cores = Runtime.getRuntime().availableProcessors() * 2;
+        long keepAliveTime = 1000;
+        int queueSize = 2048;
+        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
+        proxyService = new ThreadPoolExecutor(cores, cores,
+                keepAliveTime, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(queueSize),
+                new NamedThreadFactory("proxyService"), handler);
+
+        router = new RandHttpEndPointRouter();
+    }
+
+    private String formatUrl(String backend) {
+        return backend.endsWith("/") ? backend.substring(0, backend.length() - 1) : backend;
+    }
+
+
+    public void handle(final FullHttpRequest fullRequest, final ChannelHandlerContext ctx, HttpRequestFilter filter)
+            throws Exception {
+
+        String backendUrl = router.route(this.backendUrls);
+        final String url = backendUrl + fullRequest.uri();
+
+        if (filter.filter2(fullRequest, ctx)) {
+
+            proxyService.submit(() -> fetchGet(fullRequest, ctx, url));
+        } else {
+            limited(ctx);
+        }
+    }
+
+    private void fetchGet(final FullHttpRequest inbound, final ChannelHandlerContext ctx, final String url) {
+        try {
+
+            final Request request = new Request.Builder().url(url).get().build();
+            final Call call = client.newCall(request);
+            call.enqueue(new Callback() {
+                @Override
+                public void onFailure(Call call, IOException e) {
+                    e.printStackTrace();
+                }
+
+                @Override
+                public void onResponse(Call call, Response response) throws IOException {
+                    try {
+                        handleResponse(inbound, ctx, response);
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                    }
+                }
+            });
+        }catch (Exception e){
+            e.printStackTrace();
+        }
+    }
+
+    private void limited(final ChannelHandlerContext ctx) {
+        byte[] bytes = "被限流...".getBytes();
+        FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(bytes));
+
+        response.headers().set("Content-Type", "application/json");
+        response.headers().setInt("Content-Length", bytes.length);
+
+        ctx.write(response);
+        ctx.flush();
+    }
+
+    private void handleResponse(final FullHttpRequest fullRequest, final ChannelHandlerContext ctx, final Response endpointResponse) throws Exception {
+        FullHttpResponse response = null;
+        try {
+            String body = endpointResponse.body().string();
+            System.out.println(body);
+            byte[] bodys = body.getBytes();
+
+            response = new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(bodys));
+
+            response.headers().set("Content-Type", "application/json");
+            response.headers().setInt("Content-Length", Integer.parseInt(endpointResponse.header("Content-Length")));
+
+
+        } catch (Exception e) {
+            e.printStackTrace();
+            response = new DefaultFullHttpResponse(HTTP_1_1, NO_CONTENT);
+            ctx.close();
+        } finally {
+            if (fullRequest != null) {
+                if (!HttpUtil.isKeepAlive(fullRequest)) {
+                    ctx.write(response).addListener(ChannelFutureListener.CLOSE);
+                } else {
+                    ctx.write(response);
+                }
+            }
+            ctx.flush();
+        }
+    }
 }
diff --git a/02nio/nio02/src/main/java/io/github/kimmking/gateway/router/RandHttpEndPointRouter.java b/02nio/nio02/src/main/java/io/github/kimmking/gateway/router/RandHttpEndPointRouter.java
new file mode 100644
index 00000000..c3518363
--- /dev/null
+++ b/02nio/nio02/src/main/java/io/github/kimmking/gateway/router/RandHttpEndPointRouter.java
@@ -0,0 +1,30 @@
+package io.github.kimmking.gateway.router;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @author lizhe
+ */
+public class RandHttpEndPointRouter implements HttpEndpointRouter {
+
+    private AtomicInteger atomicInteger = new AtomicInteger(0);
+
+
+    @Override
+    public String route(List<String> endpoints) {
+
+        final int modulo = incrementAndGetModulo();
+        return endpoints.get(modulo % endpoints.size());
+    }
+
+    public final int incrementAndGetModulo() {
+        int current;
+        int next;
+        do {
+            current = this.atomicInteger.get();
+            next = current >= Integer.MAX_VALUE ? 0 : current + 1;
+        } while (!this.atomicInteger.compareAndSet(current, next));
+        return next;
+    }
+}