Spring Boot整合Thrift RPC

Spring Boot自动配置简介

在介绍RPC之前，我们先来学习下Spring Boot的自动配置。

我们前面已经提到：Spring Boot来源于Spring，并且做了众多改进，其中最有用的设计理念是约定优于配置，它通过自动配置功能（大多数开发者平时习惯设置的配置作为默认配置）为开发者快速、准确地构建出标准化的应用。

以集成MySQL数据库为例，在Spring Boot出现之前，我们要

  1. 配置JDBC驱动依赖
  2. 配置XML文件中数据源
  3. 配置XML中的DataSource Bean
  4. 配置XML中的XXXTemplate Bean
  5. 配置XML中的XXXTransactionManager Bean

有了Spring Boot的自动配置后，自动配置帮我们生成了各种DataSource、XXXTemplate、XXXTransactionManager，我们所需要做的只有一条，就是激活它

  1. maven中依赖包含自动配置的包
  2. 配置JDBC驱动依赖
  3. yaml文件中定义数据源

自动配置进行智能检测，只要满足上述3个条件，其他的Bean都会被自动生成并注入到Spring环境中。我们需要使用时只需要@Autowired一下就可以了，是不是非常简单！

由于篇幅所限，本书不会对自动配置的书写做零起点教学，如果你想了解自动配置的原理，可以参考这篇文章spring boot实战(第十三篇)自动配置原理分析

在本节的后续部分，我们会以Thrift RPC Server为例，看看自动配置是如何书写的。

RPC简介

远程过程调用(remote procedure call或简称RPC)，指的是运行于本地(客户端)的程序像调用本地程序一样，直接调用另一台计算机(服务器端)的程序，而程序员无需额外为远程交互做额外的编程。

RPC极大地简化了分布似乎系统中节点之间网络通信的开发工作量，是微服务架构中的重要组件之一。

在本书中，我们选用Thrift作为RPC框架。由于篇幅所限，我们不会对Thrift RPC作出详尽的介绍，如果你还不熟悉，可以参考官方的快速入门文档。

Spring Boot整合Thrift RPC服务端

简要来说，启动一个Thrift RPC的服务端需要如下步骤:

  1. 书写DSL(.thrift文件)，定义函数、数据结构等。
  2. 编译并生成桩代码。
  3. 编写Handler(RPC的逻辑入口)。
  4. 基于上述Handler，构造Processor。
  5. 构造Server，Thrift提供了多种服务端供选择，常用的有TThreadPoolServer(多线程服务器)和TNonblockingServer(非阻塞服务器)。
  6. 设置Server的Protocol，类似的，Thrift提供了多种传输协议，最常用的是TBinaryProtocol和TCompactProtocol。
  7. 设置Server的Transport(Factory)，用这种方式指定底层的传输协议，常用的有TFramedTransport、TNonBlockingTransport，不同的Transport可以类似Java的IOStreawm方式，相互叠加，以产生更强大的效果。

上述对Thrift服务器的架构做了简要介绍，如果想更深入了解，可以自行阅读官方源码。

首先，我们来看一下thrift定义(根据上一节的介绍，文件放在lmsia-abc-common包中)

  1. namespace java com.coder4.lmsia.abc
  3. service lmsiaAbcThrift {
  5.     string sayHi()
  6. }

调用thrift进行编译后，我们也将对应的桩文件放置在lmsia-abc-client下，目录结构可以参见上一节。

为了更方便的在Spring Boot中集成Thrift服务器，我将相应代码抽取成了公用库lmsia-thrift-server

  2. ├── build.gradle
  3. ├── gradle
  4. │   └── wrapper
  5. │       ├── gradle-wrapper.jar
  6. │       └── gradle-wrapper.properties
  7. ├── gradlew
  8. ├── gradlew.bat
  9. ├── README.md
  10. ├── settings.gradle
  11. └── src
  12.     ├── main
  13.     │   ├── java
  14.     │   │   └── com
  15.     │   │       └── coder4
  16.     │   │           └── lmsia
  17.     │   │               └── thrift
  18.     │   │                   └── server
  19.     │   │                       ├── configuration
  20.     │   │                       │   └── ThriftServerConfiguration.java
  21.     │   │                       └── ThriftServerRunnable.java
  22.     │   └── resources
  23.     │       └── META-INF
  24.     │           └── spring.factories
  25.     └── test
  26.         └── java

简单解析下项目结构：gradle相关: 与前节介绍的类似，只不过这里是单项目功能。ThriftServerConfiguration: 自动配置，当满足条件后会自动激活，激活后可自动启动Thrift RPC服务。ThriftServerRunnable: Thrift RPC服务器的构造逻辑、运行线程。spring.factories: 当我们以类库方式提供自动配置时，需要增加这个spring.factories，让别的项目能”定位到”要检查的自动配置。

