-
Thrift -- 跨语言RPC 框架
Thrift – 跨语言RPC 框架
1.简介
The Apache Thrift software framework, for scalable cross-language services development, combines a software stack with a code generation engine to build services that work efficiently and seamlessly between C++, Java, Python, PHP, Ruby, Erlang, Perl, Haskell, C#, Cocoa, JavaScript, Node.js, Smalltalk, OCaml and Delphi and other languages.
用于跨语言的服务软件开发，结合代码生成引擎，构建 C++, Java, Python, PHP, Ruby… 生成多语言的RPC 调用代码；
借助IDL 语言(可以理解这个框架的模板语言，后面在详细说) 生成不同语言的调用代码
2.下载安装
我电脑是windows10 就直接下载 win 版本 下载地址是 ：
https://thrift.apache.org/download安装环境变量 ，
我这里复制了一个副本把版本信息去除了
C:\Users\sff>thrift -version Thrift version 0.16.0- 1
- 2
安装完成
3.Thrift 架构
-
交互架构 CS架构，客户端请求服务端；这两边的代码都可以通过Thrift 来生成
-
技术架构就是 （区别于ISO 七层或者五层网络哈）
- 传输层 定义具体的网络连接传输协议 使用TCP /IP UDP
- 协议层 RPC 交互过程中数据的序列化和反序列化格式 有JSON XML 二进制数据等
- 处理层 Thrift gen 生成的各个语言的接口，然后我们服务端是实现这个接口来提供客户端的调用服务
- 服务层 接口上面 然后使用不同的IO 模型来处理数据
4.Thrift 开发优势
-
多语言 跨语言，不过要学习IDL(接口定义语言) . 这个还是相对比较简单。。
-
只要维护IDL 文件，代码生成引擎 会生产各个语言的服务端代码和客户端代码，省去定义各语言接口编写，网络传输，服务器IO模型类型(少量代码)的时间，可以专注于实现自己的业务代码；
注： 学习地址
IDL语法：
https://thrift.apache.org/docs/idl类型定义：
https://thrift.apache.org/docs/types示例：
//各个语言对应的包 namespace java com.sff.study namespace js com.sff.study namespace go com.sff.study struct Demo{ 1: i32 id 2: string desc } service DemoService { Demo getDemo(1:i32 id) bool setDemo(1:Demo demo) }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
## 生成 java go js cpp D:\Resource\FrameMiddleware\Thrift\template>thrift -o D:\Resource\FrameMiddleware\Thrift\ --gen java demo.thrift D:\Resource\FrameMiddleware\Thrift\template>thrift -o D:\Resource\FrameMiddleware\Thrift\ --gen go demo.thrift D:\Resource\FrameMiddleware\Thrift\template>thrift -o D:\Resource\FrameMiddleware\Thrift\ --gen js demo.thrift D:\Resource\FrameMiddleware\Thrift\template>thrift -o D:\Resource\FrameMiddleware\Thrift\ --gen cpp demo.thrift- 1
- 2
- 3
- 4
- 5
5.协议层协议
父类 TProtocol
- TCompacProtocol :压缩二进制编码传输
- TbinaryProtocol:二进制编码传输
- TJSONProtocol JSON 协议传输 带类型信息
- TSimpleJSONProtocol JSON 只只有数据 需要自己有反序列化工具
6.传输层协议
- TServerSocket ： Socket 常见的BIO
- TNonblockingServerTransport: 非阻塞IO Nio ；
- TFramedTransport
7.快速开始
基于前面代码生成的 java 版本的 代码
7.1 创建两个项目 导入生成的代码
7.2 添加依赖
<dependencies> <dependency> <groupId>org.apache.thriftgroupId> <artifactId>libthriftartifactId> <version>0.16.0version> dependency> <dependency> <groupId>org.slf4jgroupId> <artifactId>slf4j-simpleartifactId> <version>1.7.25version> dependency> dependencies>- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
7.3 服务端实现接口
public class DemoServiceImpl implements DemoService.Iface { @Override public Demo getDemo(int id) throws TException { Demo demo = new Demo(); int nextInt = new Random().nextInt(100); demo.setId(nextInt); demo.setDesc("我是demo" + nextInt); return null; } @Override public boolean setDemo(Demo demo) throws TException { System.out.println("保存了~~~~~：" + demo); return false; } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
主启动类 简单模式
package com.sff.study; import org.apache.thrift.protocol.TBinaryProtocol; import org.apache.thrift.server.TServer; import org.apache.thrift.server.TSimpleServer; import org.apache.thrift.transport.TServerSocket; import org.apache.thrift.transport.TServerTransport; public class ServerMain { public static void main(String[] args) { try { TServerTransport serverTransport = new TServerSocket(9090); TServer.Args args1 = new TServer.Args(serverTransport); // 业务处理器 DemoService.Processor processor = new DemoService.Processor(new DemoServiceImpl()); args1.processor(processor); // 指定序列化工厂 TBinaryProtocol.Factory factory = new TBinaryProtocol.Factory(); args1.protocolFactory(factory); // 绑定参数 TSimpleServer server = new TSimpleServer(args1); // Use this for a multithreaded server // TServer server = new TThreadPoolServer(new // TServer server = new TThreadPoolServer(new // TThreadPoolServer.Args(serverTransport).processor(processor)); System.out.println("Starting the simple server..."); server.serve(); } catch (Exception e) { e.printStackTrace(); } } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
7.4 客户端
测试连接调用
package com.sff.study; import org.apache.thrift.protocol.TBinaryProtocol; import org.apache.thrift.protocol.TProtocol; import org.apache.thrift.transport.TSocket; public class ClientMain { public static void main(String[] args) throws Exception { // 传输层IO TSocket socket = new TSocket("127.0.0.1", 9090); // 协议层 序列化 TProtocol protocol = new TBinaryProtocol(socket); DemoService.Client client = new DemoService.Client(protocol); // 打开 scoket socket.open(); Demo demo = client.getDemo(11); System.out.println(demo); // int nextInt = new Random().nextInt(100); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
7.5 调试
第一次调试
先开启服务端，后开启客户端 ，客户端报错
Connected to the target VM, address: '127.0.0.1:54713', transport: 'socket' Exception in thread "main" org.apache.thrift.TApplicationException: getDemo failed: unknown result at com.sff.study.DemoService$Client.recv_getDemo(DemoService.java:71) at com.sff.study.DemoService$Client.getDemo(DemoService.java:56) at com.sff.study.ClientMain.main(ClientMain.java:20) Disconnected from the target VM, address: '127.0.0.1:54713', transport: 'socket'- 1
- 2
- 3
- 4
- 5
- 6
发现是 DemoServiceImpl##getDemo() 返回的是null 导致 修改后可以调试通过
Connected to the target VM, address: '127.0.0.1:54809', transport: 'socket' Demo(id:11, desc:我是demo11) Disconnected from the target VM, address: '127.0.0.1:54809', transport: 'socket'- 1
- 2
- 3
8.服务端服务类型
TServer 是所有服务的父类
// 主要方法 还是子类去实现的 开启服务 serve public abstract void serve(); // 主要方法 还是子类去实现的 关闭服务 stop public void stop() {}- 1
- 2
- 3
- 4
8.1 TSimpleServer
public class TSimpleServer extends TServer { private static final Logger LOGGER = LoggerFactory.getLogger(TSimpleServer.class.getName()); public TSimpleServer(AbstractServerArgs args) { super(args); } //核心方法 开启服务 public void serve() { try { //开启监听端口 this.serverTransport_.listen(); } catch (TTransportException var9) { LOGGER.error("Error occurred during listening.", var9); return; } if (this.eventHandler_ != null) { this.eventHandler_.preServe(); } this.setServing(true); // 关闭服务置位 true 停止循环 while(!this.stopped_) { ..... try { // 阻塞方法 一直监听请求 client = this.serverTransport_.accept(); if (client != null) { ........ while(true) { if (this.eventHandler_ != null) { this.eventHandler_.processContext(connectionContext, inputTransport, outputTransport); } //获取我们自己的处理器 处理业务逻辑 processor.process(inputProtocol, outputProtocol); } } } catch (TTransportException var10) { ...... } } this.setServing(false); } //核心方法 关闭服务 public void stop() { this.stopped_ = true; this.serverTransport_.interrupt(); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
执行流程如下
缺点： 单线程 阻塞；多客户端连接会阻塞，直到前面连接的释放才可以处理下一个
8.2 TThreadPoolServer
public class TThreadPoolServer extends TServer { private static final Logger LOGGER = LoggerFactory.getLogger(TThreadPoolServer.class); private ExecutorService executorService_; private final TimeUnit stopTimeoutUnit; private final long stopTimeoutVal; public TThreadPoolServer(TThreadPoolServer.Args args) { super(args); this.stopTimeoutUnit = args.stopTimeoutUnit; this.stopTimeoutVal = (long)args.stopTimeoutVal; // 初始化的时候使用 线程池 this.executorService_ = args.executorService != null ? args.executorService : createDefaultExecutorService(args); } // 线程池参数 private static ExecutorService createDefaultExecutorService(TThreadPoolServer.Args args) { // args.maxWorkerThreads 参数为最大线程数 return new ThreadPoolExecutor(args.minWorkerThreads, args.maxWorkerThreads, 60L, TimeUnit.SECONDS, new SynchronousQueue(), new ThreadFactory() { final AtomicLong count = new AtomicLong(); public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setDaemon(true); thread.setName(String.format("TThreadPoolServer WorkerProcess-%d", this.count.getAndIncrement())); return thread; } }); } //开启监听 protected boolean preServe() { ...... this.serverTransport_.listen(); ...... return true; } public void serve() { //开启监听 if (this.preServe()) { // 真正执行的方法 this.execute(); this.executorService_.shutdownNow(); if (!this.waitForShutdown()) { LOGGER.error("Shutdown is not done after " + this.stopTimeoutVal + this.stopTimeoutUnit); } this.setServing(false); } } protected void execute() { while(!this.stopped_) { try { //和8.1 的那个一样 会一致处于阻塞 直到新的socket 到来 TTransport client = this.serverTransport_.accept(); try { //来一个连接就交给线程池处理 this.executorService_.execute(new TThreadPoolServer.WorkerProcess(client)); } catch (RejectedExecutionException var3) { if (!this.stopped_) { LOGGER.warn("ThreadPool is saturated with incoming requests. Closing latest connection."); } client.close(); } } catch (TTransportException var4) { if (!this.stopped_) { LOGGER.warn("Transport error occurred during acceptance of message", var4); } } } } public static class Args extends AbstractServerArgs<TThreadPoolServer.Args> { public int minWorkerThreads = 5; public int maxWorkerThreads = 2147483647; public ExecutorService executorService; public int stopTimeoutVal = 60; public TimeUnit stopTimeoutUnit; ..... } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
执行流程图
优点：每个Socket 都会去线程池拿一个线程去处理，适合连接数量不多的这种场景
缺点：很多客户端同时连接的时候，当客户端比较多的时候，就需要服务端有对应数量的线程来处理，而且代码中设置的上限还比较大；线程资源比较重要，比较消耗服务器性能；可能会内存溢出；
8.3 AbstractNonblockingServer
AbstractNonblockingServer 父类 主要是约定主流程 ,实现类只能使用TFrameTransport
public void serve() { if (this.startThreads()) { if (this.startListening()) { this.setServing(true); this.waitForShutdown(); this.setServing(false); this.stopListening(); } } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
8.4 TNonblockingServer React 模型
TNonblockingServer extends AbstractNonblockingServer
看看执行流程
server 父类 ==> this.startThreads() protected boolean startThreads() { .... // TNonblockingServer.SelectAcceptThread this.selectAcceptThread_ = new TNonblockingServer.SelectAcceptThread((TNonblockingServerTransport)this.serverTransport_); this.selectAcceptThread_.start(); ... } ======>TNonblockingServer.SelectAcceptThread protected class SelectAcceptThread extends AbstractSelectThread { ...... public void run() { try { .... while(!TNonblockingServer.this.stopped_) { // 调用 this.select(); this.processInterestChanges(); } ..... } // 调用 private void select() { .... this.handleRead(key); .... } } =====》 父类 this.handleRead(key); protected void handleRead(SelectionKey key) { ///this.requestInvoke(buffer) 调用 if (buffer.isFrameFullyRead() && !AbstractNonblockingServer.this.requestInvoke(buffer)) { } //this.requestInvoke(buffer) 调用 protected boolean requestInvoke(FrameBuffer frameBuffer) { //调用 frameBuffer.invoke(); return true; } //frameBuffer.invoke(); 调用 public void invoke() { .... AbstractNonblockingServer.this.processorFactory_.getProcessor(this.inTrans_).process(this.inProt_, this.outProt_); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
流程图
缺点：单线程处理，多客户单耗时操作会使得客户端阻塞；
8.5 TThreadedSelectorServer 单React
TThreadedSelectorServer extends AbstractNonblockingServer
TThreadedSelectorServer.AcceptThread 负责连接
TNonblockingServer 的多线程 (选择器线程，业务逻辑处理线程 )版本，一个Accpet 代码就不展示了
流程图如下
这个优点 缺点 ：其实不是很明显了 处理连接socket 状态和业务逻辑都给线程池去处理了
8.6 THsHaServer 单React
THsHaServer extends TNonblockingServer
TNonblockingServer 的多线程 (业务逻辑处理线程 )版本 代码就不展示了
-
相关阅读:
01Spring的Ioc思想和依赖注入手段(DI）
Eclipse里 ant 打包 BUILD FAILED 总结
中断机制-interrupt和isInterrupted源码分析、中断协商案例
【项目】若依框架如何实现批量导入，并解析出表中内容返回给前端？ - poi依赖
力扣 -- 10. 正则表达式匹配
Pake 轻松构建轻量级多端桌面应用
将 BA、DevOps 和 QA 与云测试同步
大数据学习初级入门教程（十三） —— Redis 3.x 伪分布式集群（哨兵模式）的安装、配置、启动和测试
RocksDB基本架构与原理详解
Docker极简入门：使用Docker-Compose 搭建redis集群
- 原文地址：https://blog.csdn.net/weixin_44244088/article/details/126220176
