adapter add brpc

dromara · Dec 16, 2022 · 2fee25c · 2fee25c
1 parent 21a5218
commit 2fee25c
Show file tree

Hide file tree

Showing 25 changed files with 495 additions and 130 deletions.
diff --git a/README.md b/README.md
@@ -84,7 +84,7 @@
 
 - **支持多配置中心**：基于主流配置中心实现线程池参数动态调整，实时生效，已支持 Nacos、Apollo、Zookeeper、Consul、Etcd，同时也提供 SPI 接口可自定义扩展实现
 
-- **中间件线程池管理**：集成管理常用第三方组件的线程池，已集成 Tomcat、Jetty、Undertow、Dubbo、RocketMq、Hystrix、Grpc、Motan、Okhttp3 等组件的线程池管理（调参、监控报警）
+- **中间件线程池管理**：集成管理常用第三方组件的线程池，已集成 Tomcat、Jetty、Undertow、Dubbo、RocketMq、Hystrix、Grpc、Motan、Okhttp3、Brpc 等组件的线程池管理（调参、监控报警）
 
 ---
 

diff --git a/adapter/adapter-brpc/pom.xml b/adapter/adapter-brpc/pom.xml
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>cn.dynamictp</groupId>
+        <artifactId>dynamic-tp-adapter</artifactId>
+        <version>1.0.9</version>
+    </parent>
+    <artifactId>dynamic-tp-adapter-brpc</artifactId>
+
+    <properties>
+        <brpc.version>2022.2.0</brpc.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>cn.dynamictp</groupId>
+            <artifactId>dynamic-tp-adapter-common</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.baidu.cloud</groupId>
+            <artifactId>spring-cloud-starter-baidu-starlight</artifactId>
+            <version>${brpc.version}</version>
+            <scope>compile</scope>
+            <optional>true</optional>
+        </dependency>
+    </dependencies>
+
+</project>
diff --git a/...ter/adapter-brpc/src/main/java/com/dtp/apapter/brpc/client/StarlightClientDtpAdapter.java b/...ter/adapter-brpc/src/main/java/com/dtp/apapter/brpc/client/StarlightClientDtpAdapter.java
@@ -0,0 +1,81 @@
+package com.dtp.apapter.brpc.client;
+
+import cn.hutool.core.collection.CollUtil;
+import com.baidu.cloud.starlight.api.rpc.StarlightClient;
+import com.baidu.cloud.starlight.api.rpc.threadpool.ThreadPoolFactory;
+import com.baidu.cloud.starlight.core.rpc.SingleStarlightClient;
+import com.baidu.cloud.starlight.springcloud.client.cluster.SingleStarlightClientManager;
+import com.dtp.adapter.common.AbstractDtpAdapter;
+import com.dtp.common.ApplicationContextHolder;
+import com.dtp.common.dto.ExecutorWrapper;
+import com.dtp.common.properties.DtpProperties;
+import com.dtp.common.util.ReflectionUtil;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * StarlightClientDtpAdapter related
+ *
+ * @author yanhom
+ * @since 1.1.0
+ */
+@Slf4j
+public class StarlightClientDtpAdapter extends AbstractDtpAdapter {
+
+    private static final String NAME = "brpcClientTp";
+
+    private static final String THREAD_POOL_FIELD = "threadPoolOfAll";
+
+    @Override
+    public void refresh(DtpProperties dtpProperties) {
+        refresh(NAME, dtpProperties.getBrpcTp(), dtpProperties.getPlatforms());
+    }
+
+    @Override
+    protected void initialize() {
+        super.initialize();
+
+        SingleStarlightClientManager sscManager = null;
+        Map<String, StarlightClient> scBeans = Maps.newHashMap();
+        try {
+            sscManager = ApplicationContextHolder.getBean(SingleStarlightClientManager.class);
+            scBeans = ApplicationContextHolder.getBeansOfType(StarlightClient.class);
+        } catch (Exception e) {
+            log.warn("getBean error, msg: {}", e.getMessage());
+        }
+
+        List<StarlightClient> starlightClients = Lists.newArrayList();
+        if (CollUtil.isNotEmpty(scBeans)) {
+            starlightClients.addAll(scBeans.values());
+        }
+        if (Objects.nonNull(sscManager) && CollUtil.isNotEmpty(sscManager.allSingleClients())) {
+            starlightClients.addAll(sscManager.allSingleClients().values());
+        }
+        if (CollUtil.isEmpty(starlightClients)) {
+            log.warn("Cannot find beans of type StarlightClient.");
+            return;
+        }
+
+        starlightClients.forEach(v -> {
+            val threadPoolFactory = (ThreadPoolFactory) ReflectionUtil.getFieldValue(SingleStarlightClient.class,
+                    THREAD_POOL_FIELD, v);
+            if (Objects.isNull(threadPoolFactory)) {
+                return;
+            }
+            String bizThreadPoolName = v.remoteURI().getParameter("biz_thread_pool_name") + "#client";
+            val executor = threadPoolFactory.defaultThreadPool();
+            if (Objects.nonNull(executor)) {
+                val executorWrapper = new ExecutorWrapper(bizThreadPoolName, executor);
+                initNotifyItems(bizThreadPoolName, executorWrapper);
+                executors.put(bizThreadPoolName, executorWrapper);
+            }
+        });
+        log.info("DynamicTp adapter, brpc client executors init end, executors: {}", executors);
+    }
+}
diff --git a/...ter/adapter-brpc/src/main/java/com/dtp/apapter/brpc/server/StarlightServerDtpAdapter.java b/...ter/adapter-brpc/src/main/java/com/dtp/apapter/brpc/server/StarlightServerDtpAdapter.java
@@ -0,0 +1,77 @@
+package com.dtp.apapter.brpc.server;
+
+import com.baidu.cloud.starlight.api.common.URI;
+import com.baidu.cloud.starlight.api.rpc.StarlightServer;
+import com.baidu.cloud.starlight.api.rpc.threadpool.ThreadPoolFactory;
+import com.baidu.cloud.starlight.api.transport.ServerPeer;
+import com.baidu.cloud.starlight.core.rpc.DefaultStarlightServer;
+import com.baidu.cloud.starlight.core.rpc.ServerProcessor;
+import com.baidu.cloud.starlight.transport.netty.NettyServer;
+import com.dtp.adapter.common.AbstractDtpAdapter;
+import com.dtp.common.ApplicationContextHolder;
+import com.dtp.common.dto.ExecutorWrapper;
+import com.dtp.common.properties.DtpProperties;
+import com.dtp.common.util.ReflectionUtil;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+
+import java.util.Objects;
+
+/**
+ * StarlightServerDtpAdapter related
+ *
+ * @author yanhom
+ * @since 1.1.0
+ */
+@Slf4j
+public class StarlightServerDtpAdapter extends AbstractDtpAdapter {
+
+    private static final String NAME = "brpcServerTp";
+
+    private static final String URI_FIELD = "uri";
+
+    private static final String SERVER_PEER_FIELD = "serverPeer";
+
+    private static final String THREAD_POOL_FACTORY_FIELD = "threadPoolFactory";
+
+    @Override
+    public void refresh(DtpProperties dtpProperties) {
+        refresh(NAME, dtpProperties.getBrpcTp(), dtpProperties.getPlatforms());
+    }
+
+    @Override
+    protected void initialize() {
+        super.initialize();
+
+        val bean = ApplicationContextHolder.getBean(StarlightServer.class);
+        if (!(bean instanceof DefaultStarlightServer)) {
+            return;
+        }
+        val starlightServer = (DefaultStarlightServer) bean;
+        val uri = (URI) ReflectionUtil.getFieldValue(DefaultStarlightServer.class,
+                URI_FIELD, starlightServer);
+        val serverPeer = (ServerPeer) ReflectionUtil.getFieldValue(DefaultStarlightServer.class,
+                SERVER_PEER_FIELD, starlightServer);
+
+        if (Objects.isNull(uri) || Objects.isNull(serverPeer) || !(serverPeer instanceof NettyServer)) {
+            return;
+        }
+        val processor = (ServerProcessor) serverPeer.getProcessor();
+        if (Objects.isNull(processor)) {
+            return;
+        }
+        val threadPoolFactory = (ThreadPoolFactory) ReflectionUtil.getFieldValue(ServerProcessor.class,
+                THREAD_POOL_FACTORY_FIELD, processor);
+        if (Objects.isNull(threadPoolFactory)) {
+            return;
+        }
+        String bizThreadPoolName = uri.getParameter("biz_thread_pool_name") + "#server";
+        val executor = threadPoolFactory.defaultThreadPool();
+        if (Objects.nonNull(executor)) {
+            val executorWrapper = new ExecutorWrapper(bizThreadPoolName, executor);
+            initNotifyItems(bizThreadPoolName, executorWrapper);
+            executors.put(bizThreadPoolName, executorWrapper);
+        }
+        log.info("DynamicTp adapter, brpc server executors init end, executors: {}", executors);
+    }
+}
diff --git a/adapter/pom.xml b/adapter/pom.xml
@@ -19,6 +19,7 @@
         <module>adapter-grpc</module>
         <module>adapter-motan</module>
         <module>adapter-okhttp3</module>