首先，我们来看一下ThriftServerRunnable.java

  2. package com.coder4.lmsia.thrift.server;
  4. import org.apache.thrift.TProcessor;
  5. import org.apache.thrift.protocol.TBinaryProtocol;
  6. import org.apache.thrift.protocol.TProtocolFactory;
  7. import org.apache.thrift.server.TServer;
  8. import org.apache.thrift.server.TThreadedSelectorServer;
  9. import org.apache.thrift.transport.TFramedTransport;
  10. import org.apache.thrift.transport.TNonblockingServerSocket;
  11. import org.apache.thrift.transport.TNonblockingServerTransport;
  12. import org.apache.thrift.transport.TTransportException;
  13. import org.apache.thrift.transport.TTransportFactory;
  15. import java.util.concurrent.ExecutorService;
  16. import java.util.concurrent.SynchronousQueue;
  17. import java.util.concurrent.ThreadPoolExecutor;
  18. import java.util.concurrent.TimeUnit;
  20. /**
  21.  * @author coder4
  22.  */
  23. public class ThriftServerRunnable implements Runnable {
  25.     private static final int THRIFT_PORT = 3000;
  27.     private static final int THRIFT_TIMEOUT = 5000;
  29.     private static final int THRIFT_TCP_BACKLOG = 5000;
  31.     private static final int THRIFT_CORE_THREADS = 128;
  33.     private static final int THRIFT_MAX_THREADS = 256;
  35.     private static final int THRIFT_SELECTOR_THREADS = 16;
  37.     private static final TProtocolFactory THRIFT_PROTOCOL_FACTORY = new TBinaryProtocol.Factory();
  39.     // 16MB
  40.     private static final int THRIFT_MAX_FRAME_SIZE = 16 * 1024 * 1024;
  42.     // 4MB
  43.     private static final int THRIFT_MAX_READ_BUF_SIZE = 4 * 1024 * 1024;
  45.     protected ExecutorService threadPool;
  47.     protected TServer server;
  49.     protected Thread thread;
  51.     private TProcessor processor;
  53.     private boolean isDestroy = false;
  55.     public ThriftServerRunnable(TProcessor processor) {
  56.         this.processor = processor;
  57.     }
  59.     public TServer build() throws TTransportException {
  60.         TNonblockingServerSocket.NonblockingAbstractServerSocketArgs socketArgs =
  61.                 new TNonblockingServerSocket.NonblockingAbstractServerSocketArgs();
  62.         socketArgs.port(THRIFT_PORT);
  63.         socketArgs.clientTimeout(THRIFT_TIMEOUT);
  64.         socketArgs.backlog(THRIFT_TCP_BACKLOG);
  66.         TNonblockingServerTransport transport = new TNonblockingServerSocket(socketArgs);
  68.         threadPool =
  69.                 new ThreadPoolExecutor(THRIFT_CORE_THREADS, THRIFT_MAX_THREADS,
  70.                         60L, TimeUnit.SECONDS,
  71.                         new SynchronousQueue<>());
  73.         TTransportFactory transportFactory = new TFramedTransport.Factory(THRIFT_MAX_FRAME_SIZE);
  74.         TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(transport)
  75.                 .selectorThreads(THRIFT_SELECTOR_THREADS)
  76.                 .executorService(threadPool)
  77.                 .transportFactory(transportFactory)
  78.                 .inputProtocolFactory(THRIFT_PROTOCOL_FACTORY)
  79.                 .outputProtocolFactory(THRIFT_PROTOCOL_FACTORY)
  80.                 .processor(processor);
  82.         args.maxReadBufferBytes = THRIFT_MAX_READ_BUF_SIZE;
  84.         return new TThreadedSelectorServer(args);
  85.     }
  87.     @Override
  88.     public void run() {
  89.         try {
  90.             server = build();
  91.             server.serve();
  92.         } catch (Exception e) {
  93.             e.printStackTrace();
  94.             throw new RuntimeException("Start Thrift RPC Server Exception");
  95.         }
  96.     }
  98.     public void stop() throws Exception {
  99.         threadPool.shutdown();
  100.         server.stop();
  101.     }
  103. }

我们来解释一下：

  • build方法用于构造一个可供运行的Thrift RPC Server
    1. 构造非阻塞Socket，并设置监听端口、超时
    2. 构造非阻塞Transport
    3. 构造线程池，在这里我们的服务器模型是非阻塞线程池RPC服务器。
    4. 构造底层传输协议即TFramedTransport
    5. 构造ThriftServer，并设置前面构造的非阻塞Transport、线程池、协议TBinaryProtocol
  • 整个ThriftServerRunnable类是一个线程Runnablerun，run函数中构造RPC服务，并启动服务(servee)
  • stop服务提供停止服务的方法

下面我们来看一下自动配置ThriftServerConfiguration.java：

  1. package com.coder4.lmsia.thrift.server.configuration;
  3. import com.coder4.lmsia.thrift.server.ThriftServerRunnable;
  4. import org.apache.thrift.TProcessor;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.beans.factory.DisposableBean;
  8. import org.springframework.beans.factory.InitializingBean;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
  11. import org.springframework.context.annotation.Configuration;
  13. import java.util.concurrent.TimeUnit;
  15. /**
  16.  * @author coder4
  17.  */
  18. @Configuration
  19. @ConditionalOnBean(value = {TProcessor.class})
  20. public class ThriftServerConfiguration implements InitializingBean, DisposableBean {
  22.     private Logger LOG = LoggerFactory.getLogger(ThriftServerConfiguration.class);
  24.     private static final int GRACEFUL_SHOWDOWN_SEC = 3;
  26.     @Autowired
  27.     private TProcessor processor;
  29.     private ThriftServerRunnable thriftServer;
  31.     private Thread thread;
  33.     @Override
  34.     public void destroy() throws Exception {
  35.         LOG.info("Wait for graceful shutdown on destroy(), {} seconds", GRACEFUL_SHOWDOWN_SEC);
  36.         Thread.sleep(TimeUnit.SECONDS.toMillis(GRACEFUL_SHOWDOWN_SEC));
  37.         LOG.info("Shutdown rpc server.");
  38.         thriftServer.stop();
  39.         thread.join();
  40.     }
  42.     @Override
  43.     public void afterPropertiesSet() throws Exception {
  44.         thriftServer = new ThriftServerRunnable(processor);
  45.         thread = new Thread(thriftServer);
  46.         thread.start();
  47.     }
  48. }

这是我们编写的第一个自动配置，我们稍微详细的解释一下：

  • 启动条件: 仅当服务提供了TProcessor才启用，我们稍后会在lmsia-abc项目中看到，后者封装了RPC的桩入口，提供了TProcessor。
  • InitializingBean: 自动配置实现了InitializingBean，为什么要实现这个接口呢？当这个自动配置被初始化时，所有Autowired的属性被自动注入（即Processor），而前面ThriftServerRunnable中我么已经看到，只有拿到了TProcessor，才能启动RPC服务。因此，我们使用了InitializingBean，它自带了afterPropertiesSet这个回调，会在所有属性被注入完成后，调用这个回调函数。
    • 在这里，我们调用了ThriftServerRunnable实现了Thrift RPC服务器的启动。
  • DisposableBean: 除了InitializingBean，我们还实现了DisposableBean。看名字就可以知道，这是Spring为了服务关闭时清理资源而设计的接口。事实也是如此，当服务关闭时，会依次调用每个自动配置，如果实现了DisposableBean，则回调destroy函数。
    • 在这里，我们先让线程休眠3秒，然后才关闭Thrift RPC服务，这主要是为了Graceful Shutdown而设计的(“优雅关闭”)，关于这一点，我们会在下一节会做详细讲解。

