添加监控

forezp · forezp · commit 1e7a2be1b574 · 2018-06-05T19:49:14.000+08:00
diff --git a/README.md b/README.md
@@ -1,7 +1,7 @@
 # BootNettyRpc
 
 [![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg?label=license)](https://github.com/forezp/BootNettyRpc/blob/master/LICENSE)
-[![Maven Central](https://img.shields.io/maven-central/v/com.nepxion/thunder.svg?label=maven%20central)](http://mvnrepository.com/artifact/io.github.forezp/boot-netty-rpc-core)
+[![Maven Central](https://img.shields.io/maven-central/v/io.github.forezp/boot-netty-rpc-core.svg?label=maven%20central)](http://mvnrepository.com/artifact/io.github.forezp/boot-netty-rpc-core)
 
 
 README: [English](https://github.com/forezp/BootNettyRpc/blob/master/README-en.md) | [中文](https://github.com/forezp/BootNettyRpc/blob/master/README.md)
diff --git a/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/common/constant/ConfigConstants.java b/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/common/constant/ConfigConstants.java
@@ -27,7 +27,7 @@ public class ConfigConstants {
     public static final String SERVER_POOL_KEEPALIVE_TIME = "netty.server.pool.keepalive.time";
     public static final String SERVER_POOL_KEEPALIVE_TIME_DEFAULT = "1800000";
     public static final String SERVER_POOL_REJECT_TYPE = "netty.server.pool.reject.type";
-    public static final String SERVER_POOL_REJECT_TYPE_DEFAULT = "DiscardedPolicyWithReport";
+    public static final String SERVER_POOL_REJECT_TYPE_DEFAULT = "BlockingPolicyWithReport";
 
     public static final String CLIENT_POOL_CORE_SIZE = "netty.client.pool.coresize";
     public static final String CLIENT_POOL_CORE_SIZE_DEFAULT = "4";
@@ -40,7 +40,7 @@ public class ConfigConstants {
     public static final String CLIENT_POOL_KEEPALIVE_TIME = "netty.client.pool.keepalive.time";
     public static final String CLIENT_POOL_KEEPALIVE_TIME_DEFAULT = "1800000";
     public static final String CLIENT_POOL_REJECT_TYPE = "netty.client.pool.reject.type";
-    public static final String CLIENT_POOL_REJECT_TYPE_DEFAULT = "DiscardedPolicyWithReport";
+    public static final String CLIENT_POOL_REJECT_TYPE_DEFAULT = "BlockingPolicyWithReport";
 
     public static final String NETTY_CLIENT_RENEW_INTERVAL = "netty.client.renew.interval";
     public static final String NETTY_CLIENT_RENEW_INTERVAL_DEFAULT = "30";//second
diff --git a/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/common/thread/ThreadPoolFactory.java b/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/common/thread/ThreadPoolFactory.java
@@ -36,22 +36,22 @@ public static ThreadPoolExecutor createServerPoolExecutor(String interfaceName)
 
 
     public static ThreadPoolExecutor createThreadPoolDefaultExecutor() {
-        return createThreadPoolExecutor(CommonProperties.CPUS * 1,
+        return createThreadPoolExecutor( CommonProperties.CPUS * 1,
                 CommonProperties.CPUS * 2,
                 15 * 60 * 1000,
-                false);
+                false );
     }
 
 
-    public static ThreadPoolExecutor createThreadPoolExecutor( int corePoolSize,  int maximumPoolSize, long keepAliveTime, boolean allowCoreThreadTimeOut) {
+    public static ThreadPoolExecutor createThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, boolean allowCoreThreadTimeOut) {
 
-        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize,
+        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( corePoolSize,
                 maximumPoolSize,
                 keepAliveTime,
                 TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue(  ) ,
-                new BlockingPolicyWithReport());
-        threadPoolExecutor.allowCoreThreadTimeOut(allowCoreThreadTimeOut);
+                new LinkedBlockingQueue(),
+                new BlockingPolicyWithReport() );
+        threadPoolExecutor.allowCoreThreadTimeOut( allowCoreThreadTimeOut );
 
         return threadPoolExecutor;
     }
@@ -68,18 +68,18 @@ public static ThreadPoolExecutor createThreadPoolExecutor(Map<String, ThreadPool
                                                               int coreSize, int maxSize, Long keepAliveTime, String queneType,
                                                               int queueSize, String rejectType) {
 
-        if (executorMap.get( interfaceName ) != null) {
-            return executorMap.get( interfaceName );
-        } else {
+        ThreadPoolExecutor threadPoolExecutor = executorMap.get( interfaceName );
+        if (threadPoolExecutor == null) {
             ThreadPoolExecutor newThreadPool = new ThreadPoolExecutor( CommonProperties.CPUS * coreSize
                     , CommonProperties.CPUS * maxSize, keepAliveTime, TimeUnit.MILLISECONDS
                     , createQuene( queneType, queueSize ), createThreadFactory( interfaceName )
                     , createRejectedExecutionHandler( rejectType ) );
-            if (newThreadPool != null) {
-                executorMap.putIfAbsent( interfaceName, newThreadPool );
+            threadPoolExecutor = executorMap.putIfAbsent( interfaceName, newThreadPool );
+            if (threadPoolExecutor == null) {
+                threadPoolExecutor = newThreadPool;
             }
-            return newThreadPool;
         }
+        return threadPoolExecutor;
     }
 
     private static RejectedExecutionHandler createRejectedExecutionHandler(String rejectedPolicy) {
diff --git a/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/protocol/client/RequestInterceptor.java b/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/protocol/client/RequestInterceptor.java
@@ -6,6 +6,7 @@
 import io.netty.channel.ChannelFuture;
 
 import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
 
 /**
  * ${DESCRIPTION}
@@ -30,7 +31,7 @@ public Object invokeSync(ChannelFuture channelFuture, NettyRpcRequest request) t
         cacheContainer.getSyncEntityMap().put( messageId, responseSyncEntity );
         try {
             invokeAsync( channelFuture, request );
-            barrier.await();//TODO 超时没有处理
+            barrier.await( 5000, TimeUnit.MILLISECONDS );//TODO 5s 超时没有处理
             return responseSyncEntity.getResult();
         } catch (Exception e) {
             e.printStackTrace();
diff --git a/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/protocol/client/RequstInvocationHandler.java b/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/protocol/client/RequstInvocationHandler.java
@@ -22,6 +22,7 @@
 
 /**
  * 参考了 https://www.cnblogs.com/clonen/p/6735011.html
+ *
  * @author fangzhipeng
  * create 2018-05-21
  **/
@@ -66,7 +67,7 @@ public Object run(Method method, Object[] args) throws Exception {
         NettyRpcRequest request = buildRequest( entity, method, args );
         ConnectionEntity connectionEntity = findCandidateConnection( entity );
         if (connectionEntity != null) {
-            LOG.info( "run connectionEntity:" + connectionEntity.toString() );
+            //  LOG.info( "run connectionEntity:" + connectionEntity.toString() );
             RequestInterceptor interceptor = (RequestInterceptor) NettyRpcApplication.getBean( "requestInterceptor" );
             if (request.isSyn()) {
                 Object object = interceptor.invokeSync( connectionEntity.getChannelFuture(), request );
@@ -84,14 +85,19 @@ public Object run(Method method, Object[] args) throws Exception {
             BeanUtils.copyProperties( entity, appEntity );
             ExcutorContainer excutorContainer = (ExcutorContainer) NettyRpcApplication.getBean( "excutorContainer" );
             NettyClientExcutor excutor = (NettyClientExcutor) excutorContainer.getClientExcutor();
-            try {
-                excutor.start( appEntity );
-                while (!excutor.started( appEntity )) {
+            synchronized (RequstInvocationHandler.class) {
+                try {
+                    if (excutor.started( appEntity )) {
+                        return run( method, args );
+                    }
+                    excutor.start( appEntity );
+                    while (!excutor.started( appEntity )) {
+                    }
+                    return run( method, args );
+                } catch (Exception e) {
+                    e.printStackTrace();
+                    return null;
                 }
-                return run( method, args );
-            } catch (Exception e) {
-                e.printStackTrace();
-                return null;
             }
         }
     }
diff --git a/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/protocol/client/ResponseHandler.java b/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/protocol/client/ResponseHandler.java
@@ -17,7 +17,7 @@ public class ResponseHandler extends AbstractResponseHandler {
     @Override
     public void handle(NettyRpcResponse response) {
         response.setEndTime(System.currentTimeMillis());
-        LOG.info("Client received: " + response.toString());
+      //  LOG.info("Client received: " + response.toString());
         LOG.info("耗时: " + (response.getEndTime() - response.getStartTime()) + "ms");
         //处理response
         if (response.isSyn()) {
diff --git a/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/protocol/server/NettyServerHandler.java b/boot-netty-rpc-core/src/main/java/io/github/forezp/netty/rpc/core/protocol/server/NettyServerHandler.java
@@ -37,7 +37,7 @@ public void channelRead0(final ChannelHandlerContext ctx, final NettyRpcRequest
             @Override
             public Object call() throws Exception {
 
-                LOG.info( "Server received: " + request.toString() );
+               // LOG.info( "Server received: " + request.toString() );
                 NettyRpcResponse response = new NettyRpcResponse( 1, "sucess", null );
                 excutorContainer.getServerRequestHandler().handle( request, response );
                 ctx.writeAndFlush( response );
diff --git a/examples/example-rpc-lib/src/main/java/com/forezp/examplerpclib/lib/IGreeting.java b/examples/example-rpc-lib/src/main/java/com/forezp/examplerpclib/lib/IGreeting.java
@@ -8,7 +8,7 @@
  * @author fangzhipeng
  * @create 2018-05-18
  **/
-@RpcClient(name = "server", rpcClz = "com.forezp.examplerpcserver.api.Greeting")
+@RpcClient(name = "server", rpcClz = "com.forezp.examplerpcserver.api.Greeting", isSyn = false)
 public interface IGreeting {
 
     String sayHello(String name);
diff --git a/examples/local-rpc-lient/src/main/java/com/forezp/localrpclient/LocalRpcLientApplication.java b/examples/local-rpc-lient/src/main/java/com/forezp/localrpclient/LocalRpcLientApplication.java
@@ -4,6 +4,7 @@
 import com.forezp.examplerpclib.lib.IGreeting;
 import io.github.forezp.netty.rpc.core.annotation.EnableNettyRpc;
 import io.github.forezp.netty.rpc.core.common.dto.RespDTO;
+import io.github.forezp.netty.rpc.core.common.thread.ThreadPoolFactory;
 import io.github.forezp.netty.rpc.core.protocol.client.Invoker;
 import io.github.forezp.netty.rpc.core.monitor.MonitorMessage;
 import org.springframework.boot.SpringApplication;
@@ -13,26 +14,34 @@
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;
 
+import java.util.Random;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
+
 @SpringBootApplication
 @RestController
 @EnableNettyRpc(basePackages = "com.forezp")
 public class LocalRpcLientApplication {
 
     public static void main(String[] args) {
-        SpringApplication.run(LocalRpcLientApplication.class, args);
+        SpringApplication.run( LocalRpcLientApplication.class, args );
     }
 
     @GetMapping("/test")
     public RespDTO test() throws Exception {
 
-        IGreeting invoker = (IGreeting) Invoker.invoke(IGreeting.class);
-        Object result = invoker.sayHello("sww");
-        return RespDTO.success(result);
+        IGreeting invoker = (IGreeting) Invoker.invoke( IGreeting.class );
+        Object result = invoker.sayHello( "sww" );
+
+
+        return RespDTO.success( result );
     }
 
     @PostMapping("/postmsg")
     public void post(@RequestBody MonitorMessage message) {
-        System.out.println(JSON.toJSONString(message));
+        //System.out.println(JSON.toJSONString(message));
     }
 
 }
