RPC实例Apache Thrift 下篇（2）

清疚 2022-05-30 02:49 299阅读 0赞

3、正式开始编码

我已经在CSDN的资源区上传了这个示例工程的所有代码（http://download.csdn.net/download/wanbf123/10266929）。读者可以直接到资源下载站进行下载（不收积分哦~~^\_^）。这篇文章将紧接上文，主要介绍这个工程几个主要的类代码。

3-1、编写服务端主程序

服务端主程序的类名：processor.MainProcessor，它负责在服务端启动Apache Thrift并且在服务监听启动成功后，连接到zookeeper，注册这个服务的基本信息。

这里要注意一下，Apache Thrift的服务监听是阻塞式的，所以processor.MainProcessor的Apache Thrift操作应该另起线程进行（processor.MainProcessor.StartServerThread），并且通过线程间的锁定操作，保证zookeeper的连接一定是在Apache Thrift成功启动后才进行。

  1. package processor;
  3. import java.io.IOException;
  4. import java.util.Set;
  5. import java.util.concurrent.Executors;
  7. import net.sf.json.JSONObject;
  9. import org.apache.commons.logging.Log;
  10. import org.apache.commons.logging.LogFactory;
  11. import org.apache.log4j.BasicConfigurator;
  12. import org.apache.thrift.TProcessor;
  13. import org.apache.thrift.protocol.TBinaryProtocol;
  14. import org.apache.thrift.protocol.TProtocol;
  15. import org.apache.thrift.server.ServerContext;
  16. import org.apache.thrift.server.TServerEventHandler;
  17. import org.apache.thrift.server.TThreadPoolServer;
  18. import org.apache.thrift.server.TThreadPoolServer.Args;
  19. import org.apache.thrift.transport.TServerSocket;
  20. import org.apache.thrift.transport.TTransport;
  21. import org.apache.thrift.transport.TTransportException;
  22. import org.apache.zookeeper.CreateMode;
  23. import org.apache.zookeeper.KeeperException;
  24. import org.apache.zookeeper.WatchedEvent;
  25. import org.apache.zookeeper.Watcher;
  26. import org.apache.zookeeper.ZooKeeper;
  27. import org.apache.zookeeper.ZooDefs.Ids;
  28. import org.apache.zookeeper.data.Stat;
  30. import business.BusinessServicesMapping;
  31. import thrift.iface.DIYFrameworkService;
  32. import thrift.iface.DIYFrameworkService.Iface;
  34. public class MainProcessor { 
  35.     static {
  36.         BasicConfigurator.configure();
  37.     }
  39.     /** * 日志 */
  40.     private static final Log LOGGER = LogFactory.getLog(MainProcessor.class);
  42.     private static final Integer SERVER_PORT = 8090;
  44.     /** * 专门用于锁定以保证这个主线程不退出的一个object对象 */
  45.     private static final Object WAIT_OBJECT = new Object();
  47.     /** * 标记apache thrift是否启动成功了 * 只有apache thrift启动成功了，才需要连接到zk */
  48.     private boolean isthriftStart = false;
  50.     public static void main(String[] args) {
  51.         /* * 主程序要做的事情： * * 1、启动thrift服务。并且服务调用者的请求 * 2、连接到zk，并向zk注册自己提供的服务名称，告知zk真实的访问地址、访问端口 * （向zk注册的服务，存储在BusinessServicesMapping这个类的K-V常量中） * */
  53.         //1、========启动thrift服务
  54.         MainProcessor mainProcessor = new MainProcessor();
  55.         mainProcessor.startServer();
  57.         // 一直等待，apache thrift启动完成
  58.         synchronized (mainProcessor) {
  59.             try {
  60.                 while(!mainProcessor.isthriftStart) {
  61.                     mainProcessor.wait();
  62.                 }
  63.             } catch (InterruptedException e) {
  64.                 MainProcessor.LOGGER.error(e);
  65.                 System.exit(-1);
  66.             }
  67.         }
  69.         //2、========连接到zk
  70.         try {
  71.             mainProcessor.connectZk();
  72.         } catch (IOException | KeeperException | InterruptedException e) {
  73.             MainProcessor.LOGGER.error(e);
  74.             System.exit(-1);
  75.         }
  77.         // 这个wait在业务层面，没有任何意义。只是为了保证这个守护线程不会退出
  78.         synchronized (MainProcessor.WAIT_OBJECT) {
  79.             try {
  80.                 MainProcessor.WAIT_OBJECT.wait();
  81.             } catch (InterruptedException e) {
  82.                 MainProcessor.LOGGER.error(e);
  83.                 System.exit(-1);
  84.             }
  85.         }
  86.     }
  88.     /** * 这个私有方法用于连接到zk上，并且注册相关服务 * @throws IOException * @throws InterruptedException * @throws KeeperException */
  89.     private void connectZk() throws IOException, KeeperException, InterruptedException {
  90.         // 读取这个服务提供者，需要在zk上注册的服务
  91.         Set<String> serviceNames = BusinessServicesMapping.SERVICES_MAPPING.keySet();
  92.         // 如果没有任何服务需要注册到zk，那么这个服务提供者就没有继续注册的必要了
  93.         if(serviceNames == null || serviceNames.isEmpty()) {
  94.             return;
  95.         }
  97.         // 默认的监听器
  98.         MyDefaultWatcher defaultWatcher = new MyDefaultWatcher();
  99.         // 连接到zk服务器集群，添加默认的watcher监听
  100.         ZooKeeper zk = new ZooKeeper("192.168.61.128:2181", 120000, defaultWatcher);
  102.         //创建一个父级节点Service
  103.         Stat pathStat = null;
  104.         try {
  105.             pathStat = zk.exists("/Service", defaultWatcher);
  106.             //如果条件成立，说明节点不存在（只需要判断一个节点的存在性即可）
  107.             //创建的这个节点是一个“永久状态”的节点
  108.             if(pathStat == null) {
  109.                 zk.create("/Service", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  110.             }
  111.         } catch(Exception e) {
  112.             System.exit(-1);
  113.         }
  115.         // 开始添加子级节点，每一个子级节点都表示一个这个服务提供者提供的业务服务
  116.         for (String serviceName : serviceNames) {
  117.             JSONObject nodeData = new JSONObject();
  118.             nodeData.put("ip", "127.0.0.1");
  119.             nodeData.put("port", MainProcessor.SERVER_PORT);
  120.             zk.create("/Service/" + serviceName, nodeData.toString().getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
  121.         }
  123.         //执行到这里，说明所有的service都启动完成了
  124.         MainProcessor.LOGGER.info("===================所有service都启动完成了，主线程开始启动===================");
  125.     }
  127.     /** * 这个私有方法用于开启Apache thrift服务端，并进行持续监听 * @throws TTransportException */
  128.     private void startServer() {
  129.         Thread startServerThread = new Thread(new StartServerThread());
  130.         startServerThread.start();
  131.     }
  133.     private class StartServerThread implements Runnable { 
  134.         @Override
  135.         public void run() {
  136.             MainProcessor.LOGGER.info("看到这句就说明thrift服务端准备工作 ....");
  138.             // 服务执行控制器（只要是调度服务的具体实现该如何运行）
  139.             TProcessor tprocessor = new DIYFrameworkService.Processor<Iface>(new DIYFrameworkServiceImpl());
  141.             // 基于阻塞式同步IO模型的Thrift服务，正式生产环境不建议用这个
  142.             TServerSocket serverTransport = null;
  143.             try {
  144.                 serverTransport = new TServerSocket(MainProcessor.SERVER_PORT);
  145.             } catch (TTransportException e) {
  146.                 MainProcessor.LOGGER.error(e);
  147.                 System.exit(-1);
  148.             }
  150.             // 为这个服务器设置对应的IO网络模型、设置使用的消息格式封装、设置线程池参数
  151.             Args tArgs = new Args(serverTransport);
  152.             tArgs.processor(tprocessor);
  153.             tArgs.protocolFactory(new TBinaryProtocol.Factory());
  154.             tArgs.executorService(Executors.newFixedThreadPool(100));
  156.             // 启动这个thrift服务
  157.             TThreadPoolServer server = new TThreadPoolServer(tArgs);
  158.             server.setServerEventHandler(new StartServerEventHandler());
  159.             server.serve();
  160.         }
  161.     }
  163.     /** * 为这个TThreadPoolServer对象，设置是一个事件处理器。 * 以便在TThreadPoolServer正式开始监听服务请求前，通知mainProcessor： * “Apache Thrift已经成功启动了” * @author yinwenjie * */
  164.     private class StartServerEventHandler implements TServerEventHandler { 
  165.         @Override
  166.         public void preServe() {
  167.             /* * 需要实现这个方法，以便在服务启动成功后， * 通知mainProcessor： “Apache Thrift已经成功启动了” * */
  168.             MainProcessor.this.isthriftStart = true;
  169.             synchronized (MainProcessor.this) {
  170.                 MainProcessor.this.notify();
  171.             }
  172.         }
  174.         /* (non-Javadoc) * @see org.apache.thrift.server.TServerEventHandler#createContext(org.apache.thrift.protocol.TProtocol, org.apache.thrift.protocol.TProtocol) */
  175.         @Override
  176.         public ServerContext createContext(TProtocol input, TProtocol output) {
  177.             /* * 无需实现 * */
  178.             return null;
  179.         }
  181.         @Override
  182.         public void deleteContext(ServerContext serverContext, TProtocol input, TProtocol output) {
  183.             /* * 无需实现 * */
  184.         }
  186.         @Override
  187.         public void processContext(ServerContext serverContext, TTransport inputTransport, TTransport outputTransport) {
  188.             /* * 无需实现 * */
  189.         }
  190.     }
  192.     /** * 这是默认的watcher，什么也没有，也不需要有什么<br> * 因为按照功能需求，服务器端并不需要监控zk上的任何目录变化事件 * @author yinwenjie */
  193.     private class MyDefaultWatcher implements Watcher { 
  194.         public void process(WatchedEvent event) {
  196.         }
  197.     }
  198. }

3-2、编写服务具体实现

服务端具体实现的代码很简单，就是在IDL脚本生成了java代码后，对DIYFrameworkService接口进行的实现。

  1. package processor;
  3. import java.nio.ByteBuffer;
  5. import net.sf.json.JSONObject;
  7. import org.apache.commons.lang.StringUtils;
  8. import org.apache.commons.logging.Log;
  9. import org.apache.commons.logging.LogFactory;
  10. import org.apache.thrift.TException;
  12. import business.BusinessService;
  13. import business.BusinessServicesMapping;
  14. import business.exception.BizException;
  15. import business.exception.ResponseCode;
  16. import business.pojo.AbstractPojo;
  17. import business.pojo.BusinessResponsePojo;
  18. import business.pojo.DescPojo;
  19. import thrift.iface.DIYFrameworkService.Iface;
  20. import thrift.iface.EXCCODE;
  21. import thrift.iface.RESCODE;
  22. import thrift.iface.Reponse;
  23. import thrift.iface.Request;
  24. import thrift.iface.ServiceException;
  25. import utils.JSONUtils;
  27. /** * IDL文件中，我们定义的唯一服务接口DIYFrameworkService.Iface的唯一实现 * @author yinwenjie * */
  28. public class DIYFrameworkServiceImpl implements Iface { 
  30.     /** * 日志 */
  31.     public static final Log LOGGER = LogFactory.getLog(DIYFrameworkServiceImpl.class);
  33.     /* (non-Javadoc) * @see thrift.iface.DIYFrameworkService.Iface#send(thrift.iface.Request) */
  34.     @SuppressWarnings("unchecked")
  35.     @Override
  36.     public Reponse send(Request request) throws ServiceException, TException {
  37.         /* * 由于MainProcessor中，在Apache Thrift 服务端启动时已经加入了线程池，所以这里就不需要再使用线程池了 * 这个服务方法的实现，需要做以下事情： * * 1、根据request中，描述的具体服务名称，在配置信息中查找具体的服务类 * 2、使用java的反射机制，调用具体的服务类(BusinessService接口的实现类)。 * 3、根据具体的业务处理结构，构造Reponse对象，并进行返回 * */
  39.         //1、===================
  40.         String serviceName = request.getServiceName();
  41.         String className = BusinessServicesMapping.SERVICES_MAPPING.get(serviceName);
  42.         //未发现服务
  43.         if(StringUtils.isEmpty(className)) {
  44.             return this.buildErrorReponse("无效的服务" , null);
  45.         }
  47.         //2、===================
  48.         // 首先得到以json为描述格式的请求参数信息
  49.         JSONObject paramJSON = null;
  50.         try {
  51.             byte [] paramJSON_bytes = request.getParamJSON();
  52.             if(paramJSON_bytes != null && paramJSON_bytes.length > 0) {
  53.                 String paramJSON_string = new String(paramJSON_bytes);
  54.                 paramJSON = JSONObject.fromObject(paramJSON_string);
  55.             }
  56.         } catch(Exception e) {
  57.             DIYFrameworkServiceImpl.LOGGER.error(e);
  58.             // 向调用者抛出异常
  59.             throw new ServiceException(EXCCODE.PARAMNOTFOUND, e.getMessage());
  60.         }
  62.         // 试图进行反射
  63.         BusinessService<AbstractPojo> businessServiceInstance = null;
  64.         try {
  65.             businessServiceInstance = (BusinessService<AbstractPojo>)Class.forName(className).newInstance();
  66.         } catch (Exception e) {
  67.             DIYFrameworkServiceImpl.LOGGER.error(e);
  68.             // 向调用者抛出异常
  69.             throw new ServiceException(EXCCODE.SERVICENOTFOUND, e.getMessage());
  70.         }
  72.         // 进行调用
  73.         AbstractPojo returnPojo = null;
  74.         try {
  75.             returnPojo = businessServiceInstance.handle(paramJSON);
  76.         } catch (BizException e) {
  77.             DIYFrameworkServiceImpl.LOGGER.error(e);
  78.             return this.buildErrorReponse(e.getMessage() , e.getResponseCode());
  79.         }
  81.         // 构造处理成功情况下的返回信息
  82.         BusinessResponsePojo responsePojo = new BusinessResponsePojo();
  83.         responsePojo.setData(returnPojo);
  84.         DescPojo descPojo = new DescPojo("", ResponseCode._200);
  85.         responsePojo.setDesc(descPojo);
  87.         // 生成json
  88.         String returnString = JSONUtils.toString(responsePojo);
  89.         byte[] returnBytes = returnString.getBytes();
  90.         ByteBuffer returnByteBuffer = ByteBuffer.allocate(returnBytes.length);
  91.         returnByteBuffer.put(returnBytes);
  92.         returnByteBuffer.flip();
  94.         // 构造response
  95.         Reponse reponse = new Reponse(RESCODE._200, returnByteBuffer);
  96.         return reponse;
  97.     }
  99.     /** * 这个私有方法，用于构造“Thrift中错误的返回信息” * @param erroe_mess * @return */
  100.     private Reponse buildErrorReponse(String erroe_mess , ResponseCode responseCode) {
  101.         // 构造返回信息
  102.         BusinessResponsePojo responsePojo = new BusinessResponsePojo();
  103.         responsePojo.setData(null);
  104.         DescPojo descPojo = new DescPojo(erroe_mess, responseCode == null?ResponseCode._504:responseCode);
  105.         responsePojo.setDesc(descPojo);
  107.         // 存储byteBuffer；
  108.         String responseJSON = JSONUtils.toString(responsePojo);
  109.         byte[] responseJSON_bytes = responseJSON.getBytes();
  110.         ByteBuffer byteBuffer = ByteBuffer.allocate(responseJSON_bytes.length);
  111.         byteBuffer.put(byteBuffer);
  112.         byteBuffer.flip();
  114.         Reponse reponse = new Reponse(RESCODE._500, byteBuffer);
  115.         return reponse;
  116.     }
  117. }

3-3、编写客户端实现

在上文中已经介绍过了，客户端有两件事情需要做：连接到zookeeper查询注册的服务该如何访问；然后向真实的服务提供者发起请求。代码如下：

  1. package client;
  3. import java.nio.ByteBuffer;
  4. import java.util.List;
  6. import net.sf.json.JSONObject;
  8. import org.apache.commons.lang.StringUtils;
  9. import org.apache.commons.logging.Log;
  10. import org.apache.commons.logging.LogFactory;
  11. import org.apache.log4j.BasicConfigurator;
  12. import org.apache.thrift.protocol.TBinaryProtocol;
  13. import org.apache.thrift.protocol.TProtocol;
  14. import org.apache.thrift.transport.TSocket;
  15. import org.apache.zookeeper.CreateMode;
  16. import org.apache.zookeeper.WatchedEvent;
  17. import org.apache.zookeeper.Watcher;
  18. import org.apache.zookeeper.ZooKeeper;
  19. import org.apache.zookeeper.ZooDefs.Ids;
  20. import org.apache.zookeeper.data.Stat;
  22. import thrift.iface.DIYFrameworkService.Client;
  23. import thrift.iface.Reponse;
  24. import thrift.iface.Request;
  25. import utils.JSONUtils;
  27. public class ThriftClient { 
  28.     /** * 日志 */
  29.     private static final Log LOGGER = LogFactory.getLog(ThriftClient.class);
  31.     private static final String SERVCENAME = "queryUserDetailService";
  33.     static {
  34.         BasicConfigurator.configure();
  35.     }
  37.     public static final void main(String[] main) throws Exception {
  38.         /* * 服务治理框架的客户端示例，要做以下事情： * * 1、连接到zk，查询当前zk下提供的服务列表中是否有自己需要的服务名称（queryUserDetailService） * 2、如果没有找到需要的服务名称，则客户端终止工作 * 3、如果找到了服务，则通过服务给出的ip，port，基于Thrift进行正式请求 * （这时，和zookeeper是否断开，关系就不大了） * */
  39.         // 1、===========================
  40.         // 默认的监听器
  41.         ClientDefaultWatcher defaultWatcher = new ClientDefaultWatcher();
  42.         // 连接到zk服务器集群，添加默认的watcher监听
  43.         ZooKeeper zk = new ZooKeeper("192.168.61.128:2181", 120000, defaultWatcher);
  45.         /* * 为什么客户端连接上来以后，也可能创建一个Service根目录呢？ * 因为正式的环境下，不能保证客户端一点就在服务器端全部准备好的情况下，再来做调用请求 * */
  46.         Stat pathStat = null;
  47.         try {
  48.             pathStat = zk.exists("/Service", defaultWatcher);
  49.             //如果条件成立，说明节点不存在（只需要判断一个节点的存在性即可）
  50.             //创建的这个节点是一个“永久状态”的节点
  51.             if(pathStat == null) {
  52.                 zk.create("/Service", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  53.             }
  54.         } catch(Exception e) {
  55.             System.exit(-1);
  56.         }
  58.         // 2、===========================
  59.         //获取服务列表(不需要做任何的事件监听，所以第二个参数可以为false)
  60.         List<String> serviceList = zk.getChildren("/Service", false);
  61.         if(serviceList == null || serviceList.isEmpty()) {
  62.             ThriftClient.LOGGER.info("未发现相关服务，客户端退出");
  63.             return;
  64.         }
  66.         //然后查看要找寻的服务是否在存在
  67.         boolean isFound = false;
  68.         byte[] data;
  69.         for (String serviceName : serviceList) {
  70.             if(StringUtils.equals(serviceName, ThriftClient.SERVCENAME)) {
  71.                 isFound = true;
  72.                 break;
  73.             }
  74.         }
  75.         if(!isFound) {
  76.             ThriftClient.LOGGER.info("未发现相关服务，客户端退出");
  77.             return;
  78.         } else {
  79.             data = zk.getData("/Service/" + ThriftClient.SERVCENAME, false, null);
  80.         }
  82.         /* * 执行到这里，zk的工作就完成了，接下来zk是否断开，就不重要了 * */
  83.         zk.close();
  84.         if(data == null || data.length == 0) {
  85.             ThriftClient.LOGGER.info("未发现有效的服务端地址，客户端退出");
  86.             return;
  87.         }
  88.         // 得到服务器地值说明
  89.         JSONObject serverTargetJSON = null;
  90.         String serverIp;
  91.         String serverPort;
  92.         try {
  93.             serverTargetJSON = JSONObject.fromObject(new String(data));
  94.             serverIp = serverTargetJSON.getString("ip");
  95.             serverPort = serverTargetJSON.getString("port");
  96.         } catch(Exception e) {
  97.             ThriftClient.LOGGER.error(e);
  98.             return;
  99.         }
  101.         //3、===========================
  102.         TSocket transport = new TSocket(serverIp, Integer.parseInt(serverPort));
  103.         TProtocol protocol = new TBinaryProtocol(transport);
  104.         // 准备调用参数
  105.         JSONObject jsonParam = new JSONObject();
  106.         jsonParam.put("username", "yinwenjie");
  107.         byte[] params = jsonParam.toString().getBytes();
  108.         ByteBuffer buffer = ByteBuffer.allocate(params.length);
  109.         buffer.put(params);
  110.         buffer.flip();
  111.         Request request = new Request(buffer, ThriftClient.SERVCENAME);
  113.         // 开始调用
  114.         Client client = new Client(protocol);
  115.         // 准备传输
  116.         transport.open();
  117.         // 正式调用接口
  118.         Reponse reponse = client.send(request);
  119.         byte[] responseBytes = reponse.getResponseJSON();
  120.         // 一定要记住关闭
  121.         transport.close();
  123.         // 将返回信息显示出来
  124.         ThriftClient.LOGGER.info("respinse value = " + new String(responseBytes));
  125.     }
  126. }
  128. /** * 这是默认的watcher，什么也没有，也不需要有什么<br> * 因为按照功能需求，客户端并不需要监控zk上的任何目录变化事件 * @author yinwenjie */
  129. class ClientDefaultWatcher implements Watcher {
  130.     public void process(WatchedEvent event) {
  132.     }
  133. }

3-4、工程结构说明

以上代码是服务器端、客户端的主要代码。整个工程还有其他的辅助代码，为了让各位读者能够看得清楚直接，我们将整个工程结构进行一下说明，下载后导入的工程结构如下图所示：

这里写图片描述

  1. 这是一个典型的JAVA工程。请使用 JDK 1.6+ 版本。我们将讲解整个工程结构。首先来看看这个工程中主要的package和它们的作用。
  2. business：具体的业务层逻辑都在这个包里面，其中exception包含了一个业务层异常的定义BizException，还有错误代码ResponseCode；impl包中放置具体的业务层实现，它们都必须实现BusinessService接口；Pojo是业务层对象模型。client：为了简单起见，我将服务端的实现和客户端的实现放置在一个工程中，client这个包就是客户端的实现代码了；utils包放置了两个工具类，用来进行日期格式化的DataUtils和用来进行json转换的JSONUtils。
  3. 定义的apache thrift IDL文件放置在thrift文件夹下面，名字叫做：demoHello.thrift；您可以使用它生成各种语言的代码；
  4. 工程需要maven的支持。

  5. 2016年08月08日，由网友OneZhous发现了一个程序的bug，这是由于Apache Thrift内部并不会在进行org.apache.thrift.TBaseHelper.copyBinary执行时，将java.nio.ByteBuffer自动进行flip()。所以在完成request和response对象设置后，需要开发人员自行进行flip()。感谢OneZhous对文章中的问题进行纠正，但是CSDN由于无法修改已上传的资源，所以还请各位读者在下载运行时注意这个问题：

    …… ByteBuffer buffer = ByteBuffer.allocate(params.length); buffer.put(params); buffer.flip(); // 以及位置 ByteBuffer byteBuffer = ByteBuffer.allocate(responseJSON_bytes.length); byteBuffer.put(byteBuffer); byteBuffer.flip(); ……

发表评论

表情：
评论列表 （有 0 条评论，299人围观）

还没有评论，来说两句吧...

相关阅读

    相关 RPC实例Apache Thrift 上篇

    1、概述 通过上一篇文章RPC的基本概念的介绍，相信读者已经理解了基本的RPC概念。为了加深这个理解，后面几篇文章我将详细讲解一款典型的RPC规范的实现Apache Th

    r囧r小猫r囧r小猫/ 0 赞/ 388 阅读

    相关 thrift rpc通信

    thrift rpc通信 框架   别人的简历： 负责抓取程序的开发和维护，对抓取内容进行数据提取、整理。 1、定向数据抓取程序的维护和开发，了解了Sqlite数据库

    小鱼儿小鱼儿/ 0 赞/ 526 阅读