最后，我们的自动配置默认是无法被发现的，需要一个配置文件spring.factories：

  1. org.springframework.boot.autoconfigure.EnableAutoConfiguration=com.coder4.lmsia.thrift.server.configuration.ThriftServerConfiguration

解读完lmsia-thrift-server后，我们看看如何将它整合进lmsia-abc项目中。

  1. 在lmsia-abc-server子项目中的build.gradle中加入：

    1. compile 'com.github.liheyuan:lmsia-thrift-server:0.0.1'

  2. 提供一个TProcessor，如前文所述，这是启用自动配置的必要条件，ThriftProcessorConfiguration:```javapackage com.coder4.lmsia.abc.server.configuration;

import com.coder4.lmsia.abc.thrift.LmsiaAbcThrift;import com.coder4.lmsia.abc.server.thrift.ThriftServerHandler;import org.apache.thrift.TProcessor;import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;

/**

  • @author coder4*/@Configuration@ConditionalOnProperty(name = “thriftServer.enabled”, matchIfMissing = true)public class ThriftProcessorConfiguration {

    @Bean(name = “thriftProcessor”) public TProcessor processor(ThriftServerHandler handler) {

    1.  return new LmsiaAbcThrift.Processor(handler);

    }

}

  2. 我们简单解释下：
  3. * 这也是一个自动配置，仅当配置文件中thriftServer.enabled=true时才启用(不配置默认true)
  4. * 提供的TProcessor，需要依赖ThriftServerHandler，这个就是Thrift生成的桩函数，项目结构分析中已经提到过，这是RPC服务器的逻辑入口。
  6. 怎么样，使用了自动配置后，启动一个Thrift 服务器是不是非常简单？
  8. ## Spring Boot整合Thrift RPC客户端
  10. 只有服务端是不行的，还需要有客户端。
  12. 类似地，为了方便的生成客户端，我们把代码进行了整理和抽象，放到了[lmsia-thrift-client](https://github.com/liheyuan/lmsia-thrift-client)项目中。
  14. 首先看一下项目结构：
  15. ```shell
  16. ├── build.gradle
  17. ├── gradle
  18. │   └── wrapper
  19. │       ├── gradle-wrapper.jar
  20. │       └── gradle-wrapper.properties
  21. ├── gradlew
  22. ├── gradlew.bat
  23. ├── README.md
  24. ├── settings.gradle
  25. └── src
  26.     ├── main
  27.     │   ├── java
  28.     │   │   └── com
  29.     │   │       └── coder4
  30.     │   │           └── lmsia
  31.     │   │               └── thrift
  32.     │   │                   └── client
  33.     │   │                       ├── ThriftClient.java
  34.     │   │                       ├── AbstractThriftClient.java
  35.     │   │                       ├── EasyThriftClient.java
  36.     │   │                       ├── K8ServiceThriftClient.java
  37.     │   │                       ├── K8ServiceKey.java
  38.     │   │                       ├── builder
  39.     │   │                       │   ├── EasyThriftClientBuilder.java
  40.     │   │                       │   └── K8ServiceThriftClientBuilder.java
  41.     │   │                       ├── func
  42.     │   │                       │   ├── ThriftCallFunc.java
  43.     │   │                       │   └── ThriftExecFunc.java
  44.     │   │                       ├── pool
  45.     │   │                       │   ├── TTransportPoolFactory.java
  46.     │   │                       │   └── TTransportPool.java
  47.     │   │                       └── utils
  48.     │   │                           └── ThriftUrlStr.java
  49.     │   └── resources
  50.     └── test
  51.         └── java
  52.             └── LibraryTest.java

解释下项目结构：

  • gradle相关的与之前类似，不再赘述
  • ThriftClient相关，定义了Thrift的客户端
    1. ThriftClient 抽象了客户端的接口
    2. AbstractThriftClient 实现了除连接外的Thrift Client操作
    3. EasyThriftClient 使用IP和端口直连的Thrift Client
    4. K8ServiceThriftClient 使用Kubernetes服务名字(根据微服务自动发现一节中的介绍，服务名字实际也是Host)和端口的Thrift Client，并内置了连接池。
  • func 函数编程工具类
  • builder 方便快速构造上述两种Thrift Client
  • pool 客户端连接池

本小节主要对IP、端口直连的客户端即EasyThriftClient进行介绍。关于支持服务自动发现以及连接池功能的K8ServiceThriftClient，将在下一节进行介绍。

先看一下接口定义，ThriftClient:

  1. package com.coder4.lmsia.thrift.client;
  3. import com.coder4.lmsia.thrift.client.func.ThriftCallFunc;
  4. import com.coder4.lmsia.thrift.client.func.ThriftExecFunc;
  5. import org.apache.thrift.TServiceClient;
  7. import java.util.concurrent.Future;
  9. /**
  10.  * @author coder4
  11.  */
  12. public interface ThriftClient<TCLIENT extends TServiceClient> {
  14.     /**
  15.      * sync call with return value
  16.      * @param tcall thrift rpc client call
  17.      * @param <TRET> return type
  18.      * @return
  19.      */
  20.     <TRET> TRET call(ThriftCallFunc<TCLIENT, TRET> tcall);
  22.     /**
  23.      * sync call without return value
  24.      * @param texec thrift rpc client
  25.      */
  26.     void exec(ThriftExecFunc<TCLIENT> texec);
  28.     /**
  29.      * async call with return value
  30.      * @param tcall thrift rpc client call
  31.      * @param <TRET>
  32.      * @return
  33.      */
  34.     <TRET> Future<TRET> asyncCall(ThriftCallFunc<TCLIENT, TRET> tcall);
  37.     /**
  38.      * asnyc call without return value
  39.      * @param texec thrift rpc client call
  40.      */
  41.     <TRET> Future<?> asyncExec(ThriftExecFunc<TCLIENT> texec);
  43. }

这里需要解释一下，上述实际分成了两大类:

  • exec 无返回值的rpc调用
  • call 有返回值的调用

这里使用了Java 8的函数式编程进行抽象。如果不太熟悉的朋友，可以自行查阅相关资料。

在函数式编程的帮助下，我们可以将每一个rpc调用都分为同步和异步两种，异步的调用会返回一个Future。

