Java 类org.apache.thrift.protocol.TBinaryProtocol.Factory 实例源码

public static void startRPCServer(leafServer leafserver , String ip , int port) throws Exception
{
    ServerSocket serverSocket = new ServerSocket(port,10000, InetAddress.getByName(ip));

    TServerSocket serverTransport = new TServerSocket(serverSocket);

    //设置协议工厂为TBinaryProtocolFactory
    Factory proFactory = new TBinaryProtocol.Factory();
    //关联处理器leafrpc的实现
    TProcessor processor = new leafrpc.Processor<leafrpc.Iface>(new RPCService(leafserver));
    TThreadPoolServer.Args args2 = new TThreadPoolServer.Args(serverTransport);
    args2.processor(processor);
    args2.protocolFactory(proFactory);
    TServer server = new TThreadPoolServer(args2);
    LOG.info("leaf RPCServer(type:TThreadPoolServer) start at ip:port : "+ ip +":" + port );
    server.serve();
}

public static void startRPCServer2(leafServer leafserver , String ip , int port) throws Exception
{
    //关联处理器leafrpc的实现
    TProcessor processor = new leafrpc.Processor<leafrpc.Iface>(new RPCService(leafserver));
    //传输通道，非阻塞模式
    InetSocketAddress address = new InetSocketAddress(InetAddress.getByName(ip),port);
    TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(address,10000);
    //多线程半同步半异步
    TThreadedSelectorServer.Args tArgs = new TThreadedSelectorServer.Args(serverTransport);
    tArgs.processor(processor);
    //二进制协议
    tArgs.protocolFactory(new TBinaryProtocol.Factory());
    //多线程半同步半异步的服务模型
    TServer server = new TThreadedSelectorServer(tArgs);
    LOG.info("leaf RPCServer(type:TThreadedSelectorServer) start at ip:port : "+ ip +":" + port );
    server.serve();
}

private static int testProcessor(TProcessor processor, List<ToIntFunction<HostAndPort>> clients)
        throws Exception
{
    try (TServerSocket serverTransport = new TServerSocket(0)) {
        TProtocolFactory protocolFactory = new Factory();
        TTransportFactory transportFactory = new TFramedTransport.Factory();
        TServer server = new TSimpleServer(new Args(serverTransport)
                .protocolFactory(protocolFactory)
                .transportFactory(transportFactory)
                .processor(processor));

        Thread serverThread = new Thread(server::serve);
        try {
            serverThread.start();

            int localPort = serverTransport.getServerSocket().getLocalPort();
            HostAndPort address = HostAndPort.fromParts("localhost", localPort);

            int sum = 0;
            for (ToIntFunction<HostAndPort> client : clients) {
                sum += client.applyAsInt(address);
            }
            return sum;
        }
        finally {
            server.stop();
            serverThread.interrupt();
        }
    }
}

public static void startRPCServer3(leafServer leafserver , String ip , int port) throws Exception
{
    TProcessor processor = new leafrpc.Processor<leafrpc.Iface>(new RPCService(leafserver));
    //传输通道，非阻塞模式
    InetSocketAddress address = new InetSocketAddress(InetAddress.getByName(ip),port);
    TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(address,10000);
    TNonblockingServer.Args args = new TNonblockingServer.Args(serverTransport);
    args.protocolFactory(new TBinaryProtocol.Factory());
    args.transportFactory(new TFramedTransport.Factory());
    args.processorFactory(new TProcessorFactory(processor));
    TServer server = new TNonblockingServer(args);
    LOG.info("leaf RPCServer(type:TNonblockingServerSocket) start at ip:port : "+ ip +":" + port );
    server.serve();
}

public static void startRPCServer4(leafServer leafserver , String ip , int port) throws Exception
{
    TProcessor processor = new leafrpc.Processor<leafrpc.Iface>(new RPCService(leafserver));
    //传输通道，非阻塞模式
    InetSocketAddress address = new InetSocketAddress(InetAddress.getByName(ip),port);
    TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(address,10000);
    THsHaServer.Args  args = new THsHaServer.Args(serverTransport);
    args.processor(processor);
    args.protocolFactory(new TBinaryProtocol.Factory());
    args.transportFactory(new TFramedTransport.Factory());
    TServer server = new THsHaServer(args);
    LOG.info("leaf RPCServer(type:THsHaServer) start at ip:port : "+ ip +":" + port );
    server.serve();
}

