Thrift不同服務類型的使用探索

  • 2019 年 10 月 5 日
  • 筆記

Thrift是一個軟體框架，用來進行可擴展且跨語言的服務的開發。它結合了功能強大的軟體堆棧和程式碼生成引擎，以構建在 C++, Java, Python, PHP, Ruby, Erlang, Perl, Haskell, C#, Cocoa, JavaScript, Node.js, Smalltalk, and OCaml 等等程式語言間無縫結合的、高效的服務。

Thrift最初由facebook開發，07年四月開放源碼，08年5月進入apache孵化器。thrift允許你定義一個簡單的定義文件中的數據類型和服務介面。以作為輸入文件，編譯器生成程式碼用來方便地生成RPC客戶端和伺服器通訊的無縫跨程式語言。

本篇博文編寫的目的是對Thrfit不同的服務類型進行整理，並結合程式碼示例進行說明。

一、目標

本篇博文編寫的目的是對Thrfit不同的服務類型進行整理，並結合程式碼示例進行說明。博文主要包含如下幾個部分：

1. 實例程式碼準備 2. 對不同的服務類型進行介紹說明，並給出示例 3. 非同步客戶端調用實例 4. Nifty庫的使用，包含服務端和客戶端程式碼示例

二、實例

2.1 說明

在這個示例中，我們主要在用戶介面中定義三個介面：保存用戶，根據name獲取用戶列表以及刪除用戶

如：

/**       * 保存用戶       *       * @param user       */      public boolean save(com.xxx.tutorial.thrift.entity.User user) throws org.apache.thrift.TException;        /**       * 根據name獲取用戶列表       *       * @param name       */      public java.util.List<com.xxx.tutorial.thrift.entity.User> findUsersByName(java.lang.String name) throws org.apache.thrift.TException;        /**       * 刪除用戶       *       * @param userId       */      public void deleteByUserId(int userId) throws com.xxx.tutorial.thrift.exception.UserNotFoundException, org.apache.thrift.TException;

2.2 編寫thrift文件

  • user.thrift用於定義用戶類
namespace java com.xxx.tutorial.thrift.entity    /**   * 用戶類   */  struct  User {    1:i32 userId,    2:string name  }
  • exception.thrift用於自定義異常
include "user.thrift"  include "exception.thrift"    namespace java com.xxx.tutorial.thrift.service    /**   * 用戶服務   */  service  UserService {      /**保存用戶*/    bool save(1:user.User user),      /**根據name獲取用戶列表*/    list<user.User> findUsersByName(1:string name),      /**刪除用戶*/    void deleteByUserId(1:i32 userId) throws (1: exception.UserNotFoundException e)  }

2.3 產生程式碼

根據thrift文件生成Java程式碼，這裡就不再描述，請參考以前的文章【一步步完成thrift rpc示例】。

2.4 服務介面程式碼

將生成的Java程式碼放入thrift-demo-interface模塊。如：

2.6 服務實現程式碼

在thrift-demo-service模組增加UserService的實現類。

UserServiceImpl.java的內容如下：

package com.xxx.tutorial.thrift.service.impl;    import java.util.Arrays;  import java.util.List;  import java.util.logging.Logger;    import org.apache.thrift.TException;    import com.xxx.tutorial.thrift.entity.User;  import com.xxx.tutorial.thrift.exception.UserNotFoundException;  import com.xxx.tutorial.thrift.service.UserService;    /**   * @author wangmengjun   *   */  public class UserServiceImpl implements UserService.Iface {      private static final Logger logger = Logger.getLogger(UserServiceImpl.class.getName());      public boolean save(User user) throws TException {      logger.info("方法save的參數user的內容==>" + user.toString());      return true;    }      public List<User> findUsersByName(String name) throws TException {      logger.info("方法findUsersByName的參數name的內容==>" + name);      return Arrays.asList(new User(1, "Wang"), new User(2, "Mengjun"));    }      public void deleteByUserId(int userId) throws UserNotFoundException, TException {      /**       * 直接模擬拋出異常，用於測試       */      logger.info("方法deleteByUserId的參數userId的內容==>" + userId);      throw new UserNotFoundException("1001", String.format("userId=%d的用戶不存在", userId));    }  }

經過上述幾個操作，用戶服務介面以及實現類的編寫就完畢了。

接下來，就能夠創建Thrift服務~

三、Thrift不同服務端類型

3.1 服務端類型

查看Thrift的TServer層次結構，我們可以看出，Thrift的服務端類型有如下幾種。服務端類型的描述如下：

  • TSimpleServer —— 單執行緒伺服器端使用標準的阻塞式 I/O
