C#使用Thrift作为RPC框架实战(四)之TSocket

  • A+
所属分类:.NET技术
摘要

  在前几个小节中我们讲了Thrift框架的基本概念以及重要的名称空间,接下来的几个小节,我们将站在实战的角度来深入讲解一些Thrift的重要类型。本小节我先要讲一下Thrift框架支持TCP通信的类,客户端TSocket,服务器端TServerSocket。


前言

  在前几个小节中我们讲了Thrift框架的基本概念以及重要的名称空间,接下来的几个小节,我们将站在实战的角度来深入讲解一些Thrift的重要类型。本小节我先要讲一下Thrift框架支持TCP通信的类,客户端TSocket,服务器端TServerSocket。

客户端TSocket

  Tsocket作为Thrift框架实现TCP通信的底层类型(上面两层分别为Protocol层和Client层),我们首先来看一下TSocket的构造函数:

  public TSocket(TcpClient client);   public TSocket(string host, int port);   public TSocket(string host, int port, int timeout); 

Tsocket有三个构造函数:

  • + 第一个构造需要我们自己维护一个TcpClient,对于熟悉.net Socket通信的同学来说,这个很简单,就不在此赘述了
  • + 第二个和第三个构造函数相似,唯一的不同是在是否有设置超时时间TimeOut这个参数上

构造函数 有无timeout参数的问题

  我们重点来讲一些后两个构造函数上,其实后两个构造函数在内部实现上并无差别,看一下源码我们就清晰了:

public TSocket(string host, int port) : this(host, port, 0) 		{ 		}  public TSocket(string host, int port, int timeout) 		{ 			this.host = host; 			this.port = port; 			this.timeout = timeout; 			this.InitSocket(); 		} 

在内部实现上如果我们不设置timeout参数,它会被设置为0,然后还是调用三个参数的构造构造函数的。是不是我们调用这个两个构造函数去实例化这个类没有一点差别呢?答案是 否定的。他们在调用**Open**方法时走了不同的代码分支:

 

if (this.timeout == 0)  { 	this.client.Connect(this.host, this.port);  }  else  { 				TSocket.ConnectHelper connectHelper = new TSocket.ConnectHelper(this.client); 				IAsyncResult asyncResult =  this.client.BeginConnect(this.host, this.port,  new AsyncCallback(TSocket.ConnectCallback), connectHelper); 				if (!asyncResult.AsyncWaitHandle.WaitOne(this.timeout) || !this.client.Connected) 				{                ...... 

 

  

 

这里先说明一点Timeout参数被赋值给TcpLient类型的SendTimeout和ReceiveTimeout参数上:

public int Timeout 		{ 			set 			{ 				TcpClient arg_1E_0 = this.client; 				TcpClient arg_18_0 = this.client; 				this.timeout = value; 				arg_18_0.SendTimeout = value; 				arg_1E_0.ReceiveTimeout = value; 			} 		} 

如果你没有设置timeout参数,需要记住一点,host参数你要传IPv6对应的字符串,如果你传了ipv4对应的字符串,你将收到莫名其妙的三种类型的错误:

  1. 调用sendto方法前没有设置远程终结点
  2. 远程主机关闭了现有链接
  3. 内部错误

thrift框架在错误提示上有点不友好,给出的错误提示没有一点用处。这个错误解决的方法我们可以从源码上找到问题所在,请看一下代码:

internal static TcpClient CreateTcpClient() 		{ 			return new TcpClient(AddressFamily.InterNetworkV6) 			{ 				Client =  				{ 					DualMode = true 				} 			}; 		} 

  上面代码我们在**InitSocket**方法中找到创建TcpClient类型的方法,我们一下代码已经知道了原因,因为将TcpClient类型定位到了InterNetworkV6的类型,如果我们创建时传了ipv4的地址,就会出上述问题。如果,我们设置了timeout参数,既是我们传了ipv4的地址也不会有问题,这个可能和connect,beginconnect两种链接方式的内部实现有关吧。

服务端

服务器端就是一个监听连接,处理请求的过程,上文我们已经讲过服务器端的处理大致处理过程,这里不再赘述。

TMultiplexedProtocol和TMultiplexedProcessor

  接下来我们将一下合并监听端口的主要的处理类,TMultiplexedProtocol为客户端使用类,TMultiplexedProcessor为服务器端使用类。前面的文章,我们提到过这两个类怎么用,也对两个类的调用方法进行的简单的封装处理,这里我想讲一下它们的内部时怎么处理请求的。

想要说明一个类的实现原理,最好的方法时带着大家去看下它的源代码,首先,我们看一下TMultiplexedProtocol的部分源码:

 

public override void WriteMessageBegin(TMessage tMessage) 		{ 			TMessageType type = tMessage.Type; 			if (type == TMessageType.Call || type == TMessageType.Oneway) 			{ 				base.WriteMessageBegin(new TMessage(this.ServiceName  + TMultiplexedProtocol.SEPARATOR + tMessage.Name, tMessage.Type,  tMessage.SeqID)); 				return; 			} 			base.WriteMessageBegin(tMessage); 		}

 

看过源码后,我们一目了然,是的,它把ServiceName写到了请求中,那么在服务器端时怎么处理的呢?同样,我们看下服务器端的处理方法:

 

public bool Process(TProtocol iprot, TProtocol oprot) 		{ 			bool result; 			try 			{ 				TMessage message = iprot.ReadMessageBegin(); 				if (message.Type != TMessageType.Call &&  message.Type != TMessageType.Oneway) 				{ 					this.Fail(oprot, message,  TApplicationException.ExceptionType.InvalidMessageType,  "Message type CALL or ONEWAY expected"); 					result = false; 				} 				else 				{ 					int num =  message.Name.IndexOf(TMultiplexedProtocol.SEPARATOR); 					if (num < 0) 					{ 						this.Fail(oprot, message,  TApplicationException.ExceptionType.InvalidProtocol,  "Service name not found in message name: " + message.Name  + ". Did you forget to use a TMultiplexProtocol in your client?"); 						result = false; 					} 					else 					{ 						string text = message.Name.Substring(0,  num); 						TProcessor tProcessor; 						if (!this.ServiceProcessorMap.TryGetValue(text, out tProcessor)) 						{ 							this.Fail(oprot, message,  TApplicationException.ExceptionType.InternalError,  "Service name not found: " + text  + ". Did you forget to call RegisterProcessor()?"); 							result = false; 						} 						else 						{ 							TMessage messageBegin = new  TMessage(message.Name.Substring(text.Length +  TMultiplexedProtocol.SEPARATOR.Length), message.Type, message.SeqID); 							result = tProcessor.Process(new  TMultiplexedProcessor.StoredMessageProtocol(iprot, messageBegin), oprot); 						} 					} 				} 			} 

 

  

 

看到源码中的第一个else分支,它解析出serviceName,然后在中ServiceProcessorMap集合中获取我们之前注册过的对应的请求处理器。

其他一些问题

  • + 服务器端被调用的方法不能返回Null类型,否则调用方法会抛出异常

  • + thrift框架进行RPC调用是不是线程安全的,因此,线程安全部分需要自己去处理

结尾

本小节我们讲述了Tsocket在实战中会遇到的一些坑,希望对您有帮助。