C#使用Thrift作为RPC框架入门(二)

上一篇 文章中我们讲述了Thrif的基本知识,包括在C#语言下使用需要用到的工具以及使用nuget安装thrift开发包,还描述了它支持的数据类型,以及它支持IDL的描述文件,和一个简单的例子。

接上文,我这里再补充两点关于IDL描述文件的:

  • Thrift支持Byte类型,我们需要用i8来表示,它对应C#中的sbyte类型,如果我们用byte关键字,我们会看到一个【WARNING】
  • 在使用字节数组时,我们需要用binary类型,如果我们用list,也会看到一个【WARNING】

生成的代码–两个接口两个类

我们从上一篇文章最后那个简单的例子中可以看到,Thrift框架把我们标记为service的结构生成了对应的一个类,这个类中有两个内部接口两个内部类。这就是我们使用该框架的重点部分。

两个内部接口

这两组接口代表了框架给我生成的两组方法,这两组方法中一组(ISync)是用于同步调用的方法,另一组(Iface)是用于异步调用的方法,而Iface又继承了ISync接口。

  • Iface接口被一个为签名为Client的类继承,这个类是我们使用RPF框架调用的客户端。
  • ISync接口被作为一个签名为Processor类的构造函数的参数,Processor作为响应我们客户端请求的处理器类,我们需要自己实现具体的Handler,去处理客户端的请求,该实现是作为Processor的构造函数的实参

两个内部类

在讲两个内部接口的时候,我们以及提过了这两个内部类的是干什么用的,接下来我们将通过一个具体的示例,来感受一下这个两个类的使用方法

示例

客户端代码:

  static void Main(string[] args)
        {
           TTransport framedTransport
                = new TSocket("127.0.0.1",9999);
            Thrift.Protocol.TCompactProtocol compactProtocol =
                new Thrift.Protocol.TCompactProtocol(framedTransport);
            ThriftIDL.Services.PeopleService.Client client 
                = new ThriftIDL.Services.PeopleService.Client(compactProtocol);
            framedTransport.Open();
            People people = new People();
            client.SetPeople(people);
           
        }

服务器端代码:

static void Main(string[] args)
        {
            Thrift.Transport.TServerSocket serverSocket = new Thrift.Transport.TServerSocket(9999,1000);
            TProcessor processor=   new ThriftIDL.Services.PeopleService.Processor(new PeoperServiceHandler());
            Thrift.Server.TSimpleServer server = new Thrift.Server.TSimpleServer(processor,serverSocket);
            server.Serve();
        }

服务器端处理器对应的代码:

 public class PeoperServiceHandler : ThriftIDL.Services.PeopleService.Iface
    {
        public People GEtPeople()
        {
            throw new NotImplementedException();
        }

        public void SetPeople(People people)
        {
            throw new NotImplementedException();
        }
    }

我这里只做演示,并没有实现具体的方法。

多路复用处理器

对我们服务器端的代码细细品味之后,我们会突然发现,如果我们有多个service,那么我们就要写多个这样的服务端代码,这个似乎是我们不能接收,更要命的是,我们要每个service都要对应一个监听端口,这个有点恐怖。

Thrift框架想我们之所想,急我们之所急,它提供了一个多路复用的处理器类–TMultiplexedProtocol(客户端)以及TMultiplexedProcessor(服务器端),接下来我们看一下TMultiplexedProcessor类是怎么使用的

示例2

服务器端代码:

 TMultiplexedProcessor multiplexedProcessor = new TMultiplexedProcessor();
 multiplexedProcessor.RegisterProcessor("serviceName1"
              , new Service1.Processor(new Service1Handler()));
               multiplexedProcessor.RegisterProcessor("serviceName2"
              , new Service2.Processor(new Service2Handler()));
               multiplexedProcessor.RegisterProcessor("serviceName3"
              , new Service3.Processor(new Service3Handler()));
            TServerSocket serverSocket = new Thrift.Transport.TServerSocket(Port, 4000);
         TThreadPoolServer   server = new TThreadPoolServer(multiplexedProcessor, serverSocket);
          server.Serve();

代码中的Service1.Processor、Service2.Processor、Service3.Processor是我们定义的service生成的代码类对应的客户端调用代码如下:

RPC:ServiceName1:

 Thrift.Transport.TSocket socket = new Thrift.Transport.TSocket(remoteAddress, port, 4000);
            TCompactProtocol compactProtocol = new TCompactProtocol(socket);
             TMultiplexedProtocol multiplexedProtocol = new TMultiplexedProtocol(compactProtocol, "serverName1");
            Service1.Client client1=new Service1.Client(multiplexedProtocol);
            socket.Open();

