Thrfit 服务端请求处理流程

2024-09-05 09:48
文章标签 流程 请求 服务端 处理 thrfit

本文主要是介绍Thrfit 服务端请求处理流程，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

使用同步的非阻塞的服务端的请求处理流程

实现

IDL

  • helloworld.thrift
namespace java io.github.helloworlde.thriftstruct HelloMessage {1: required string message,
}struct HelloResponse {1: required string message,
}service HelloService {HelloResponse sayHello(1: HelloMessage request);
}

服务端实现

使用 TThreadedSelectorServer 作为服务端，支持接收连接，处理 IO 事件，执行请求由不同的线程实现；底层连接使用 ServerSocket

public class NonblockingServer {@SneakyThrowspublic static void main(String[] args) {HelloServiceImpl helloService = new HelloServiceImpl();HelloService.Processor<HelloService.Iface> helloServiceProcessor = new HelloService.Processor<>(helloService);TNonblockingServerTransport transport = new TNonblockingServerSocket(9090);// 配置参数以及处理器TThreadedSelectorServer.Args serverArgs = new TThreadedSelectorServer.Args(transport).selectorThreads(4).workerThreads(10).acceptQueueSizePerThread(20).processor(helloServiceProcessor);TServer server = new TThreadedSelectorServer(serverArgs);server.serve();}
}

请求处理流程

1. 启动 Server

TServer server = new TThreadedSelectorServer(serverArgs);
server.serve();
  • org.apache.thrift.server.AbstractNonblockingServer#serve

启动 Server，启动用于连接的线程 AcceptThread 和用于处理 IO 事件的多个线程 SelectorThread；然后开始监听 IO 事件，由线程池处理请求

public void serve() {// 启动if (!startThreads()) {return;}// 开始监听if (!startListening()) {return;}// 修改状态setServing(true);// 阻塞直到关闭waitForShutdown();setServing(false);// 停止监听器stopListening();
}
  • org.apache.thrift.server.TThreadedSelectorServer#startThreads

启动用于连接的线程 AcceptThread 和用于处理 IO 事件的多个线程 SelectorThread

protected boolean startThreads() {try {// 创建选择线程，并添加到集合中for (int i = 0; i < args.selectorThreads; ++i) {selectorThreads.add(new SelectorThread(args.acceptQueueSizePerThread));}// 创建处理连接的负载均衡， 创建处理连接的线程acceptThread = new AcceptThread((TNonblockingServerTransport) serverTransport_, createSelectorThreadLoadBalancer(selectorThreads));// 启动选择线程for (SelectorThread thread : selectorThreads) {thread.start();}// 启动连接的线程acceptThread.start();return true;} catch (IOException e) {LOGGER.error("Failed to start threads!", e);return false;}
}
  • org.apache.thrift.transport.TNonblockingServerSocket#listen

开始监听

public void listen() throws TTransportException {// Make sure not to block on acceptif (serverSocket_ != null) {try {serverSocket_.setSoTimeout(0);} catch (SocketException sx) {LOGGER.error("Socket exception while setting socket timeout", sx);}}
}

2. 处理连接事件

  • org.apache.thrift.server.TThreadedSelectorServer.AcceptThread#run

连接事件由 AcceptThread 线程独立处理；会循环监听 Selector事件，当有新的连接事件时，会建立连接

public void run() {try {if (eventHandler_ != null) {// 通知 Server 开始启动eventHandler_.preServe();}while (!stopped_) {// 选择处理连接select();}} finally {acceptSelector.close();TThreadedSelectorServer.this.stop();}
}
  • org.apache.thrift.server.TThreadedSelectorServer.AcceptThread#select

会不断从 Selector获取事件，判断如果是 accept 事件，则处理，并建立连接

private void select() {try {// 等待连接事件acceptSelector.select();// 处理接收到的事件Iterator<SelectionKey> selectedKeys = acceptSelector.selectedKeys().iterator();while (!stopped_ && selectedKeys.hasNext()) {SelectionKey key = selectedKeys.next();selectedKeys.remove();if (!key.isValid()) {continue;}if (key.isAcceptable()) {// 建立连接handleAccept();} else {LOGGER.warn("Unexpected state in select! " + key.interestOps());}}} catch (IOException e) {LOGGER.warn("Got an IOException while selecting!", e);}
}
  • org.apache.thrift.server.TThreadedSelectorServer.AcceptThread#handleAccept

会通过底层的 ServerSocketChannel 建立连接，然后将这个连接添加到 SelectorThread 的队列中，由 SelectorThread处理 IO 事件

private void handleAccept() {// 建立连接final TNonblockingTransport client = doAccept();if (client != null) {// 将连接传递给选择线程final SelectorThread targetThread = threadChooser.nextThread();// 如果策略是尽快建立连接，则添加到处理的队列中if (args.acceptPolicy == Args.AcceptPolicy.FAST_ACCEPT || invoker == null) {doAddAccept(targetThread, client);} else {try {// 如果是 FAIR_ACCEPT，则提交异步任务进行添加invoker.submit(new Runnable() {public void run() {doAddAccept(targetThread, client);}});} catch (RejectedExecutionException rx) {LOGGER.warn("ExecutorService rejected accept registration!", rx);// close immediatelyclient.close();}}}
}
  • org.apache.thrift.transport.TNonblockingServerSocket#acceptImpl

建立连接，返回新的 TNonblockingSocket

protected TNonblockingSocket acceptImpl() throws TTransportException {if (serverSocket_ == null) {throw new TTransportException(TTransportException.NOT_OPEN, "No underlying server socket.");}try {// 接受连接SocketChannel socketChannel = serverSocketChannel.accept();if (socketChannel == null) {return null;}// 使用 Channel 构建 SocketTNonblockingSocket tsocket = new TNonblockingSocket(socketChannel);tsocket.setTimeout(clientTimeout_);return tsocket;} catch (IOException iox) {throw new TTransportException(iox);}
}
  • org.apache.thrift.server.TThreadedSelectorServer.SelectorThread#addAcceptedConnection

将连接添加到 SelectorThread的队列中，由 SelectorThread处理 IO 事件

public boolean addAcceptedConnection(TNonblockingTransport accepted) {try {// 放入队列中acceptedQueue.put(accepted);} catch (InterruptedException e) {LOGGER.warn("Interrupted while adding accepted connection!", e);return false;}selector.wakeup();return true;
}

3. 处理 IO 事件

IO 事件由 SelectorThread 处理

  • org.apache.thrift.server.TThreadedSelectorServer.SelectorThread#run

轮询读取事件，如果是 IO 事件，则分别处理；如果是新的连接，则注册 Selector

public void run() {try {while (!stopped_) {// 选择读取或写入事件select();// 处理新的连接processAcceptedConnections();// 改变需要改变的状态processInterestChanges();}// 如果停止了，则清理选择for (SelectionKey selectionKey : selector.keys()) {cleanupSelectionKey(selectionKey);}} catch (Throwable t) {LOGGER.error("run() on SelectorThread exiting due to uncaught error", t);} finally {// 关闭selector.close();TThreadedSelectorServer.this.stop();}
}
  • org.apache.thrift.server.TThreadedSelectorServer.SelectorThread#select

处理 IO 事件，根据事件类型分别处理读取或者写入

private void select() {try {// 获取事件doSelect();Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();while (!stopped_ && selectedKeys.hasNext()) {SelectionKey key = selectedKeys.next();selectedKeys.remove();// 如果无效则跳过if (!key.isValid()) {cleanupSelectionKey(key);continue;}if (key.isReadable()) {// 如果是读取则处理读取事件handleRead(key);} else if (key.isWritable()) {// 如果是写入则处理写入handleWrite(key);} else {LOGGER.warn("Unexpected state in select! " + key.interestOps());}}} catch (IOException e) {LOGGER.warn("Got an IOException while selecting!", e);}
}
处理读取事件
  • org.apache.thrift.server.AbstractNonblockingServer.AbstractSelectThread#handleRead

在处理读取事件时，会读取整个帧，当完全读取时，会调用 requestInvoke 方法，通过线程池处理请求

protected void handleRead(SelectionKey key) {// 获取帧FrameBuffer buffer = (FrameBuffer) key.attachment();// 如果没有可读取的，则清理if (!buffer.read()) {cleanupSelectionKey(key);return;}// if the buffer's frame read is complete, invoke the method.// 如果 buffer 完全读取，则执行处理，如果失败则清理if (buffer.isFrameFullyRead()) {if (!requestInvoke(buffer)) {cleanupSelectionKey(key);}}
}
  • org.apache.thrift.server.TThreadedSelectorServer#requestInvoke

处理调用，会将帧封装为 Runnable 任务，提交给线程池执行

protected boolean requestInvoke(FrameBuffer frameBuffer) {// 封装为 RunnableRunnable invocation = getRunnable(frameBuffer);if (invoker != null) {try {// 执行处理invoker.execute(invocation);return true;} catch (RejectedExecutionException rx) {LOGGER.warn("ExecutorService rejected execution!", rx);return false;}} else {// Invoke on the caller's thread// 如果没有线程池，由当前线程直接处理invocation.run();return true;}
}
处理写入事件
  • org.apache.thrift.server.AbstractNonblockingServer.AbstractSelectThread#handleWrite

处理写入事件，调用 FrameBuffer 的写入方法进行处理

protected void handleWrite(SelectionKey key) {FrameBuffer buffer = (FrameBuffer) key.attachment();if (!buffer.write()) {cleanupSelectionKey(key);}
}
  • org.apache.thrift.server.AbstractNonblockingServer.FrameBuffer#write

由 Transport 执行写入，最终由 SocketChannel 执行，将响应内容发送给客户端

public boolean write() {if (state_ == FrameBufferState.WRITING) {// 写入if (trans_.write(buffer_) < 0) {return false;}// 如果没有待写入的，则切换到读取if (buffer_.remaining() == 0) {prepareRead();}return true;}return false;
}

4. 执行请求

  • org.apache.thrift.server.AbstractNonblockingServer.FrameBuffer#invoke

在处理读取事件时，会将 FrameBuffer 包装为 Runnable，提交给线程池执行；最终由 FrameBuffer处理
会获取 Processor，然后调用 process 方法进行处理

public void invoke() {frameTrans_.reset(buffer_.array());response_.reset();try {// 如果有事件处理器，则触发if (eventHandler_ != null) {eventHandler_.processContext(context_, inTrans_, outTrans_);}// 获取处理器，调用处理方法processorFactory_.getProcessor(inTrans_).process(inProt_, outProt_);responseReady();return;} catch (TException te) {}// This will only be reached when there is a throwable.state_ = FrameBufferState.AWAITING_CLOSE;requestSelectInterestChange();
}
  • org.apache.thrift.TBaseProcessor#process

在处理时，根据方法名获取具体的处理函数，然后调用响应的处理方法进行处理

public void process(TProtocol in, TProtocol out) throws TException {TMessage msg = in.readMessageBegin();ProcessFunction fn = processMap.get(msg.name);if (fn == null) {TProtocolUtil.skip(in, TType.STRUCT);in.readMessageEnd();TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Invalid method name: '"+msg.name+"'");out.writeMessageBegin(new TMessage(msg.name, TMessageType.EXCEPTION, msg.seqid));x.write(out);out.writeMessageEnd();out.getTransport().flush();} else {fn.process(msg.seqid, in, out, iface);}
}
  • org.apache.thrift.ProcessFunction#process

读取请求信息，反序列化为对象，然后调用 getResult 方法执行实现逻辑，获取响应；如果不是 oneway 的请求，则将相应结果写入流中，发送给客户端

public final void process(int seqid,TProtocol iprot,TProtocol oprot,I iface) throws TException {// 获取空参数实例T args = getEmptyArgsInstance();// 读取args.read(iprot);iprot.readMessageEnd();TSerializable result = null;byte msgType = TMessageType.REPLY;// 获取结果result = getResult(iface, args);// 如果不是 oneway 的，则写入响应结果if (!isOneway()) {oprot.writeMessageBegin(new TMessage(getMethodName(), msgType, seqid));result.write(oprot);oprot.writeMessageEnd();oprot.getTransport().flush();}
}
  • io.github.helloworlde.thrift.HelloService.Processor.sayHello#getResult

由生成的代码处理，会先构建一个响应结构体，然后调用相应的方法进行处理，返回结果

public sayHello_result getResult(I iface, sayHello_args args) throws org.apache.thrift.TException {sayHello_result result = new sayHello_result();result.success = iface.sayHello(args.request);return result;
}
  • io.github.helloworlde.thrift.HelloServiceImpl#sayHello

具体的逻辑处理，返回响应

public HelloResponse sayHello(HelloMessage request) throws TException {String message = request.getMessage();HelloResponse response = new HelloResponse();response.setMessage("Hello " + message);return response;
}

4. 写入响应

  • org.apache.thrift.ProcessFunction#process

在处理完请求之后，会判断是否是 oneway 请求，如果不是，则会执行写入响应

if(!isOneway()) {oprot.writeMessageBegin(new TMessage(getMethodName(), msgType, seqid));result.write(oprot);oprot.writeMessageEnd();oprot.getTransport().flush();
}
  • org.apache.thrift.protocol.TBinaryProtocol#writeMessageBegin

写入响应时，会先写入响应头；会将版本信息，消息类型，方法的名称和请求 ID 一起写入

public void writeMessageBegin(TMessage message) throws TException {if (strictWrite_) {int version = VERSION_1 | message.type;writeI32(version);writeString(message.name);writeI32(message.seqid);} else {writeString(message.name);writeByte(message.type);writeI32(message.seqid);}
}
  • io.github.helloworlde.thrift.HelloResponse.HelloResponseStandardScheme#write

随后写入响应内容，将对象序列化为字节

public void write(org.apache.thrift.protocol.TProtocol oprot, HelloResponse struct) throws org.apache.thrift.TException {struct.validate();oprot.writeStructBegin(STRUCT_DESC);if (struct.message != null) {oprot.writeFieldBegin(MESSAGE_FIELD_DESC);oprot.writeString(struct.message);oprot.writeFieldEnd();}oprot.writeFieldStop();oprot.writeStructEnd();
}

然后会写入响应结尾符，由 SelectorThread 处理写入事件，最终将请求发送给客户端

参考文档

  • helloworlde/thrift-java-sample

这篇关于Thrfit 服务端请求处理流程的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1138622。

相关文章

Go语言使用Buffer实现高性能处理字节和字符

Go语言使用Buffer实现高性能处理字节和字符

《Go语言使用Buffer实现高性能处理字节和字符》在Go中,bytes.Buffer是一个非常高效的类型,用于处理字节数据的读写操作,本文将详细介绍一下如何使用Buffer实现高性能处理字节和... 目录1. bytes.Buffer 的基本用法1.1. 创建和初始化 Buffer1.2. 使用 Writ
阅读更多...
Java后端接口中提取请求头中的Cookie和Token的方法

Java后端接口中提取请求头中的Cookie和Token的方法

《Java后端接口中提取请求头中的Cookie和Token的方法》在现代Web开发中,HTTP请求头（Header）是客户端与服务器之间传递信息的重要方式之一,本文将详细介绍如何在Java后端（以Sp... 目录引言1. 背景1.1 什么是 HTTP 请求头？1.2 为什么需要提取请求头？2. 使用 Spr
阅读更多...
Python视频处理库VidGear使用小结

Python视频处理库VidGear使用小结

《Python视频处理库VidGear使用小结》VidGear是一个高性能的Python视频处理库,本文主要介绍了Python视频处理库VidGear使用小结,文中通过示例代码介绍的非常详细,对大家的... 目录一、VidGear的安装二、VidGear的主要功能三、VidGear的使用示例四、VidGea
阅读更多...
Python结合requests和Cheerio处理网页内容的操作步骤

Python结合requests和Cheerio处理网页内容的操作步骤

《Python结合requests和Cheerio处理网页内容的操作步骤》Python因其简洁明了的语法和强大的库支持,成为了编写爬虫程序的首选语言之一,requests库是Python中用于发送HT... 目录一、前言二、环境搭建三、requests库的基本使用四、Cheerio库的基本使用五、结合req
阅读更多...
使用Python处理CSV和Excel文件的操作方法

使用Python处理CSV和Excel文件的操作方法

《使用Python处理CSV和Excel文件的操作方法》在数据分析、自动化和日常开发中,CSV和Excel文件是非常常见的数据存储格式,ython提供了强大的工具来读取、编辑和保存这两种文件,满足从基... 目录1. CSV 文件概述和处理方法1.1 CSV 文件格式的基本介绍1.2 使用 python 内
阅读更多...
Python实现NLP的完整流程介绍

Python实现NLP的完整流程介绍

《Python实现NLP的完整流程介绍》这篇文章主要为大家详细介绍了Python实现NLP的完整流程,文中的示例代码讲解详细,具有一定的借鉴价值,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1. 编程安装和导入必要的库2. 文本数据准备3. 文本预处理3.1 小写化3.2 分词（Tokenizatio
阅读更多...
如何使用celery进行异步处理和定时任务(django)

如何使用celery进行异步处理和定时任务(django)

《如何使用celery进行异步处理和定时任务(django)》文章介绍了Celery的基本概念、安装方法、如何使用Celery进行异步任务处理以及如何设置定时任务,通过Celery,可以在Web应用中... 目录一、celery的作用二、安装celery三、使用celery 异步执行任务四、使用celery
阅读更多...
SpringBoot操作spark处理hdfs文件的操作方法

SpringBoot操作spark处理hdfs文件的操作方法

《SpringBoot操作spark处理hdfs文件的操作方法》本文介绍了如何使用SpringBoot操作Spark处理HDFS文件,包括导入依赖、配置Spark信息、编写Controller和Ser... 目录SpringBoot操作spark处理hdfs文件1、导入依赖2、配置spark信息3、cont
阅读更多...
SpringBoot使用minio进行文件管理的流程步骤

SpringBoot使用minio进行文件管理的流程步骤

《SpringBoot使用minio进行文件管理的流程步骤》MinIO是一个高性能的对象存储系统,兼容AmazonS3API,该软件设计用于处理非结构化数据,如图片、视频、日志文件以及备份数据等,本文... 目录一、拉取minio镜像二、创建配置文件和上传文件的目录三、启动容器四、浏览器登录 minio五、
阅读更多...
SpringBoot实现websocket服务端及客户端的详细过程

SpringBoot实现websocket服务端及客户端的详细过程

《SpringBoot实现websocket服务端及客户端的详细过程》文章介绍了WebSocket通信过程、服务端和客户端的实现,以及可能遇到的问题及解决方案,感兴趣的朋友一起看看吧... 目录一、WebSocket通信过程二、服务端实现1.pom文件添加依赖2.启用Springboot对WebSocket
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2