public static void main(String[] args) {
    try {
        int port = 9119;
        TServerTransport serverTransport = new TServerSocket(port);
        Factory proFactory = new TBinaryProtocol.Factory();
        Processor<Iface> processor = new Example.Processor<Example.Iface>(new Example.Iface() {

            @Override
            public void pong() throws TException {
                System.out.println("pong");
            }

            @Override
            public void ping() throws TException {
                System.out.println("ping");
            }
        });
        Args thriftArgs = new Args(serverTransport);
        thriftArgs.processor(processor);
        thriftArgs.protocolFactory(proFactory);
        TServer tserver = new TThreadPoolServer(thriftArgs);
        System.out.println("启动监听:" + port);
        tserver.serve();
    } catch (TTransportException e) {
        e.printStackTrace();
    }
}

@Override
protected void assertEqualsExcepted(List<org.apache.parquet.thrift.test.compat.StructWithUnionV2> expected, List<Object> found) throws Exception {
  List<StructWithUnionV2> scroogeExpected = new ArrayList<StructWithUnionV2>();
  for (org.apache.parquet.thrift.test.compat.StructWithUnionV2 tbase : expected) {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    TProtocol out = new Factory().getProtocol(new TIOStreamTransport(baos));
    tbase.write(out);
    TProtocol in = new Factory().getProtocol(new TIOStreamTransport(new ByteArrayInputStream(baos.toByteArray())));
    scroogeExpected.add(StructWithUnionV2$.MODULE$.decode(in));
  }
  assertEquals(scroogeExpected, found);
 }

/**
 * Start a new server.
 * @param port 
 * @throws TTransportException
 */
public ThriftMountainCarEnv(int port) throws TTransportException{
    TServerSocket serverTransport = new TServerSocket(port);
    ThriftEnvironment.Processor processor = new ThriftEnvironment.Processor(this);
    Factory protFactory = new TBinaryProtocol.Factory(true, true);
    TServer server = new TThreadPoolServer(processor, serverTransport, protFactory);
    System.out.println("Starting server on port " + port + " ...");
    server.serve();
}

/**
 * @param args
 */
@SuppressWarnings("unchecked")
public static void main(String[] args) {
    try{
        /*
        TNonblockingServerSocket socket = new TNonblockingServerSocket(PORT);
        final UserService.Processor processor = 
            new UserService.Processor<Iface>(new UserServerHandler());
        THsHaServer.Args arg = new THsHaServer.Args(socket);
        */
        /*
        org.apache.thrift.protocol.TBinaryProtocol.Factory protoFactory = new TBinaryProtocol.Factory(true, true);
        TServerTransport serverTransport = new TServerSocket(PORT);
        UserService.Processor processor = new UserService.Processor<Iface>(new UserServerHandler());
        THsHaServer.Args arg = new THsHaServer.Args((TNonblockingServerTransport) serverTransport);

        arg.protocolFactory(new TCompactProtocol.Factory());
        arg.transportFactory(new TFramedTransport.Factory());
        arg.processorFactory(new TProcessorFactory(processor));
        */
        TServerSocket serverTransport = new TServerSocket(PORT);
        UserService.Processor processor = new UserService.Processor(new UserServerHandler());
        Factory protFactory = new TBinaryProtocol.Factory(true, true);

        Args arg = new Args(serverTransport);
        arg.processor(processor);
        arg.protocolFactory(protFactory);

        TServer server = new TThreadPoolServer(arg);

        // TServer server = new THsHaServer(arg);

        System.out.println("service begin...");
        server.serve();
    }catch(Exception e) {
        e.printStackTrace();
        System.out.println("UserServer.java main function");
    }

}

/**
 * Start thrift server
 */
@SuppressWarnings("unchecked")
private void StartServer() {
    try{
        String DBName = "cadalrectest-77";

        TServerSocket serverTransport = new TServerSocket(7911);
        RecAPI.Processor processor = new RecAPI.Processor(new RecAPIImp(DBName));
        Factory factory = new TBinaryProtocol.Factory(true, true);

        Args args = new Args(serverTransport);
        args.processor(processor);
        args.protocolFactory(factory);

        TServer server = new TThreadPoolServer(args);
        System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");
        System.out.println("!!! CADAL new recommendation service is started !!!");
        System.out.println("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!");

        server.serve();

    }catch(Exception e){
        e.printStackTrace();

        System.out.println("-------------------------------------");
        System.out.println("--- Thrift service can not start! ---");
        System.out.println("-------------------------------------");
    }
}

public void startServer(){

        try {
/*          serverTransport = new TServerSocket(8585);
            Args args = new Args(serverTransport);
            Factory portFactory = new TBinaryProtocol.Factory(true, true);
            args.protocolFactory(portFactory);  
            Recommend.Processor process=new Processor(new RecommendServer());
            args.processor(process);

            TServer server = new TThreadPoolServer(args);
*/
            TServerSocket serverTransport = new TServerSocket(8585);
            Recommend.Processor processor = new Recommend.Processor(new RecommendServer());
            Factory protFactory = new TBinaryProtocol.Factory(true, true);

            Args args = new Args(serverTransport);
            args.processor(processor);
            args.protocolFactory(protFactory);

            TServer server = new TThreadPoolServer(args);

            server.serve();
        } catch (TTransportException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }

protected ThriftServerInfo startServer() throws Throwable {
    // 获取一个监听端口
    final int port = choseListenPort();
    ThriftServerInfo serverInfo = new ThriftServerInfo(LOACLHOST, port);
    final AtomicReference<Throwable> ex = new AtomicReference<Throwable>();

    Thread runner = new Thread("thrift-server-starter") {
        @Override
        public void run() {
            try {
                TServerTransport serverTransport = new TServerSocket(port);
                Factory proFactory = new TBinaryProtocol.Factory();
                Processor<Iface> processor = new Example.Processor<Example.Iface>(new Example.Iface() {

                    @Override
                    public void pong() throws TException {
                        logger.info("pong");
                    }

                    @Override
                    public void ping() throws TException {
                        logger.info("ping");
                    }
                });
                Args thriftArgs = new Args(serverTransport);
                thriftArgs.processor(processor);
                thriftArgs.protocolFactory(proFactory);
                TServer tserver = new TThreadPoolServer(thriftArgs);
                servers.add(tserver);
                logger.info("启动测试服务监听：" + port);
                tserver.serve();
            } catch (TTransportException e) {
                logger.error("thrift服务器启动失败", e);
                ex.set(e);
            }
        }
    };

    runner.start();

    Throwable throwable = ex.get();
    if (throwable != null) {
        throw throwable;
    }
    // 等待服务器启动
    Thread.sleep(1000);
    return serverInfo;
}

protected ThriftServerInfo startMulitServiceServer() throws Throwable {
    // 获取一个监听端口
    final int port = choseListenPort();
    ThriftServerInfo serverInfo = new ThriftServerInfo(LOACLHOST, port);
    final AtomicReference<Throwable> ex = new AtomicReference<Throwable>();
    // TODO
    Thread runner = new Thread("thrift-server-starter") {
        @Override
        public void run() {
            try {
                TMultiplexedProcessor processor = new TMultiplexedProcessor();
                TServerTransport serverTransport = new TServerSocket(port);
                Factory proFactory = new TBinaryProtocol.Factory();

                processor.registerProcessor("example", new Example.Processor<Example.Iface>(new Example.Iface() {

                    @Override
                    public void pong() throws TException {
                        logger.info("example pong");
                    }

                    @Override
                    public void ping() throws TException {
                        logger.info("example ping");
                    }
                }));

                processor.registerProcessor("other", new Other.Processor<Other.Iface>(new Other.Iface() {

                    @Override
                    public void pong() throws TException {
                        logger.info("other pong");
                    }

                    @Override
                    public void ping() throws TException {
                        logger.info("other ping");
                    }
                }));
                Args thriftArgs = new Args(serverTransport);
                thriftArgs.processor(processor);
                thriftArgs.protocolFactory(proFactory);
                TServer tserver = new TThreadPoolServer(thriftArgs);
                servers.add(tserver);
                logger.info("启动测试服务监听：" + port);
                tserver.serve();
            } catch (TTransportException e) {
                logger.error("thrift服务器启动失败", e);
                ex.set(e);
            }
        }
    };

    runner.start();

    Throwable throwable = ex.get();
    if (throwable != null) {
        throw throwable;
    }
    // 等待服务器启动
    Thread.sleep(1000);
    return serverInfo;
}