再来看一下AbstractThriftClient:

  1. /**
  2.  * @(#)AbstractThriftClient.java, Aug 01, 2017.
  3.  * <p>
  4.  * Copyright 2017 fenbi.com. All rights reserved.
  5.  * FENBI.COM PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
  6.  */
  7. package com.coder4.lmsia.thrift.client;
  9. import com.coder4.lmsia.thrift.client.func.ThriftCallFunc;
  10. import com.coder4.lmsia.thrift.client.func.ThriftExecFunc;
  11. import org.apache.thrift.TServiceClient;
  12. import org.apache.thrift.TServiceClientFactory;
  13. import org.apache.thrift.protocol.TBinaryProtocol;
  14. import org.apache.thrift.protocol.TProtocol;
  15. import org.apache.thrift.transport.TTransport;
  17. import java.util.concurrent.ExecutorService;
  18. import java.util.concurrent.Future;
  19. import java.util.concurrent.LinkedBlockingDeque;
  20. import java.util.concurrent.ThreadPoolExecutor;
  21. import java.util.concurrent.TimeUnit;
  23. /**
  24.  * @author coder4
  25.  */
  26. public abstract class AbstractThriftClient<TCLIENT extends TServiceClient> implements ThriftClient<TCLIENT> {
  28.     protected static final int THRIFT_CLIENT_DEFAULT_TIMEOUT = 5000;
  30.     protected static final int THRIFT_CLIENT_DEFAULT_MAX_FRAME_SIZE = 1024 * 1024 * 16;
  32.     private Class<?> thriftClass;
  34.     private static final TBinaryProtocol.Factory protocolFactory = new TBinaryProtocol.Factory();
  36.     private TServiceClientFactory<TCLIENT> clientFactory;
  38.     // For async call
  39.     private ExecutorService threadPool;
  41.     public void init() {
  42.         try {
  43.             clientFactory = getThriftClientFactoryClass().newInstance();
  44.         } catch (Exception e) {
  45.             throw new RuntimeException();
  46.         }
  48.         if (!check()) {
  49.             throw new RuntimeException("Client config failed check!");
  50.         }
  52.         threadPool = new ThreadPoolExecutor(
  53.                 10, 100, 0,
  54.                 TimeUnit.MICROSECONDS, new LinkedBlockingDeque<>());
  55.     }
  57.     protected boolean check() {
  58.         if (thriftClass == null) {
  59.             return false;
  60.         }
  61.         return true;
  62.     }
  64.     @Override
  65.     public <TRET> Future<TRET> asyncCall(ThriftCallFunc<TCLIENT, TRET> tcall) {
  66.         return threadPool.submit(() -> this.call(tcall));
  67.     }
  69.     @Override
  70.     public <TRET> Future<?> asyncExec(ThriftExecFunc<TCLIENT> texec) {
  71.         return threadPool.submit(() -> this.exec(texec));
  72.     }
  74.     protected TCLIENT createClient(TTransport transport) throws Exception {
  75.         // Step 1: get TProtocol
  76.         TProtocol protocol = protocolFactory.getProtocol(transport);
  78.         // Step 2: get client
  79.         return clientFactory.getClient(protocol);
  80.     }
  82.     private Class<TServiceClientFactory<TCLIENT>> getThriftClientFactoryClass() {
  83.         Class<TCLIENT> clientClazz = getThriftClientClass();
  84.         if (clientClazz == null) {
  85.             return null;
  86.         }
  87.         for (Class<?> clazz : clientClazz.getDeclaredClasses()) {
  88.             if (TServiceClientFactory.class.isAssignableFrom(clazz)) {
  89.                 return (Class<TServiceClientFactory<TCLIENT>>) clazz;
  90.             }
  91.         }
  92.         return null;
  93.     }
  95.     private Class<TCLIENT> getThriftClientClass() {
  96.         for (Class<?> clazz : thriftClass.getDeclaredClasses()) {
  97.             if (TServiceClient.class.isAssignableFrom(clazz)) {
  98.                 return (Class<TCLIENT>) clazz;
  99.             }
  100.         }
  101.         return null;
  102.     }
  104.     public void setThriftClass(Class<?> thriftClass) {
  105.         this.thriftClass = thriftClass;
  106.     }
  107. }

上述抽象的Thrift客户端实现了如下功能：

  1. 客户端线程池，这里主要是为异步调用准备的，与之前构造的服务端的线程池是完全不同的。
    • asyncCall和asyncExec使用了线程池来完成异步调用
  2. thriftClass 存储了Thrift的桩代码了类，不同业务生成的ThriftClass不一样，所以这里存储了class。
  3. createClient提供了共用函数，传入一个transport，即可构造生成一个Thrift Client，特别注意的是，这里设定的通信协议为TBinaryProtocol，必须与服务端保持一致，否则无法成功通信。