/**   * Simple singlethreaded server for testing.   *   */  public class TSimpleServer extends TServer {  ... ...  }
  • TThreadPoolServer —— 多執行緒伺服器端使用標準的阻塞式 I/O
/**   * Server which uses Java's built in ThreadPool management to spawn off   * a worker pool that   *   */  public class TThreadPoolServer extends TServer {    ... ...    }
  • TNonblockingServer —— 多執行緒伺服器端使用非阻塞式 I/O
  /**   * A nonblocking TServer implementation. This allows for fairness amongst all   * connected clients in terms of invocations.   *   * This server is inherently single-threaded. If you want a limited thread pool   * coupled with invocation-fairness, see THsHaServer.   *   * To use this server, you MUST use a TFramedTransport at the outermost   * transport, otherwise this server will be unable to determine when a whole   * method call has been read off the wire. Clients must also use TFramedTransport.   */  public class TNonblockingServer extends AbstractNonblockingServer {    ... ...    }
  • THsHaSercver —— 是TNonblockingServer的擴展, 多執行緒伺服器端使用非阻塞式 I/O
/**   * An extension of the TNonblockingServer to a Half-Sync/Half-Async server.   * Like TNonblockingServer, it relies on the use of TFramedTransport.   */  public class THsHaServer extends TNonblockingServer {    ... ...  }
  • TThreadedSelectorServer —— 多執行緒伺服器端使用非阻塞式 I/O
TThreadedSelectorServer是對以上NonblockingServer的擴充, 其分離了Accept和Read/Write的Selector執行緒,   同時引入Worker工作執行緒池. 它也是種Half-sync/Half-async的服務模型。
/**   * A Half-Sync/Half-Async server with a separate pool of threads to handle   * non-blocking I/O. Accepts are handled on a single thread, and a configurable   * number of nonblocking selector threads manage reading and writing of client   * connections. A synchronous worker thread pool handles processing of requests.   *   * Performs better than TNonblockingServer/THsHaServer in multi-core   * environments when the the bottleneck is CPU on the single selector thread   * handling I/O. In addition, because the accept handling is decoupled from   * reads/writes and invocation, the server has better ability to handle back-   * pressure from new connections (e.g. stop accepting when busy).   *   * Like TNonblockingServer, it relies on the use of TFramedTransport.   */  public class TThreadedSelectorServer extends AbstractNonblockingServer {  ... ...  }

另外，Facebook還開源了Nifty -- 一種基於netty的thrift服務端和客戶端實現。Nifty是facebook公司開源的，基於netty的thrift服務端和客戶端實現。然後使用此包就可以快速發布出基於netty的高效的服務端和客戶端程式碼。【https://github.com/facebook/nifty】

3.2 服務端創建步驟

（1） 創建一個transport對象 （2） 為transport對象創建輸入輸出protocol （3） 基於輸入輸出protocol創建processor （4） 等待連接請求並將之交給processor處理

