重點來了,本文全面闡述一下我們的RPC是怎么實現並如何使用的,跟Kubernetes和Openstack怎么結合。
在選型一文中說到我們選定的RPC框架是Apache Thrift,它的用法是在Main方法中重啟服務,在Client端連接服務去調用,
而我的想法是要跟Dubblo、HSF的用法一樣,因為很多人都熟習這兩個框架的用法,特別是我們好幾個項目都是基於EDAS開發的,而且世面上用Dubbo的公司也很多。
順便再說一下我們對於RPC的幾點要求:
- 1,兼容Dubbo和HSF的使用方法,支持版本和服務分組,支持項目隔離
- 2,客戶端重試機制,可以配置次數和間隔時間
- 3,客戶端線程池
- 4,服務端可以平滑無縫升級而不影響客戶端的使用
兼容Dubbo就必然要使用Spring框架,那我們就直接上Spring Boot好了,號稱Spring Boot為微服務開發而生,一步到位,將Thrift跟Spring Boot整合。
版本和服務分組可以通過Kubernetes的Service的Label來實現,我們客戶端在查找服務的時候通過這兩個標簽再加上接口類的Label來定位Service的Cluster IP,這里不直接使用Service名稱來調用服務的原因是通過Label查詢Servcie更加靈活一些,Service的名稱不受限制,隨時可以啟動一個帶有相同Label的新Service來替換舊的Service.
項目隔離可以用Kubernetes的namespace來實現,一個namespace是一個項目,當然項目之間也可以互相調用,默認情況下是整個Kubernetes集群的服務都是可以被調用到的如果在沒有指定namespace的情況下。
客戶端重試機制用代理Thrift連接的方式來實現,在連接或接口方法調用異常時發起重新連接,參考:https://liveramp.com/engineering/reconnecting-thrift-client/
客戶端連接池是由於在WEB項目中每次用戶發起請求是在一個獨立的線程中,而Thrift的Client Socket連接不是線程安全的,因此要為每個用戶准備一個Socket連接,有點像數據庫的連接池,這個可以用apache的commons pool2來實現,這個有很多網友的文章可以參考,本文就不在贅述了。
服務端平滑升級可以使用Kubernetes的Kubectl rolling-update來實現,它的機制是先創建一個RC,然后新建一個新版本Pod,停掉一個舊版本Pod,逐步來完成整個RC的更新,最后刪除舊的RC,把新的RC名稱改為舊的RC名稱,升級過程如下圖:
這里會有一個問題,因為有一個時間段會新舊RC共存,由於Service是基於RC的Label建立的,那過來的請求是不是會得到兩種結果?
如果是的話要防止這樣的情況發生就要像上面說的,將整個Service替換,先啟動一個新的Service跟舊的Service有相同Label,然后刪除舊的Service以及RC,在發生服務請求的時候Thrift Client在找不到舊的服務的時候根據Label重新查找Service就會切換到新的Service上。
下面展示一下大概的實現及使用方法,假設你熟習Kubernetes或者簡單了解,熟習Docker。
服務端
配置
<bean class="io.masir.testcloud.thrift.HelloImpl" id="helloImpl"/> <bean class="io.masir.testcloud.thrift.ThriftSpringProviderBean" init-method="init" id="providerBean"> <property name="serviceInterface" value="io.masir.testcloud.thrift.HelloService"/> <property name="serviceVersion" value="1.0.0"/> <property name="serviceGroup" value="testServiceGroup"/> <property name="target" ref="helloImpl"/> </bean>
ThriftSpringProviderBean核心代碼 這是Thrift和Spring整合的核心代碼,可以借鑒其它網友的Thrift Spring實例。
public class ThriftSpringProviderBean extends Thread { private int port = 10809; private String serviceInterface; private String serviceVersion; private String serviceGroup; private Object target; public void run() { try { TServerSocket serverTransport = new TServerSocket(getPort()); Class Processor = Class.forName(getServiceInterface() + "$Processor"); Class Iface = Class.forName(getServiceInterface() + "$Iface"); Constructor con = Processor.getConstructor(Iface); TProcessor processor = (TProcessor) con.newInstance(getTarget()); TBinaryProtocol.Factory protFactory = new TBinaryProtocol.Factory(true, true); TThreadPoolServer.Args args = new TThreadPoolServer.Args(serverTransport); args.protocolFactory(protFactory); args.processor(processor); TServer