RPC:ServiceName2:

 Thrift.Transport.TSocket socket = new Thrift.Transport.TSocket(remoteAddress, port, 4000);
            TCompactProtocol compactProtocol = new TCompactProtocol(socket);
             TMultiplexedProtocol multiplexedProtocol = new TMultiplexedProtocol(compactProtocol, "serverName2");
            Service2.Client client1=new Service2.Client(multiplexedProtocol);
            socket.Open();

RPC:ServiceName3:

 Thrift.Transport.TSocket socket = new Thrift.Transport.TSocket(remoteAddress, port, 4000);
            TCompactProtocol compactProtocol = new TCompactProtocol(socket);
             TMultiplexedProtocol multiplexedProtocol = new TMultiplexedProtocol(compactProtocol, "serverName3");
            Service3.Client client1=new Service3.Client(multiplexedProtocol);
            socket.Open();

以上就是多路复用处理器,在开发过程中的使用方法,代码用使用到的类型,我们会在下一节中讲解。

那么,现在问题又来了,如果service数量巨多的话,这样我们会得到大量的重复的代码,我们应该怎样处理这种情况呢?对!我们对Client和Processor做进一步的封装,这里的封装我使用了约定胜于配置的架构策略。

封装后的多路复用

服务器端代码:

 public class RPCServer
    {
        TThreadPoolServer server = null;
        TMultiplexedProcessor multiplexedProcessor = null;
        public RPCServer(int Port)
        {
            multiplexedProcessor = new TMultiplexedProcessor();
            TServerSocket serverSocket = new Thrift.Transport.TServerSocket(Port, 4000);
            server = new TThreadPoolServer(multiplexedProcessor, serverSocket,new Thrift.Transport.TTransportFactory()
                ,new Thrift.Protocol.TCompactProtocol.Factory());
        }

        public void RegisterProcessor(string serverName, TProcessor processor)
        {
            multiplexedProcessor.RegisterProcessor(serverName,processor);
        }

        public void RegisterProcessor<T>(object ProcessorHandler) where T: TProcessor
        {
            string[] strArray = typeof(T).FullName.Split(new string[] { ".", "+" }, StringSplitOptions.RemoveEmptyEntries);
            string serverName = strArray[strArray.Length - 2].ToUpper();

            ConstructorInfo[] constructorInfos = typeof(T).GetConstructors();
            TProcessor processor = (T)constructorInfos[0].Invoke(new object[] { ProcessorHandler });

            multiplexedProcessor.RegisterProcessor(serverName, processor);
        }

        /// <summary>
        /// 会阻塞当前线程
        /// </summary>
        public void Start()
        {
            server.Serve();
        }

        public void Stop()
        {
            server.Stop();
        }
    }

客户端代码:

   public class RPCClient<T> : IDisposable
    {
        public T Instance;
        Thrift.Transport.TSocket socket = null;
        public RPCClient(string remoteAddress, int port)
        {
            socket = new Thrift.Transport.TSocket(remoteAddress, port, 4000);
            TCompactProtocol compactProtocol = new TCompactProtocol(socket);
            string[] strArray = typeof(T).FullName.Split(new string[] { ".", "+" }, StringSplitOptions.RemoveEmptyEntries);
            string serverName = strArray[strArray.Length - 2].ToUpper();
            TMultiplexedProtocol multiplexedProtocol = new TMultiplexedProtocol(compactProtocol, serverName);
            ConstructorInfo constructorInfo = typeof(T).GetConstructor(new Type[] { typeof(TProtocol) });
            Instance = (T)constructorInfo.Invoke(new object[] { multiplexedProtocol });
        }

        public void Open()
        {
            socket.Open();
        }

        public void Close()
        {
            socket.Close();
        }
        ......

我们使用一下代码进行服务器端service的注册:

  RPCProxy.RPCServer rPCServer = new RPCProxy.RPCServer(52364);
            rPCServer.RegisterProcessor("ServiceName1"
                , new ServiceName1.Processor(new ServiceName1Handler()));
        rPCServer.RegisterProcessor<ServiceName2.Processor>(,);
        rPCServer.RegisterProcessor<ServiceName3.Processor>(,);
        rPCServer.Start();

客户端的使用 用一下代码:

            RPCProxy.RPCClient<ServiceName1.Client> rPCClient1 =
                new RPCProxy.RPCClient<ServiceName1.Client>("127.0.0.1", 52364);

                  RPCProxy.RPCClient<ServiceName2.Client> rPCClient2 =
                new RPCProxy.RPCClient<ServiceName2.Client>("127.0.0.1", 52364);
                ......

结尾

这一小节我们详细讲解了Thrift框架生成的代码在Client和Server的使用方法,已经多路复用处理器的使用方法,我们也对多路复用进行了封装优化,使其更易用,其中我们使用了一下Thrift提供的类,我们并没有详细的讲解。我们将在下一节讲thrift的框架设计时来对这些类进行说明。

Tags: