由浅入深了解Thrift之客户端连接池化

package com.wy.thriftpool.commzkpool;

import java.lang.instrument.IllegalClassFormatException;
import java.lang.reflect.Constructor;

import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TBinaryProtocol.Factory;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadedSelectorServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.springframework.beans.factory.InitializingBean;

import com.wy.thrift.service.UserService.Processor;
import com.wy.thriftpool.commzkpool.support.ThriftServerAddressReporter;
import com.wy.thriftpool.commzkpool.support.ThriftServerIpTransfer;
import com.wy.thriftpool.commzkpool.support.impl.LocalNetworkIpTransfer;

/**
 * thrift server端,向zk中注册server address
 * 
 * @author wy
 * 
 */
public class ThriftServiceServerFactory implements InitializingBean {

    // thrift server 服务端口
    private Integer port;
    // default 权重
    private Integer priority = 1;
    // service实现类
    private Object service;
    // thrift server 注册路径
    private String configPath;

    private ThriftServerIpTransfer ipTransfer;
    // thrift server注册类
    private ThriftServerAddressReporter addressReporter;
    // thrift server开启服务
    private ServerThread serverThread;

    @Override
    public void afterPropertiesSet() throws Exception {
        if (ipTransfer == null) {
            ipTransfer = new LocalNetworkIpTransfer();
        }
        String ip = ipTransfer.getIp();
        if (ip == null) {
            throw new NullPointerException("cant find server ip...");
        }
        String hostname = ip + ":" + port + ":" + priority;
        Class<? extends Object> serviceClass = service.getClass();
        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
        Class<?>[] interfaces = serviceClass.getInterfaces();
        if (interfaces.length == 0) {
            throw new IllegalClassFormatException("service-class should implements Iface");
        }

        // reflect,load "Processor";
        Processor<?> processor = null;
        for (Class<?> clazz : interfaces) {
            String cname = clazz.getSimpleName();
            if (!cname.equals("Iface")) {
                continue;
            }
            String pname = clazz.getEnclosingClass().getName() + "$Processor";
            try {
                Class<?> pclass = classLoader.loadClass(pname);
                if (!pclass.isAssignableFrom(Processor.class)) {
                    continue;
                }
                Constructor<?> constructor = pclass.getConstructor(clazz);
                processor = (Processor<?>) constructor.newInstance(service);
                break;
            } catch (Exception e) {
                // TODO
            }
        }

        if (processor == null) {
            throw new IllegalClassFormatException("service-class should implements Iface");
        }
        // 需要单独的线程,因为serve方法是阻塞的.
        serverThread = new ServerThread(processor, port);
        serverThread.start();
        // report
        if (addressReporter != null) {
            addressReporter.report(configPath, hostname);
        }
    }

    class ServerThread extends Thread {
        private TServer server;

        ServerThread(Processor<?> processor, int port) throws Exception {
            // 设置传输通道
            TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(port);
            // 设置二进制协议
            Factory protocolFactory = new TBinaryProtocol.Factory();
            
            TThreadedSelectorServer.Args tArgs = new TThreadedSelectorServer.Args(serverTransport);
            tArgs.processor(processor);
            tArgs.transportFactory(new TFramedTransport.Factory());
            tArgs.protocolFactory(protocolFactory);
            int num = Runtime.getRuntime().availableProcessors() * 2 + 1;
            tArgs.selectorThreads(num);
            tArgs.workerThreads(num * 10);
            
            // 网络服务模型
            server = new TThreadedSelectorServer(tArgs);
            
        }

        @Override
        public void run() {
            try {
                server.serve();
            } catch (Exception e) {
                //TODO
            }
        }

        public void stopServer() {
            server.stop();
        }
    }

    public void close() {
        serverThread.stopServer();
    }

    public void setService(Object service) {
        this.service = service;
    }

    public void setPriority(Integer priority) {
        this.priority = priority;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    public void setIpTransfer(ThriftServerIpTransfer ipTransfer) {
        this.ipTransfer = ipTransfer;
    }

    public void setAddressReporter(ThriftServerAddressReporter addressReporter) {
        this.addressReporter = addressReporter;
    }

    public void setConfigPath(String configPath) {
        this.configPath = configPath;
    }
}