由于call和exec与连接实现较为相关，因此并未在这一层中实现，最后我们来看一下EasyThriftClient:

  1. package com.coder4.lmsia.thrift.client;
  3. import com.coder4.lmsia.thrift.client.func.ThriftCallFunc;
  4. import com.coder4.lmsia.thrift.client.func.ThriftExecFunc;
  5. import org.apache.thrift.TServiceClient;
  6. import org.apache.thrift.transport.TFramedTransport;
  7. import org.apache.thrift.transport.TSocket;
  8. import org.apache.thrift.transport.TTransport;
  10. /**
  11.  * @author coder4
  12.  */
  13. public class EasyThriftClient<TCLIENT extends TServiceClient> extends AbstractThriftClient<TCLIENT> {
  15.     private static final int EASY_THRIFT_BUFFER_SIZE = 1024 * 16;
  17.     protected String thriftServerHost;
  19.     protected int thriftServerPort;
  21.     @Override
  22.     protected boolean check() {
  23.         if (thriftServerHost == null || thriftServerHost.isEmpty()) {
  24.             return false;
  25.         }
  26.         if (thriftServerPort <= 0) {
  27.             return false;
  28.         }
  29.         return super.check();
  30.     }
  32.     private TTransport borrowTransport() throws Exception {
  33.         TSocket socket = new TSocket(thriftServerHost, thriftServerPort, THRIFT_CLIENT_DEFAULT_TIMEOUT);
  35.         TTransport transport = new TFramedTransport(
  36.                 socket, THRIFT_CLIENT_DEFAULT_MAX_FRAME_SIZE);
  38.         transport.open();
  40.         return transport;
  41.     }
  43.     private void returnTransport(TTransport transport) {
  44.         if (transport != null && transport.isOpen()) {
  45.             transport.close();
  46.         }
  47.     }
  49.     private void returnBrokenTransport(TTransport transport) {
  50.         if (transport != null && transport.isOpen()) {
  51.             transport.close();
  52.         }
  53.     }
  55.     @Override
  56.     public <TRET> TRET call(ThriftCallFunc<TCLIENT, TRET> tcall) {
  58.         // Step 1: get TTransport
  59.         TTransport tpt = null;
  60.         try {
  61.             tpt = borrowTransport();
  62.         } catch (Exception e) {
  63.             throw new RuntimeException(e);
  64.         }
  66.         // Step 2: get client & call
  67.         try {
  68.             TCLIENT tcli = createClient(tpt);
  69.             TRET ret = tcall.call(tcli);
  70.             returnTransport(tpt);
  71.             return ret;
  72.         } catch (Exception e) {
  73.             returnBrokenTransport(tpt);
  74.             throw new RuntimeException(e);
  75.         }
  76.     }
  78.     @Override
  79.     public void exec(ThriftExecFunc<TCLIENT> texec) {
  80.         // Step 1: get TTransport
  81.         TTransport tpt = null;
  82.         try {
  83.             tpt = borrowTransport();
  84.         } catch (Exception e) {
  85.             throw new RuntimeException(e);
  86.         }
  88.         // Step 2: get client & exec
  89.         try {
  90.             TCLIENT tcli = createClient(tpt);
  91.             texec.exec(tcli);
  92.             returnTransport(tpt);
  93.         } catch (Exception e) {
  94.             returnBrokenTransport(tpt);
  95.             throw new RuntimeException(e);
  96.         }
  97.     }
  99.     public String getThriftServerHost() {
  100.         return thriftServerHost;
  101.     }
  103.     public void setThriftServerHost(String thriftServerHost) {
  104.         this.thriftServerHost = thriftServerHost;
  105.     }
  107.     public int getThriftServerPort() {
  108.         return thriftServerPort;
  109.     }
  111.     public void setThriftServerPort(int thriftServerPort) {
  112.         this.thriftServerPort = thriftServerPort;
  113.     }

简单解释下上述代码

  1. 需要外部传入RPC服务器的主机名和端口 thriftServerHost和thriftServerPort
  2. borrowTransport完成Transport(Thrift中类似Socket的抽象) 的构造，注意这里要使用TFramedTransport，与之前服务端的构造保持一致。
  3. returnTransport关闭Transport
  4. returnBrokenTransport关闭出异常的Transport
  5. call和exec 在拿到Transport后，使用函数式编程的方式，完成rpc调用，如果有异常则关闭连接。

最后我们来看一下对应的Builder，EasyThriftClientBuilder:

  1. package com.coder4.lmsia.thrift.client.builder;
  3. import com.coder4.lmsia.thrift.client.EasyThriftClient;
  4. import org.apache.thrift.TServiceClient;
  6. /**
  7.  * @author coder4
  8.  */
  9. public class EasyThriftClientBuilder<TCLIENT extends TServiceClient> {
  11.     private final EasyThriftClient<TCLIENT> client = new EasyThriftClient<>();
  13.     protected EasyThriftClient<TCLIENT> build() {
  14.         client.init();
  15.         return client;
  16.     }
  18.     protected EasyThriftClientBuilder<TCLIENT> setHost(String host) {
  19.         client.setThriftServerHost(host);
  20.         return this;
  21.     }
  23.     protected EasyThriftClientBuilder<TCLIENT> setPort(int port) {
  24.         client.setThriftServerPort(port);
  25.         return this;
  26.     }
  28.     protected EasyThriftClientBuilder<TCLIENT> setThriftClass(Class<?> thriftClass) {
  29.         client.setThriftClass(thriftClass);
  30.         return this;
  31.     }
  32. }

Builder的代码比较简单，就是以链式调用的方式，通过主机和端口，方便地构造一个EasyThriftClient。

看了EasyThriftClient后下面我们来看一下如何集成到项目中。

在Gradle子项目划分与微服务的代码结构一节中，我们已经提到，将每个微服务的RPC客户端放在xx-client子工程中，现在我们再来回顾下lmsia-abc-client的目录结构。

  1. ├── build.gradle
  2. └── src
  3.     ├── main
  4.     │   ├── java
  5.     │   │   └── com
  6.     │   │       └── coder4
  7.     │   │           └── lmsia
  8.     │   │               └── abc
  9.     │   │                   └── client
  10.     │   │                       ├── configuration
  11.     │   │                       │   └── LmsiaAbcThriftClientConfiguration.java
  12.     │   │                       ├── LmsiaAbcEasyThriftClientBuilder.java
  13.     │   │                       └── LmsiaK8ServiceThriftClientBuilder.java
  14.     │   └── resources
  15.     │       └── META-INF
  16.     │           └── spring.factories
  17.     └── test

我们简单介绍一下：

  1. LmsiaAbcThriftClientConfiguration: 客户端自动配置，当激活时，自动生成lmsia-abc对应的RPC服务的客户端。引用者直接@Autowired一下，就可以使用了。
  2. LmsiaAbcEasyThriftClientBuilder: EasyThriftClient构造器，主要是自动配置需要。
  3. spring.factories: 与服务端的自动配置类似，需要在这个文件中指定自动配置的类路径，才能让Spring Boot自动扫描到自动配置。
  4. 其他K8ServiceThriftClient相关的部分，我们将在下一小节进行介绍。

LmsiaAbcEasyThriftClientBuilder文件：

  1. package com.coder4.lmsia.abc.client;
  3. import com.coder4.lmsia.abc.thrift.LmsiaAbcThrift;
  4. import com.coder4.lmsia.abc.thrift.LmsiaAbcThrift.Client;
  5. import com.coder4.lmsia.thrift.client.ThriftClient;
  6. import com.coder4.lmsia.thrift.client.builder.EasyThriftClientBuilder;
  8. /**
  9.  * @author coder4
  10.  */
  11. public class LmsiaAbcEasyThriftClientBuilder extends EasyThriftClientBuilder<Client> {
  13.     public LmsiaAbcEasyThriftClientBuilder(String host, int port) {
  14.         setThriftClass(LmsiaAbcThrift.class);
  16.         setHost(host);
  17.         setPort(port);
  18.     }
  20.     public static ThriftClient<Client> buildClient(String host, int port) {
  21.         return new LmsiaAbcEasyThriftClientBuilder(host, port).build();
  22.     }
  24. }

上述Builder完成了实际的参数填充，主要有：

  1. ThriftClient的桩代码类设置(LmsiaAbcThrift.class)
  2. 设置主机名和端口

LmsiaAbcClientConfiguration文件：

  1. package com.coder4.lmsia.abc.client.configuration;
  3. import com.coder4.lmsia.abc.client.LmsiaAbcEasyThriftClientBuilder;
  4. import com.coder4.lmsia.abc.client.LmsiaK8ServiceClientBuilder;
  5. import com.coder4.lmsia.abc.thrift.LmsiaAbcThrift;
  6. import com.coder4.lmsia.abc.thrift.LmsiaAbcThrift.Client;
  7. import com.coder4.lmsia.thrift.client.K8ServiceKey;
  8. import com.coder4.lmsia.thrift.client.ThriftClient;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11. import org.springframework.beans.factory.annotation.Qualifier;
  12. import org.springframework.beans.factory.annotation.Value;
  13. import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
  14. import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
  15. import org.springframework.context.annotation.Bean;
  16. import org.springframework.context.annotation.Configuration;
  17. import org.springframework.context.annotation.Profile;
  19. @Configuration
  20. public class LmsiaAbcThriftClientConfiguration {
  22.     private Logger LOG = LoggerFactory.getLogger(getClass());
  24.     @Bean(name = "lmsiaAbcThriftClient")
  25.     @ConditionalOnMissingBean(name = "lmsiaAbcThriftClient")
  26.     @ConditionalOnProperty(name = {"lmsiaAbcThriftServer.host", "lmsiaAbcThriftServer.port"})
  27.     public ThriftClient<Client> easyThriftClient(
  28.             @Value("${lmsiaAbcThriftServer.host}") String host,
  29.             @Value("${lmsiaAbcThriftServer.port}") int port
  30.     ) {
  31.         LOG.info("######## LmsiaAbcClientConfiguration ########");
  32.         LOG.info("easyClient host = {}, port = {}", host, port);
  33.         return LmsiaAbcEasyThriftClientBuilder.buildClient(host, port);
  34.     }
  36. }

如上所示，满足两个条件时，会自动构造LmsiaAbcEasyThriftClient：

  1. 还没有生成其他的LmsiaAbcEasyThriftClient(ConditionalOnMissingBean)
  2. 配置中指定了lmsiaAbcThriftServer.host和lmsiaAbcThriftServer.port

根据我们前面的介绍，大家应该能理解，虽然有自动配置，但上述配置是一种很糟糕的方式。试想一下，如果我们的服务依赖了5个其他RPC服务，那么岂不是要分别配置5组IP和端口？此外，这种方式也无法支持节点的负载均衡。

如何解决这个问题呢？我们将在K8ServiceThriftClient中解决。

本小节的最后，我们看一下spring.factories:

  1. org.springframework.boot.autoconfigure.EnableAutoConfiguration=com.coder4.lmsia.abc.client.configuration.LmsiaAbcThriftClientConfiguration

和之前lmsia-abc-server子工程中的文件类似，这里设置了自动配置的详细类路径，方便Spring Boot的自动扫描。

K8ServiceThriftClient

在对EasyThriftClient的介绍中，我们发现了一个问题，需要单独配置IP和端口，不支持服务自动发现。

此外，在这个客户端的实现中，默认每次都要建立新的连接。而对于后端服务而言，RPC的服务端和客户端多数都是在内网环境中，连接情况比较稳定，可以通过连接池的方式减少连接握手开销，从而提升RPC服务的性能。如果你对连接池的原理还不太熟悉，可以参考百科连接池

为此，我们本将介绍K8ServiceThriftClient，它很好的解决了上述问题。

首先，我们使用commons-pool2来构建了TTransport层的连接池。

TTransportPoolFactory:

  1. package com.coder4.lmsia.thrift.client.pool;
  3. import com.coder4.lmsia.thrift.client.K8ServiceKey;
  4. import org.apache.commons.pool2.BaseKeyedPooledObjectFactory;
  5. import org.apache.commons.pool2.PooledObject;
  6. import org.apache.commons.pool2.impl.DefaultPooledObject;
  7. import org.apache.thrift.transport.TFramedTransport;
  8. import org.apache.thrift.transport.TSocket;
  9. import org.apache.thrift.transport.TTransport;
  11. /**
  12.  * @author coder4
  13.  */
  14. public class TTransportPoolFactory extends BaseKeyedPooledObjectFactory<K8ServiceKey, TTransport> {
  16.     protected static final int THRIFT_CLIENT_DEFAULT_TIMEOUT = 5000;
  18.     protected static final int THRIFT_CLIENT_DEFAULT_MAX_FRAME_SIZE = 1024 * 1024 * 16;
  20.     @Override
  21.     public TTransport create(K8ServiceKey key) throws Exception {
  22.         if (key != null) {
  23.             String host = key.getK8ServiceHost();
  24.             int port = key.getK8ServicePort();
  25.             TSocket socket = new TSocket(host, port, THRIFT_CLIENT_DEFAULT_TIMEOUT);
  27.             TTransport transport = new TFramedTransport(
  28.                     socket, THRIFT_CLIENT_DEFAULT_MAX_FRAME_SIZE);
  30.             transport.open();
  32.             return transport;
  33.         } else {
  34.             return null;
  35.         }
  36.     }
  38.     @Override
  39.     public PooledObject<TTransport> wrap(TTransport transport) {
  40.         return new DefaultPooledObject<>(transport);
  41.     }
  43.     @Override
  44.     public void destroyObject(K8ServiceKey key, PooledObject<TTransport> obj) throws Exception {
  45.         obj.getObject().close();
  46.     }
  48.     @Override
  49.     public boolean validateObject(K8ServiceKey key, PooledObject<TTransport> obj) {
  50.         return obj.getObject().isOpen();
  51.     }
  53. }

上述代码主要完成以下功能：

  1. 连接超时配置(5秒)
  2. create, 生成新连接（TTransport），这里与之前的EasyThriftClient非常类似，不再赘述
  3. 验证连接是否有效，通过TTransport的isOpen判断。

TTransportPool:

  1. package com.coder4.lmsia.thrift.client.pool;
  3. import com.coder4.lmsia.thrift.client.K8ServiceKey;
  4. import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
  5. import org.apache.thrift.transport.TTransport;
  6. import org.slf4j.Logger;
  7. import org.slf4j.LoggerFactory;
  9. /**
  10.  * @author coder4
  11.  */
  12. public class TTransportPool extends GenericKeyedObjectPool<K8ServiceKey, TTransport> {
  14.     private Logger LOG = LoggerFactory.getLogger(getClass());
  16.     private static int MAX_CONN = 1024;
  17.     private static int MIN_IDLE_CONN = 8;
  18.     private static int MAX_IDLE_CONN = 32;
  20.     public TTransportPool(TTransportPoolFactory factory) {
  21.         super(factory);
  23.         setTimeBetweenEvictionRunsMillis(45 * 1000);
  24.         setNumTestsPerEvictionRun(5);
  25.         setMaxWaitMillis(30 * 1000);
  27.         setMaxTotal(MAX_CONN);
  28.         setMaxTotalPerKey(MAX_CONN);
  29.         setMinIdlePerKey(MIN_IDLE_CONN);
  30.         setMaxTotalPerKey(MAX_IDLE_CONN);
  32.         setTestOnCreate(true);
  33.         setTestOnBorrow(true);
  34.         setTestWhileIdle(true);
  35.     }
  37.     @Override
  38.     public TTransportPoolFactory getFactory() {
  39.         return (TTransportPoolFactory) super.getFactory();
  40.     }
  42.     public void returnBrokenObject(K8ServiceKey key, TTransport transport) {
  43.         try {
  44.             invalidateObject(key, transport);
  45.         } catch (Exception e) {
  46.             LOG.warn("return broken key " + key);
  47.             e.printStackTrace();
  48.         }
  49.     }
  50. }

上述代码主要是完成连接池的配置，比较直观：

  1. 设置最大连接数1024
  2. 设置最大空闲数32，最小空闲数8，每间隔45秒尝试更改维护连接池中的连接数量。
  3. 当每次”创建”、从池子中”借用”、”空闲”时，检查连接是否有效。

下面我们来看一下如何在K8ServiceThriftClient中使用：

  1. package com.coder4.lmsia.thrift.client;
  3. import com.coder4.lmsia.thrift.client.func.ThriftCallFunc;
  4. import com.coder4.lmsia.thrift.client.func.ThriftExecFunc;
  5. import com.coder4.lmsia.thrift.client.pool.TTransportPool;
  6. import com.coder4.lmsia.thrift.client.pool.TTransportPoolFactory;
  7. import org.apache.thrift.TServiceClient;
  8. import org.apache.thrift.transport.TTransport;
  10. public class K8ServiceThriftClient<TCLIENT extends TServiceClient>
  11.         extends AbstractThriftClient<TCLIENT> {
  13.     private K8ServiceKey k8ServiceKey;
  15.     private TTransportPool connPool;
  17.     @Override
  18.     public void init() {
  19.         super.init();
  20.         // check
  21.         if (k8ServiceKey == null) {
  22.             throw new RuntimeException("invalid k8ServiceName or k8Serviceport");
  23.         }
  24.         // init pool
  25.         connPool = new TTransportPool(new TTransportPoolFactory());
  26.     }
  28.     @Override
  29.     public <TRET> TRET call(ThriftCallFunc<TCLIENT, TRET> tcall) {
  31.         // Step 1: get TTransport
  32.         TTransport tpt = null;
  33.         K8ServiceKey key = getConnBorrowKey();
  34.         try {
  35.             tpt = connPool.borrowObject(key);
  36.         } catch (Exception e) {
  37.             throw new RuntimeException(e);
  38.         }
  40.         // Step 2: get client & call
  41.         try {
  42.             TCLIENT tcli = createClient(tpt);
  43.             TRET ret = tcall.call(tcli);
  44.             returnTransport(key, tpt);
  45.             return ret;
  46.         } catch (Exception e) {
  47.             returnBrokenTransport(key, tpt);
  48.             throw new RuntimeException(e);
  49.         }
  50.     }
  52.     @Override
  53.     public void exec(ThriftExecFunc<TCLIENT> texec) {
  54.         // Step 1: get TTransport
  55.         TTransport tpt = null;
  56.         K8ServiceKey key = getConnBorrowKey();
  57.         try {
  59.             // borrow transport
  60.             tpt = connPool.borrowObject(key);
  61.         } catch (Exception e) {
  62.             throw new RuntimeException(e);
  63.         }
  65.         // Step 2: get client & exec
  66.         try {
  67.             TCLIENT tcli = createClient(tpt);
  68.             texec.exec(tcli);
  69.             returnTransport(key, tpt);
  70.         } catch (Exception e) {
  71.             returnBrokenTransport(key, tpt);
  72.             throw new RuntimeException(e);
  73.         }
  74.     }
  76.     private K8ServiceKey getConnBorrowKey() {
  77.         return k8ServiceKey;
  78.     }
  80.     private void returnTransport(K8ServiceKey key, TTransport transport) {
  81.         connPool.returnObject(key, transport);
  82.     }
  84.     private void returnBrokenTransport(K8ServiceKey key, TTransport transport) {
  85.         connPool.returnBrokenObject(key, transport);
  86.     }
  88.     public K8ServiceKey getK8ServiceKey() {
  89.         return k8ServiceKey;
  90.     }
  92.     public void setK8ServiceKey(K8ServiceKey k8ServiceKey) {
  93.         this.k8ServiceKey = k8ServiceKey;
  94.     }
  96. }

上述大部分代码和EasyThriftClient非常接近，有差异的部分主要是与连接的”借用”、”归还”相关的：

  1. 在call和exec中，借用连接
    • getConnBorrowKey先构造一个key，包含了主机名和端口。这里的主机名是微服务的自动发现中提到的Kubernetes服务，如果你对相关原理不太熟悉，可以自行回顾对应章节。
    • 从connPool中借用一个连接（TTransport）
    • 剩余发起rpc调用的步骤就和EasyThriftClient相同了，不再赘述。
  2. 当rpc调用结束后
    • 正常结束，调用connPool.returnObject将TTransport归还到连接池中。
    • 非正常结束，调用connPool.returnBrokenTransport，让连接池销毁这个连接，以防后续借用到这个可能出错的TTransport。

类似的，我们也配套了对应的Builder：

  1. package com.coder4.lmsia.thrift.client.builder;
  3. import com.coder4.lmsia.thrift.client.EasyThriftClient;
  4. import org.apache.thrift.TServiceClient;
  6. /**
  7.  * @author coder4
  8.  */
  9. public class EasyThriftClientBuilder<TCLIENT extends TServiceClient> {
  11.     private final EasyThriftClient<TCLIENT> client = new EasyThriftClient<>();
  13.     protected EasyThriftClient<TCLIENT> build() {
  14.         client.init();
  15.         return client;
  16.     }
  18.     protected EasyThriftClientBuilder<TCLIENT> setHost(String host) {
  19.         client.setThriftServerHost(host);
  20.         return this;
  21.     }
  23.     protected EasyThriftClientBuilder<TCLIENT> setPort(int port) {
  24.         client.setThriftServerPort(port);
  25.         return this;
  26.     }
  28.     protected EasyThriftClientBuilder<TCLIENT> setThriftClass(Class<?> thriftClass) {
  29.         client.setThriftClass(thriftClass);
  30.         return this;
  31.     }
  32. }

上述Builder主要是设置所需的两个参数，Host和Port，看起来和EasyThriftClient并没有什么不同？

别着急，我们继续看一下lmsia-abc-client中的集成：

  1. package com.coder4.lmsia.abc.client;
  3. import com.coder4.lmsia.abc.thrift.LmsiaAbcThrift;
  4. import com.coder4.lmsia.abc.thrift.LmsiaAbcThrift.Client;
  5. import com.coder4.lmsia.thrift.client.K8ServiceKey;
  6. import com.coder4.lmsia.thrift.client.ThriftClient;
  7. import com.coder4.lmsia.thrift.client.builder.K8ServiceThriftClientBuilder;
  9. /**
  10.  * @author coder4
  11.  */
  12. public class LmsiaK8ServiceThriftClientBuilder extends K8ServiceThriftClientBuilder<Client> {
  14.     public LmsiaK8ServiceThriftClientBuilder(K8ServiceKey k8ServiceKey) {
  15.         setThriftClass(LmsiaAbcThrift.class);
  17.         setK8ServiceKey(k8ServiceKey);
  18.     }
  20.     public static ThriftClient<Client> buildClient(K8ServiceKey k8ServiceKey) {
  21.         return new LmsiaK8ServiceThriftClientBuilder(k8ServiceKey).build();
  22.     }
  24. }

在集成的时候，我们需要传入一个key，可以手动制定，也可以自动配置

我们看一下完整的自动配置代码，LmsiaAbcThriftClientConfiguration:

  1. public class LmsiaAbcThriftClientConfiguration {
  3.     @Bean(name = "lmsiaAbcThriftClient")
  4.     @ConditionalOnMissingBean(name = "lmsiaAbcThriftClient")
  5.     @ConditionalOnProperty(name = {"lmsiaAbcThriftServer.host", "lmsiaAbcThriftServer.port"})
  6.     public ThriftClient<Client> easyThriftClient(
  7.             @Value("${lmsiaAbcThriftServer.host}") String host,
  8.             @Value("${lmsiaAbcThriftServer.port}") int port
  9.     ) {
  10.         LOG.info("######## LmsiaAbcThriftClientConfiguration ########");
  11.         LOG.info("easyThriftClient host = {}, port = {}", host, port);
  12.         return LmsiaAbcEasyThriftClientBuilder.buildClient(host, port);
  13.     }
  15.     @Bean(name = "lmsiaAbcThriftClient")
  16.     @ConditionalOnMissingBean(name = "lmsiaAbcThriftClient")
  17.     public ThriftClient<LmsiaAbcThrift.Client> k8ServiceThriftClient() {
  18.         LOG.info("######## LmsiaAbcThriftClientConfiguration ########");
  19.         K8ServiceKey k8ServiceKey = new K8ServiceKey(K8_SERVICE_NAME, K8_SERVICE_PORT);
  20.         LOG.info("k8ServiceThriftClient key:" + k8ServiceKey);
  21.         return LmsiaK8ServiceThriftClientBuilder.buildClient(k8ServiceKey);
  22.     }
  24.     //...
  26. }

对比easyThriftClient和k8ServiceThriftClient不难发现，K8ServiceThriftClient的参数，是通过常量直接写死的。也就是我们在微服务的自动发现与负载均衡中提到的，约定好服务的命名规则。

看下常量定义：

  1. public class LmsiaAbcConstant {
  3.     // ......
  5.     public static final String PROJECT_NAME = "lmsia-abc";
  7.     public static final String K8_SERVICE_NAME = PROJECT_NAME + "-server";
  9.     public static final int K8_SERVICE_PORT = 3000;
  11.     // ......
  12. }

这样以来，一旦确定了项目名，那么Kubernetes中的服务名字也确定了。因此，k8ServiceThriftClient自动配置会被自动激活，即只要引用了lmsia-abc-client这个包，就会自动配置好一个RPC客户端，是不是非常方便？

我们来看一下具体的使用例子：

  1. import com.coder4.lmsia.thrift.client.ThriftClient;
  3. public class LmsiaAbctProxy {
  5.     @Autowired
  6.     private ThriftClient<Client> client;
  8.     public String hello() {
  9.         return client.call(cli -> cli.sayHi());
  10.     }

至此，我们已经完成了在Spring Boo中集成Thrift RPC的服务端、客户端的工作。

  • 服务端，我们通过ThriftServerConfiguration、ThriftProcessorConfiguration自动配置了Thrift RPC服务端。
  • 客户端，通过Kubernetes的服务功能，自动配置了带服务发现功能的Thrift RPC客户端K8ServiceThriftClient。该客户端同时内置了连接池，用于节省连接开销。