+        <module>adapter-brpc</module>
     </modules>
 
     <dependencies>

diff --git a/common/src/main/java/com/dtp/common/properties/DtpProperties.java b/common/src/main/java/com/dtp/common/properties/DtpProperties.java
@@ -127,6 +127,11 @@ public class DtpProperties {
      */
     private List<SimpleTpProperties> okhttp3Tp;
 
+    /**
+     * Brpc thread pools.
+     */
+    private List<SimpleTpProperties> brpcTp;
+
     /**
      * Notify platform configs.
      */

diff --git a/core/src/main/java/com/dtp/core/thread/OrderedDtpExecutor.java b/core/src/main/java/com/dtp/core/thread/OrderedDtpExecutor.java
@@ -19,7 +19,6 @@
  * are run in a certain order.
  *
  * @author dragon-zhang
- * @date 2022/12/12 09:46
  */
 @Slf4j
 public class OrderedDtpExecutor extends DtpExecutor {

diff --git a/docs/docs/README.md b/docs/docs/README.md
@@ -70,7 +70,7 @@ features:
 
 - **支持多配置中心**：基于主流配置中心实现线程池参数动态调整，实时生效，已支持 Nacos、Apollo、Zookeeper、Consul、Etcd，同时也提供 SPI 接口可自定义扩展实现
 
-- **中间件线程池管理**：集成管理常用第三方组件的线程池，已集成Tomcat、Jetty、Undertow、Dubbo、RocketMq、Hystrix、Grpc、Motan、Okhttp3 等组件的线程池管理（调参、监控报警）
+- **中间件线程池管理**：集成管理常用第三方组件的线程池，已集成Tomcat、Jetty、Undertow、Dubbo、RocketMq、Hystrix、Grpc、Motan、Okhttp3、Brpc 等组件的线程池管理（调参、监控报警）
 
 
 # 技术架构 

diff --git a/docs/docs/guide/introduction/functions.md b/docs/docs/guide/introduction/functions.md
@@ -50,4 +50,4 @@ star: true
 
 - **支持多配置中心**：基于主流配置中心实现线程池参数动态调整，实时生效，已支持 Nacos、Apollo、Zookeeper、Consul、Etcd，同时也提供 SPI 接口可自定义扩展实现
 
-- **中间件线程池管理**：集成管理常用第三方组件的线程池，已集成Tomcat、Jetty、Undertow、Dubbo、RocketMq、Hystrix、Grpc、Motan、Okhttp3 等组件的线程池管理（调参、监控报警）
+- **中间件线程池管理**：集成管理常用第三方组件的线程池，已集成Tomcat、Jetty、Undertow、Dubbo、RocketMq、Hystrix、Grpc、Motan、Okhttp3、Brpc 等组件的线程池管理（调参、监控报警）
diff --git a/docs/docs/guide/middleware/middleware.md b/docs/docs/guide/middleware/middleware.md
@@ -37,6 +37,8 @@ star: true
 8. Motan 线程池管理
 
 9. Okhttp3 线程池管理
+
+10. Brpc 线程池管理
 :::
 
 依赖如下，使用时需要手动引入对应依赖
@@ -96,13 +98,21 @@ star: true
         <version>1.1.0</version>
     </dependency>
 ```
+
+```xml
+    <dependency>
+        <groupId>cn.dynamictp</groupId>
+        <artifactId>dynamic-tp-spring-boot-starter-adapter-brpc</artifactId>
+        <version>1.1.0</version>
+    </dependency>
+```
 ::: tip
 
 1.三方组件线程池配置请参考 快速使用 / 配置文件
 
 2.Tomcat、Jetty、Undertow 线程池目前只享有动态调参和监控功能，没通知报警功能
 
-3.Dubbo、RocketMq、Hystrix、Grpc、Motan、Okhttp3 享有动态调参、监控、通知告警完整的功能
+3.Dubbo、RocketMq、Hystrix、Grpc、Motan、Okhttp3、Brpc 享有动态调参、监控、通知告警完整的功能
 
 4.注意配置时 threadPoolName 规则，配置文件有注释
 

diff --git a/docs/docs/guide/use/config.md b/docs/docs/guide/use/config.md
@@ -82,11 +82,16 @@ spring:
           corePoolSize: 100
           maximumPoolSize: 200
           keepAliveTime: 60
-      okhttp3Tp:                                      # okhttp3 线程池配置
+      okhttp3Tp:                                    # okhttp3 线程池配置
         - threadPoolName: okHttpClientTp
           corePoolSize: 100
           maximumPoolSize: 200
           keepAliveTime: 60
+      brpcTp:                                       # brpc 线程池配置
+        - threadPoolName: biz1#server               # 名称规则：biz_thread_pool_name + "#" + client/server
+          corePoolSize: 100
+          maximumPoolSize: 200
+          keepAliveTime: 60
       dubboTp:                                     # dubbo 线程池配置
         - threadPoolName: dubboTp#20880            # 名称规则：dubboTp + "#" + 协议端口
           threadPoolAliasName: 测试线程池            

diff --git a/example/example-adapter/pom.xml b/example/example-adapter/pom.xml
@@ -13,6 +13,7 @@
         <maven.deploy.skip>true</maven.deploy.skip>
         <motan.version>1.1.13</motan.version>
         <okhttp.version>3.14.9</okhttp.version>
+        <brpc.version>2022.2.0</brpc.version>
     </properties>
 
     <dependencies>
@@ -95,6 +96,12 @@
             <version>${dtp.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>cn.dynamictp</groupId>
+            <artifactId>dynamic-tp-spring-boot-starter-adapter-brpc</artifactId>
+            <version>${dtp.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>cn.dynamictp</groupId>
             <artifactId>dynamic-tp-spring-boot-starter-extension-limiter-redis</artifactId>
@@ -202,6 +209,17 @@
             <version>${okhttp.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.baidu.cloud</groupId>
+            <artifactId>spring-cloud-starter-baidu-starlight</artifactId>
+            <version>${brpc.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>ch.qos.logback</groupId>
+                    <artifactId>logback-classic</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
     </dependencies>
 
     <build>

diff --git a/example/example-adapter/src/main/java/com/dtp/example/adapter/ExampleAdapterApplication.java b/example/example-adapter/src/main/java/com/dtp/example/adapter/ExampleAdapterApplication.java
@@ -1,5 +1,6 @@
 package com.dtp.example.adapter;
 
+import com.baidu.cloud.starlight.springcloud.server.annotation.StarlightScan;
 import com.dtp.core.spring.EnableDynamicTp;
 import org.apache.dubbo.config.spring.context.annotation.EnableDubbo;
 import org.springframework.boot.SpringApplication;
@@ -12,6 +13,7 @@
 @EnableDubbo
 @ImportResource(locations = {"classpath:motan_server.xml"})
 @SpringBootApplication
+@StarlightScan
 public class ExampleAdapterApplication {
 
     public static void main(String[] args) {

diff --git a/example/example-adapter/src/main/java/com/dtp/example/adapter/brpc/BrpcClientService.java b/example/example-adapter/src/main/java/com/dtp/example/adapter/brpc/BrpcClientService.java
@@ -0,0 +1,30 @@
+package com.dtp.example.adapter.brpc;
+
+import com.baidu.cloud.starlight.springcloud.client.annotation.RpcProxy;
+import org.springframework.stereotype.Service;
+
+/**
+ * BrpcClientService related
+ *
+ * @author yanhom
+ * @since 1.1.0
+ */
+@Service
+public class BrpcClientService {
+
+    /**
+     * 使用注解引用服务，指定服务端IP Port，采用brpc协议调用
+     */
+    @RpcProxy(remoteUrl = "localhost:8777", protocol = "brpc")
+    private UserService userService;
+
+    /**
+     * 使用注解引用服务，指定服务端IP Port，采用springrest(http)协议调用
+     */
+    @RpcProxy(remoteUrl = "localhost:8777", protocol = "springrest")
+    private UserService restUserService;
+
+    public String getUserName(Long userId) {
+        return userService.getUserName(userId);
+    }
+}
diff --git a/example/example-adapter/src/main/java/com/dtp/example/adapter/brpc/UserService.java b/example/example-adapter/src/main/java/com/dtp/example/adapter/brpc/UserService.java
@@ -0,0 +1,12 @@
+package com.dtp.example.adapter.brpc;
+
+/**
+ * UserService related
+ *
+ * @author yanhom
+ * @since 1.1.0
+ */
+public interface UserService {
+
+    String getUserName(Long userId);
+}