如：

    try {          /**         * 1. 創建Transport         */        TServerSocket serverTransport = new TServerSocket(SERVER_PORT);        TServer.Args tArgs = new TServer.Args(serverTransport);          /**         * 2. 為Transport創建Protocol         */        tArgs.protocolFactory(new TBinaryProtocol.Factory());        // tArgs.protocolFactory(new TCompactProtocol.Factory());        // tArgs.protocolFactory(new TJSONProtocol.Factory());          /**         * 3. 為Protocol創建Processor         */        TProcessor tprocessor = new UserService.Processor<UserService.Iface>(new UserServiceImpl());        tArgs.processor(tprocessor);            /**         * 4. 創建Server並啟動         *         * org.apache.thrift.server.TSimpleServer - 簡單的單執行緒服務模型，一般用於測試         */        TServer server = new TSimpleServer(tArgs);        logger.info("UserService TSimpleServer start ....");        server.serve();          } catch (Exception e) {        logger.severe("Server start error!!!" + e.getLocalizedMessage());        e.printStackTrace();      }

接下來，我們就一起來完成不同服務端類型的程式碼示例以及客戶端調用實例~

四、TSimpleServer服務類型

4.1 服務端

package com.xxx.tutorial.thrift.server;    import java.util.logging.Logger;    import org.apache.thrift.TProcessor;  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 com.xxx.tutorial.thrift.service.UserService;  import com.xxx.tutorial.thrift.service.impl.UserServiceImpl;    /**   * @author wangmengjun   *   */  public class TSimpleServerExample {      private static final Logger logger = Logger.getLogger(TSimpleServerExample.class.getName());      private static final int SERVER_PORT = 9123;      public static void main(String[] args) {        try {          /**         * 1. 創建Transport         */        TServerSocket serverTransport = new TServerSocket(SERVER_PORT);        TServer.Args tArgs = new TServer.Args(serverTransport);          /**         * 2. 為Transport創建Protocol         */        tArgs.protocolFactory(new TBinaryProtocol.Factory());        // tArgs.protocolFactory(new TCompactProtocol.Factory());        // tArgs.protocolFactory(new TJSONProtocol.Factory());          /**         * 3. 為Protocol創建Processor         */        TProcessor tprocessor = new UserService.Processor<UserService.Iface>(new UserServiceImpl());        tArgs.processor(tprocessor);            /**         * 4. 創建Server並啟動         *         * org.apache.thrift.server.TSimpleServer - 簡單的單執行緒服務模型，一般用於測試         */        TServer server = new TSimpleServer(tArgs);        logger.info("UserService TSimpleServer start ....");        server.serve();          } catch (Exception e) {        logger.severe("Server start error!!!" + e.getLocalizedMessage());        e.printStackTrace();      }    }  }

4.2 客戶端

package com.xxx.tutorial.thrift.client;    import java.util.List;  import java.util.logging.Logger;    import org.apache.thrift.TException;  import org.apache.thrift.protocol.TBinaryProtocol;  import org.apache.thrift.protocol.TProtocol;  import org.apache.thrift.transport.TSocket;  import org.apache.thrift.transport.TTransport;  import org.apache.thrift.transport.TTransportException;    import com.xxx.tutorial.thrift.entity.User;  import com.xxx.tutorial.thrift.exception.UserNotFoundException;  import com.xxx.tutorial.thrift.service.UserService;    /**   *   * @author wangmengjun   *   */  public class UserClient {      private static final Logger logger = Logger.getLogger(UserClient.class.getName());      public static void main(String[] args) {        try {          TTransport transport = new TSocket("127.0.0.1", 9123);        TProtocol protocol = new TBinaryProtocol(transport);          UserService.Client client = new UserService.Client(protocol);        transport.open();          /**         * 查詢User列表         */        List<User> users = client.findUsersByName("wang");        logger.info("client.findUsersByName()方法結果 == >" + users);          /**         * 保存User         */        boolean isUserSaved = client.save(new User(101, "WMJ"));        logger.info("user saved result == > " + isUserSaved);          /**         * 刪除用戶         */        client.deleteByUserId(1002);          transport.close();        } catch (TTransportException e) {        logger.severe("TTransportException==>" + e.getLocalizedMessage());      } catch (UserNotFoundException e) {        logger.severe("UserNotFoundException==>" + e.getMessage());      } catch (TException e) {        logger.severe("TException==>" + e.getLocalizedMessage());      }    }  }

4.3 測試

  • 服務端控制台輸出
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".  SLF4J: Defaulting to no-operation (NOP) logger implementation  SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.  六月 09, 2017 8:29:14 下午 com.xxx.tutorial.thrift.server.TSimpleServerExample main  資訊: UserService TSimpleServer start ....  六月 09, 2017 8:29:27 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl findUsersByName  資訊: 方法findUsersByName的參數name的內容==>wang  六月 09, 2017 8:29:27 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl save  資訊: 方法save的參數user的內容==>User(userId:101, name:WMJ)  六月 09, 2017 8:29:27 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl deleteByUserId  資訊: 方法deleteByUserId的參數userId的內容==>1002
  • 客戶端控制台輸出
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".  SLF4J: Defaulting to no-operation (NOP) logger implementation  SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.  Received 1  六月 09, 2017 8:29:27 下午 com.xxx.tutorial.thrift.client.UserClient main  資訊: client.findUsersByName()方法結果 == >[User(userId:1, name:Wang), User(userId:2, name:Mengjun)]  Received 2  六月 09, 2017 8:29:27 下午 com.xxx.tutorial.thrift.client.UserClient main  資訊: user saved result == > true  Received 3  六月 09, 2017 8:29:27 下午 com.xxx.tutorial.thrift.client.UserClient main  嚴重: UserNotFoundException==>userId=1002的用戶不存在

五、TThreadPoolServer 服務類型

5.1 服務端

package com.xxx.tutorial.thrift.server;    import java.util.logging.Logger;    import org.apache.thrift.TProcessor;  import org.apache.thrift.protocol.TBinaryProtocol;  import org.apache.thrift.server.TServer;  import org.apache.thrift.server.TThreadPoolServer;  import org.apache.thrift.transport.TServerSocket;    import com.xxx.tutorial.thrift.service.UserService;  import com.xxx.tutorial.thrift.service.impl.UserServiceImpl;    /**   *   * @author wangmengjun   *   */  public class TThreadPoolServerExample {      private static final Logger logger = Logger.getLogger(TThreadPoolServerExample.class.getName());      private static final int SERVER_PORT = 9123;      public static void main(String[] args) {        try {          /**         * 1. 創建Transport         */        TServerSocket serverTransport = new TServerSocket(SERVER_PORT);        TThreadPoolServer.Args tArgs = new TThreadPoolServer.Args(serverTransport);          /**         * 2. 為Transport創建Protocol         */        tArgs.protocolFactory(new TBinaryProtocol.Factory());        // tArgs.protocolFactory(new TCompactProtocol.Factory());        // tArgs.protocolFactory(new TJSONProtocol.Factory());          /**         * 3. 為Protocol創建Processor         */        TProcessor tprocessor = new UserService.Processor<UserService.Iface>(new UserServiceImpl());        tArgs.processor(tprocessor);            /**         * 4. 創建Server並啟動         *         * org.apache.thrift.server.TThreadPoolServer         */        TServer server = new TThreadPoolServer(tArgs);        logger.info("UserService TThreadPoolServer start ....");        server.serve();          } catch (Exception e) {        logger.severe("Server start error!!!" + e.getLocalizedMessage());        e.printStackTrace();      }    }  }

5.2 客戶端

客戶端的程式碼可以和TSimpleServer中使用的Client程式碼一致，

package com.xxx.tutorial.thrift.client;    import java.util.List;  import java.util.logging.Logger;    import org.apache.thrift.TException;  import org.apache.thrift.protocol.TBinaryProtocol;  import org.apache.thrift.protocol.TProtocol;  import org.apache.thrift.transport.TSocket;  import org.apache.thrift.transport.TTransport;  import org.apache.thrift.transport.TTransportException;    import com.xxx.tutorial.thrift.entity.User;  import com.xxx.tutorial.thrift.exception.UserNotFoundException;  import com.xxx.tutorial.thrift.service.UserService;    /**   *   * @author wangmengjun   *   */  public class UserClient {      private static final Logger logger = Logger.getLogger(UserClient.class.getName());      public static void main(String[] args) {        try {          TTransport transport = new TSocket("127.0.0.1", 9123);        TProtocol protocol = new TBinaryProtocol(transport);          UserService.Client client = new UserService.Client(protocol);        transport.open();          /**         * 查詢User列表         */        List<User> users = client.findUsersByName("wang");        logger.info("client.findUsersByName()方法結果 == >" + users);          /**         * 保存User         */        boolean isUserSaved = client.save(new User(101, "WMJ"));        logger.info("user saved result == > " + isUserSaved);          /**         * 刪除用戶         */        client.deleteByUserId(1002);          transport.close();        } catch (TTransportException e) {        logger.severe("TTransportException==>" + e.getLocalizedMessage());      } catch (UserNotFoundException e) {        logger.severe("UserNotFoundException==>" + e.getMessage());      } catch (TException e) {        logger.severe("TException==>" + e.getLocalizedMessage());      }    }  }

5.3 測試

  • 服務端控制台輸出
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".  SLF4J: Defaulting to no-operation (NOP) logger implementation  SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.  六月 09, 2017 8:31:44 下午 com.xxx.tutorial.thrift.server.TThreadPoolServerExample main  資訊: UserService TThreadPoolServer start ....  六月 09, 2017 8:32:05 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl findUsersByName  資訊: 方法findUsersByName的參數name的內容==>wang  六月 09, 2017 8:32:05 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl save  資訊: 方法save的參數user的內容==>User(userId:101, name:WMJ)  六月 09, 2017 8:32:05 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl deleteByUserId  資訊: 方法deleteByUserId的參數userId的內容==>1002
  • 客戶端控制台輸出
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".  SLF4J: Defaulting to no-operation (NOP) logger implementation  SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.  Received 1  六月 09, 2017 8:32:05 下午 com.xxx.tutorial.thrift.client.UserClient main  資訊: client.findUsersByName()方法結果 == >[User(userId:1, name:Wang), User(userId:2, name:Mengjun)]  Received 2  六月 09, 2017 8:32:05 下午 com.xxx.tutorial.thrift.client.UserClient main  資訊: user saved result == > true  Received 3  六月 09, 2017 8:32:05 下午 com.xxx.tutorial.thrift.client.UserClient main  嚴重: UserNotFoundException==>userId=1002的用戶不存在

六、TNonblockingServer 服務類型

6.1 服務端

package com.xxx.tutorial.thrift.server;    import java.util.logging.Logger;    import org.apache.thrift.TProcessor;  import org.apache.thrift.protocol.TBinaryProtocol;  import org.apache.thrift.server.TNonblockingServer;  import org.apache.thrift.server.TServer;  import org.apache.thrift.transport.TFramedTransport;  import org.apache.thrift.transport.TNonblockingServerSocket;  import org.apache.thrift.transport.TNonblockingServerTransport;    import com.xxx.tutorial.thrift.service.UserService;  import com.xxx.tutorial.thrift.service.impl.UserServiceImpl;    /**   *   * @author wangmengjun   *   */  public class TNonblockingServerExample {      private static final Logger logger = Logger.getLogger(TNonblockingServerExample.class.getName());      private static final int SERVER_PORT = 9123;      public static void main(String[] args) {        try {          /**         * 1. 創建Transport         */        TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(SERVER_PORT);        TNonblockingServer.Args tArgs = new TNonblockingServer.Args(serverTransport);          /**         * 2. 為Transport創建Protocol         */        tArgs.transportFactory(new TFramedTransport.Factory());        tArgs.protocolFactory(new TBinaryProtocol.Factory());        // tArgs.protocolFactory(new TCompactProtocol.Factory());        // tArgs.protocolFactory(new TJSONProtocol.Factory());          /**         * 3. 為Protocol創建Processor         */        TProcessor tprocessor = new UserService.Processor<UserService.Iface>(new UserServiceImpl());        tArgs.processor(tprocessor);          /**         * 4. 創建Server並啟動         */        TServer server = new TNonblockingServer(tArgs);        logger.info("UserService TNonblockingServer start ....");        server.serve();        } catch (Exception e) {        logger.severe("Server start error!!!" + e.getLocalizedMessage());        e.printStackTrace();      }    }  }

6.2 客戶端

package com.xxx.tutorial.thrift.client;    import java.util.List;  import java.util.logging.Logger;    import org.apache.thrift.TException;  import org.apache.thrift.protocol.TBinaryProtocol;  import org.apache.thrift.protocol.TProtocol;  import org.apache.thrift.transport.TFramedTransport;  import org.apache.thrift.transport.TSocket;  import org.apache.thrift.transport.TTransport;  import org.apache.thrift.transport.TTransportException;    import com.xxx.tutorial.thrift.entity.User;  import com.xxx.tutorial.thrift.exception.UserNotFoundException;  import com.xxx.tutorial.thrift.service.UserService;    public class UserClient2 {      private static final Logger logger = Logger.getLogger(UserClient.class.getName());      public static void main(String[] args) {        try {          TTransport transport = new TFramedTransport(new TSocket("127.0.0.1", 9123, 3000));        TProtocol protocol = new TBinaryProtocol(transport);          UserService.Client client = new UserService.Client(protocol);        transport.open();          /**         * 查詢User列表         */        List<User> users = client.findUsersByName("wang");        logger.info("client.findUsersByName()方法結果 == >" + users);          /**         * 保存User         */        boolean isUserSaved = client.save(new User(101, "WMJ"));        logger.info("user saved result == > " + isUserSaved);          /**         * 刪除用戶         */        client.deleteByUserId(1002);          transport.close();        } catch (TTransportException e) {        logger.severe("TTransportException==>" + e.getLocalizedMessage());      } catch (UserNotFoundException e) {        logger.severe("UserNotFoundException==>" + e.getLocalizedMessage());      } catch (TException e) {        logger.severe("TException==>" + e.getLocalizedMessage());      }    }  }

6.3 測試

  • 服務端控制台輸出
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".  SLF4J: Defaulting to no-operation (NOP) logger implementation  SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.  六月 09, 2017 8:34:38 下午 com.xxx.tutorial.thrift.server.TNonblockingServerExample main  資訊: UserService TNonblockingServer start ....  六月 09, 2017 8:34:59 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl findUsersByName  資訊: 方法findUsersByName的參數name的內容==>wang  六月 09, 2017 8:34:59 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl save  資訊: 方法save的參數user的內容==>User(userId:101, name:WMJ)  六月 09, 2017 8:34:59 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl deleteByUserId  資訊: 方法deleteByUserId的參數userId的內容==>1002
  • 客戶端控制台輸出
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".  SLF4J: Defaulting to no-operation (NOP) logger implementation  SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.  Received 1  六月 09, 2017 8:34:59 下午 com.xxx.tutorial.thrift.client.UserClient2 main  資訊: client.findUsersByName()方法結果 == >[User(userId:1, name:Wang), User(userId:2, name:Mengjun)]  Received 2  六月 09, 2017 8:34:59 下午 com.xxx.tutorial.thrift.client.UserClient2 main  資訊: user saved result == > true  Received 3  六月 09, 2017 8:34:59 下午 com.xxx.tutorial.thrift.client.UserClient2 main  嚴重: UserNotFoundException==>userId=1002的用戶不存在

七、THsHaServer服務類型

7.1 服務端

package com.xxx.tutorial.thrift.server;    import java.util.logging.Logger;    import org.apache.thrift.TProcessor;  import org.apache.thrift.protocol.TBinaryProtocol;  import org.apache.thrift.server.THsHaServer;  import org.apache.thrift.server.TServer;  import org.apache.thrift.transport.TFramedTransport;  import org.apache.thrift.transport.TNonblockingServerSocket;    import com.xxx.tutorial.thrift.service.UserService;  import com.xxx.tutorial.thrift.service.impl.UserServiceImpl;    /**   * @author wangmengjun   *   */  public class THsHaServerExample {      private static final Logger logger = Logger.getLogger(THsHaServerExample.class.getName());      private static final int SERVER_PORT = 9123;      public static void main(String[] args) {        try {          /**         * 1. 創建Transport         */        //TServerSocket serverTransport = new TServerSocket(SERVER_PORT);        TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(SERVER_PORT);        THsHaServer.Args tArgs = new THsHaServer.Args(serverTransport);          /**         * 2. 為Transport創建Protocol         */        tArgs.transportFactory(new TFramedTransport.Factory());        tArgs.protocolFactory(new TBinaryProtocol.Factory());        // tArgs.protocolFactory(new TCompactProtocol.Factory());        // tArgs.protocolFactory(new TJSONProtocol.Factory());          /**         * 3. 為Protocol創建Processor         */        TProcessor tprocessor = new UserService.Processor<UserService.Iface>(new UserServiceImpl());        tArgs.processor(tprocessor);            /**         * 4. 創建Server並啟動         *         */        //半同步半非同步的服務模型        TServer server = new THsHaServer(tArgs);        logger.info("UserService TSimpleServer start ....");        server.serve();          } catch (Exception e) {        logger.severe("Server start error!!!" + e.getLocalizedMessage());        e.printStackTrace();      }    }  }

7.2 客戶端

package com.xxx.tutorial.thrift.client;    import java.util.List;  import java.util.logging.Logger;    import org.apache.thrift.TException;  import org.apache.thrift.protocol.TBinaryProtocol;  import org.apache.thrift.protocol.TProtocol;  import org.apache.thrift.transport.TFramedTransport;  import org.apache.thrift.transport.TSocket;  import org.apache.thrift.transport.TTransport;  import org.apache.thrift.transport.TTransportException;    import com.xxx.tutorial.thrift.entity.User;  import com.xxx.tutorial.thrift.exception.UserNotFoundException;  import com.xxx.tutorial.thrift.service.UserService;    public class UserClient2 {      private static final Logger logger = Logger.getLogger(UserClient.class.getName());      public static void main(String[] args) {        try {          TTransport transport = new TFramedTransport(new TSocket("127.0.0.1", 9123, 3000));        TProtocol protocol = new TBinaryProtocol(transport);          UserService.Client client = new UserService.Client(protocol);        transport.open();          /**         * 查詢User列表         */        List<User> users = client.findUsersByName("wang");        logger.info("client.findUsersByName()方法結果 == >" + users);          /**         * 保存User         */        boolean isUserSaved = client.save(new User(101, "WMJ"));        logger.info("user saved result == > " + isUserSaved);          /**         * 刪除用戶         */        client.deleteByUserId(1002);          transport.close();        } catch (TTransportException e) {        logger.severe("TTransportException==>" + e.getLocalizedMessage());      } catch (UserNotFoundException e) {        logger.severe("UserNotFoundException==>" + e.getLocalizedMessage());      } catch (TException e) {        logger.severe("TException==>" + e.getLocalizedMessage());      }    }  }

7.3 測試

  • 服務端控制台輸出
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".  SLF4J: Defaulting to no-operation (NOP) logger implementation  SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.  六月 09, 2017 8:57:26 下午 com.xxx.tutorial.thrift.server.THsHaServerExample main  資訊: UserService TSimpleServer start ....  六月 09, 2017 8:57:43 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl findUsersByName  資訊: 方法findUsersByName的參數name的內容==>wang  六月 09, 2017 8:57:43 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl save  資訊: 方法save的參數user的內容==>User(userId:101, name:WMJ)  六月 09, 2017 8:57:43 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl deleteByUserId  資訊: 方法deleteByUserId的參數userId的內容==>1002
  • 客戶端控制台輸出
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".  SLF4J: Defaulting to no-operation (NOP) logger implementation  SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.  Received 1  六月 09, 2017 8:57:43 下午 com.xxx.tutorial.thrift.client.UserClient2 main  資訊: client.findUsersByName()方法結果 == >[User(userId:1, name:Wang), User(userId:2, name:Mengjun)]  Received 2  六月 09, 2017 8:57:43 下午 com.xxx.tutorial.thrift.client.UserClient2 main  資訊: user saved result == > true  Received 3  六月 09, 2017 8:57:43 下午 com.xxx.tutorial.thrift.client.UserClient2 main  嚴重: UserNotFoundException==>userId=1002的用戶不存在

八、Nifty服務類型Facebook還開源了Nifty -- 一種基於netty的thrift服務端和客戶端實現。

Nifty是facebook公司開源的，基於netty的thrift服務端和客戶端實現。

使用Nifty，我們只要只要導入Nifty的jar包即可。

    <dependency>        <groupId>com.facebook.nifty</groupId>        <artifactId>nifty-core</artifactId>        <version>0.10.0</version>      </dependency>

8.1 服務端

package com.xxx.tutorial.thrift.server;    import org.apache.thrift.TProcessor;    import com.facebook.nifty.core.NettyServerTransport;  import com.facebook.nifty.core.ThriftServerDef;  import com.facebook.nifty.core.ThriftServerDefBuilder;  import com.xxx.tutorial.thrift.service.UserService;  import com.xxx.tutorial.thrift.service.impl.UserServiceImpl;    public class NiftyServer {      public static void main(String[] args) {        // Create the handler      UserService.Iface userServiceImpl = new UserServiceImpl();        // Create the processor      TProcessor processor = new UserService.Processor<>(userServiceImpl);        // Build the server definition      ThriftServerDef serverDef = new ThriftServerDefBuilder().withProcessor(processor).listen(9123).build();        // Create the server transport      final NettyServerTransport server = new NettyServerTransport(serverDef);        // Create netty boss and executor thread pools  /*    ExecutorService bossExecutor = Executors.newCachedThreadPool();      ExecutorService workerExecutor = Executors.newCachedThreadPool();        // Start the server      server.start(bossExecutor, workerExecutor);*/      System.out.println("啟動~~~");      server.start();        // Arrange to stop the server at shutdown      Runtime.getRuntime().addShutdownHook(new Thread() {        @Override        public void run() {          try {            server.stop();          } catch (InterruptedException e) {            Thread.currentThread().interrupt();          }        }      });    }  }

8.2 客戶端

如下示例給出NiftyClient來完成調用，使用NiftyClient需要導入相關jar包。

    <dependency>        <groupId>com.facebook.nifty</groupId>        <artifactId>nifty-client</artifactId>        <version>0.10.0</version>      </dependency>
package com.xxx.tutorial.thrift.client;    import java.net.InetSocketAddress;  import java.util.List;  import java.util.logging.Logger;    import org.apache.thrift.TException;  import org.apache.thrift.protocol.TBinaryProtocol;  import org.apache.thrift.transport.TTransport;  import org.apache.thrift.transport.TTransportException;    import com.facebook.nifty.client.NiftyClient;  import com.xxx.tutorial.thrift.entity.User;  import com.xxx.tutorial.thrift.service.UserService;    public class ThriftNiftyClient {      private static final Logger logger = Logger.getLogger(ThriftNiftyClient.class.getName());      @SuppressWarnings({ "resource" })    public static void main(String[] args) {        NiftyClient niftyClient = new NiftyClient();      InetSocketAddress address = new InetSocketAddress("127.0.0.1", 9123);      try {        TTransport transport = niftyClient.connectSync(address);        TBinaryProtocol tp = new TBinaryProtocol(transport);        UserService.Client userService = new UserService.Client(tp);        logger.info("userService.findUsersByName 方法調用~");        List<User> users = userService.findUsersByName("wang");        logger.info("userService.findUsersByName 調用結果~");        logger.info("users ==> " + users);        } catch (TTransportException e) {        logger.severe("TTransportException ==> " + e.getMessage());      } catch (InterruptedException e) {        logger.severe("InterruptedException ==> " + e.getMessage());      } catch (TException e) {        logger.severe("TException ==> " + e.getMessage());      }    }  }

8.3 測試

  • 服務端控制台輸出
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".  SLF4J: Defaulting to no-operation (NOP) logger implementation  SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.  啟動~~~  六月 09, 2017 9:01:44 下午 com.xxx.tutorial.thrift.service.impl.UserServiceImpl findUsersByName  資訊: 方法findUsersByName的參數name的內容==>wang
  • 客戶端控制台輸出
六月 09, 2017 9:01:44 下午 com.xxx.tutorial.thrift.client.ThriftNiftyClient main  資訊: userService.findUsersByName 方法調用~  Received 1  六月 09, 2017 9:01:44 下午 com.xxx.tutorial.thrift.client.ThriftNiftyClient main  資訊: userService.findUsersByName 調用結果~  六月 09, 2017 9:01:44 下午 com.xxx.tutorial.thrift.client.ThriftNiftyClient main  資訊: users ==> [User(userId:1, name:Wang), User(userId:2, name:Mengjun)]

至此，Thrift多種服務端的編寫已經完成，但是上述Client端的編寫基本上採用同步來實現~

下面給出一個非同步客戶端的示例。

九、非同步客戶端

package com.xxx.tutorial.thrift.client;    import java.util.List;  import java.util.concurrent.CountDownLatch;  import java.util.concurrent.TimeUnit;  import java.util.logging.Logger;    import org.apache.thrift.async.AsyncMethodCallback;  import org.apache.thrift.async.TAsyncClientManager;  import org.apache.thrift.protocol.TBinaryProtocol;  import org.apache.thrift.protocol.TProtocolFactory;  import org.apache.thrift.transport.TNonblockingSocket;  import org.apache.thrift.transport.TNonblockingTransport;    import com.xxx.tutorial.thrift.entity.User;  import com.xxx.tutorial.thrift.service.UserService;    public class UserAsynClient {      private static final Logger logger = Logger.getLogger(UserAsynClient.class.getName());      public static void main(String[] args) {      try {        TAsyncClientManager clientManager = new TAsyncClientManager();        TNonblockingTransport transport = new TNonblockingSocket("127.0.0.1", 9123, 3000);          TProtocolFactory tprotocol = new TBinaryProtocol.Factory();          UserService.AsyncClient asyncClient = new UserService.AsyncClient(tprotocol, clientManager, transport);          System.out.println("Client start .....");          CountDownLatch latch = new CountDownLatch(1);        /*AsynCallback_FindUsersByName callBack = new AsynCallback_FindUsersByName(latch);        asyncClient.findUsersByName("wang", callBack);*/          AsynCallback_SaveUser saveUserCallBack = new AsynCallback_SaveUser(latch);        asyncClient.save(new User(10001,"A name"), saveUserCallBack);          System.out.println("call method findUsersByName_call .... end");          boolean wait = latch.await(30, TimeUnit.SECONDS);          System.out.println("latch.await =:" + wait);        } catch (Exception e) {        e.printStackTrace();      }      System.out.println("startClient end.");    }      public static class AsynCallback_FindUsersByName implements AsyncMethodCallback<List<User>> {        private CountDownLatch latch;        public AsynCallback_FindUsersByName(CountDownLatch latch) {        this.latch = latch;      }        public void onComplete(List<User> response) {          logger.info("onComplete ==> findUsersByName_call");        try {          logger.info("findUsersByName_call response ==> " + response.toString());        } finally {          latch.countDown();        }        }        public void onError(Exception exception) {        logger.severe("onError ==> " + exception.getMessage());        latch.countDown();      }    }      public static class AsynCallback_SaveUser implements AsyncMethodCallback<Boolean> {        private CountDownLatch latch;        public AsynCallback_SaveUser(CountDownLatch latch) {        this.latch = latch;      }        public void onComplete(Boolean response) {        logger.info("onComplete ==> save_call");        try {          logger.info("save_call response ==> " + response.toString());        } finally {          latch.countDown();        }      }        public void onError(Exception exception) {        logger.severe("onError ==> " + exception.getMessage());        latch.countDown();      }      }  }

上述採用AsyncClient 來完成調用，使用AsyncClient 的時候，需要編寫一個用於回調的執行緒類。

UserService.AsyncClient asyncClient = new UserService.AsyncClient(tprotocol, clientManager, transport);

使用上述給出的NiftyServer類，啟動服務，然後運行UserAsynClient類，就完成一個非同步客戶端示例。

運行結果~

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".  SLF4J: Defaulting to no-operation (NOP) logger implementation  SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.  Client start .....  call method findUsersByName_call .... end  Received 0  六月 12, 2017 10:52:44 上午 com.xxx.tutorial.thrift.client.UserAsynClient$AsynCallback_SaveUser onComplete  資訊: onComplete ==> save_call  六月 12, 2017 10:52:44 上午 com.xxx.tutorial.thrift.client.UserAsynClient$AsynCallback_SaveUser onComplete  資訊: save_call response ==> true  latch.await =:true  startClient end.

至此，本篇博文的目標內容就完成了。

Tags:

VirMach 便宜 VPS