grpc（三）之grpc客户端使用连接池

时间：2022-05-27 22:51:32

　　本文使用commons-pool2来实现连接池应用

1、定义一个产生连接池的工厂，需要继承BasePooledObjectFactory，其用处是生产和销毁连接池中保存的对象。根据需求，现在池子里保存的应该是grpc客户端对象。

　　GrpcClientFactory类：

package com.oy.grpc;

import org.apache.commons.pool2.BasePooledObjectFactory;

import org.apache.commons.pool2.PooledObject;

import org.apache.commons.pool2.impl.DefaultPooledObject;

import com.oy.grpc.client.GrpcClient;

import com.oy.utils.UtilFunctions;

public class GrpcClientFactory extends BasePooledObjectFactory<GrpcClient> {

    @Override

    public GrpcClient create() throws Exception {

        return new GrpcClient("localhost", );

    }

    @Override

    public PooledObject<GrpcClient> wrap(GrpcClient client) {

        return new DefaultPooledObject<>(client);

    }

    @Override

    public void destroyObject(PooledObject<GrpcClient> p) throws Exception {

        UtilFunctions.log.info("==== GrpcClientFactory#destroyObject ====");

        p.getObject().shutdown();

        super.destroyObject(p);

    }

}

2、连接池GrpcClientPool类

package com.oy.grpc;

import org.apache.commons.pool2.impl.GenericObjectPool;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import com.oy.grpc.client.GrpcClient;

import com.oy.utils.UtilFunctions;

public class GrpcClientPool {

    private static GenericObjectPool<GrpcClient> objectPool = null;

    static {

        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();

        // 池中的最大连接数

        poolConfig.setMaxTotal(8);

        // 最少的空闲连接数

        poolConfig.setMinIdle(0);

        // 最多的空闲连接数

        poolConfig.setMaxIdle(8);

        // 当连接池资源耗尽时,调用者最大阻塞的时间,超时时抛出异常 单位:毫秒数

        poolConfig.setMaxWaitMillis(-1);

        // 连接池存放池化对象方式,true放在空闲队列最前面,false放在空闲队列最后

        poolConfig.setLifo(true);

        // 连接空闲的最小时间,达到此值后空闲连接可能会被移除,默认即为30分钟

        poolConfig.setMinEvictableIdleTimeMillis(1000L * 60L * 30L);// 连接耗尽时是否阻塞,默认为true

        poolConfig.setBlockWhenExhausted(true);

        objectPool = new GenericObjectPool<>(new GrpcClientFactory(), poolConfig);

    }

    public static GrpcClient borrowObject() {

        try {

            GrpcClient client = objectPool.borrowObject();

            UtilFunctions.log.info("=======total threads created: " + objectPool.getCreatedCount());

            return client;

        } catch (Exception e) {

            UtilFunctions.log.error("objectPool.borrowObject error, msg:{}, exception:{}", e.toString(), e);

        }

        return createClient();

    }

    public static void returnObject(GrpcClient client) {

        try {

            objectPool.returnObject(client);

        } catch (Exception e) {

            UtilFunctions.log.error("objectPool.returnObject error, msg:{}, exception:{}", e.toString(), e);

        }

    }

    private static GrpcClient createClient() {

        return new GrpcClient("localhost", 23333);

    }

}

3、客户端程序

　　这里仅仅简单列出了客户端GrpcClient类的代码，其他代码包括服务端代码见另一篇博客grpc（一）之helloworld。

package com.oy.grpc.client;

import java.lang.reflect.Method;

import java.util.concurrent.TimeUnit;

import com.oy.grpc.BookServiceGrpc;

import com.oy.grpc.GrpcClientPool;

import com.oy.grpc.GrpcLib.GrpcReply;

import com.oy.grpc.GrpcLib.addBookRequest;

import com.oy.grpc.GrpcLib.getUserByIdRequest;

import com.oy.grpc.UserServiceGrpc;

import com.oy.utils.UtilFunctions;

import io.grpc.ManagedChannel;

import io.grpc.ManagedChannelBuilder;

import io.grpc.StatusRuntimeException;

public class GrpcClient {

    public static String host = "localhost";

    private final ManagedChannel channel;

    private final UserServiceGrpc.UserServiceBlockingStub userBlockingStub;

    private final BookServiceGrpc.BookServiceBlockingStub bookBlockingStub;

    public GrpcClient(String host, int port) {

        channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build();

        userBlockingStub = UserServiceGrpc.newBlockingStub(channel);

        bookBlockingStub = BookServiceGrpc.newBlockingStub(channel);

    }

    public void shutdown() throws InterruptedException {

        channel.shutdown().awaitTermination(, TimeUnit.SECONDS);

    }

    @SuppressWarnings({ "rawtypes" })

    public static Object call(String rpcMethoddName, Object... args) throws Exception {

        UtilFunctions.log.info("=========== GrpcClient#call begin ===========");

        GrpcClient client = null;

        try {

            client = GrpcClientPool.borrowObject();

            // client = new GrpcClient(host, 23333);

            Class[] argsTypes = new Class[args.length];

            for (int i = ; i < args.length; i++) {

                UtilFunctions.log.info("args types: {}", args[i].getClass());

                argsTypes[i] = args[i].getClass();

            }

            Method method = client.getClass().getMethod(rpcMethoddName, argsTypes);

            Object result = method.invoke(client, args);

            UtilFunctions.log.info("=========== GrpcClient#call end ===========");

            return result;

        } catch (Exception e) {

            UtilFunctions.log.error("GrpcClient#call error, msg:{}, exception:{}", e.toString(), e);

            return null;

        } finally {

            if (client != null) {

                GrpcClientPool.returnObject(client);

                // client.shutdown();

            }

        }

    }

    // ============= User module =============

    public Object getUserById(Integer id) {

        UtilFunctions.log.info("=========== GrpcClient#getUserById begin ===========");

        getUserByIdRequest request = getUserByIdRequest.newBuilder().setId(id).build();

        GrpcReply response;

        try {

            response = userBlockingStub.getUserById(request);

            UtilFunctions.log.info("GrpcClient#getUserById response, code:{}, data:{}", response.getCode(),

                    response.getData());

        } catch (StatusRuntimeException e) {

            UtilFunctions.log.error("GrpcClient#addBook error, msg:{}, exception:{}", e.toString(), e);

            return null;

        }

        return response;

    }

    // ============= Book module =============

    public Object addBook(Integer id, String name, Double price) {

        UtilFunctions.log.info("=========== GrpcClient#addBook begin ===========");

        addBookRequest request = addBookRequest.newBuilder().setId(id).setName(name).setPrice(price).build();

        GrpcReply response;

        try {

            response = bookBlockingStub.addBook(request);

            UtilFunctions.log.info("GrpcClient#addBook response, code:{}, data:{}", response.getCode(),

                    response.getData());

            UtilFunctions.log.info("=========== GrpcClient#addBook end ===========");

        } catch (StatusRuntimeException e) {

            UtilFunctions.log.error("GrpcClient#addBook error, msg:{}, exception:{}", e.toString(), e);

            return null;

        }

        return response;

    }

}

4、客户端测试

package com.oy.grpc.client;

import com.oy.grpc.GrpcClientPool;

import com.oy.grpc.GrpcLib.GrpcReply;

import com.oy.utils.UtilFunctions;

public class TestService {

    public static void main(String[] args) throws Exception {

        for (int i = ; i < ; i++) {

            new Thread(new Runnable() {

                @Override

                public void run() {

                    GrpcReply result = null;

                    try {

                        // result = (GrpcReply) GrpcClient.call("getUserById", Integer.valueOf("1"));

                        // result = (GrpcReply) GrpcClient.call("getUserById", 2);

                        result = (GrpcReply) GrpcClient.call("addBook", , "thinking in java", 50.0);

                    } catch (Exception e) {

                        e.printStackTrace();

                    }

                    if (result != null) {

                        UtilFunctions.log.info("client call interface, get code:{}, data:{}", result.getCode(),

                                result.getData());

                    }

　　　　　　　　　　　　// 如果注释掉下面两句，则客户端程序结束后，服务端报java.io.IOException: 远程主机强迫关闭了一个现有的连接。

                    // UtilFunctions.log.info("TestService#main: objectPool is closing...");

                    // GrpcClientPool.getObjectPool().close();

                }

            }).start();

        }

    }

}

　　运行testService类的main()方法，客户端能正常调用grpc server得到数据，但是grpc服务端报错：

2019-04-11 14:29:30.458  INFO 1192 --- [-worker-ELG-3-1] i.g.n.NettyServerTransport.connections   : Transport failed

java.io.IOException: 远程主机强迫关闭了一个现有的连接。

　　出现这个问题的原因：客户端强制断开连接。参考https://*.com/questions/46802521/io-grpc-netty-nettyservertransport-notifyterminated，

　　grpc（三）之grpc客户端使用连接池

　　我在GrpcClientFactory里面也实现了销毁方法：

@Override

public void destroyObject(PooledObject<GrpcClient> p) throws Exception {

    UtilFunctions.log.info("==== GrpcClientFactory#destroyObject ====");

    p.getObject().shutdown();

    super.destroyObject(p);

}

　　但是运行testService类的main()方法结束后服务端程序就结束了，程序没有主动调用destroyObject()方法销毁池子中的对象，所以grpcClient也没有shutdown，所以报错。

5、启动客户端springboot项目来测试

package com.oy;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import com.oy.grpc.client.TestService;

@SpringBootApplication

public class Grpc007ClientMainApplication {

    public static void main(String[] args) throws Exception {

        SpringApplication.run(Grpc007ClientMainApplication.class, args);

        TestService.main(args);

    }

}

　　但是这样当关闭客户端程序，还是出现同样的问题。其实很好理解，因为关闭客户端程序时，池中的对象还处于空闲状态，没有销毁，destroyObject()方法没有调用，所以grpcClient也没有shutdown。

　　

6、解决方法

　　客户端程序关闭时，池也要close。

package com.oy;

import javax.annotation.PreDestroy;

import org.apache.commons.pool2.impl.GenericObjectPool;

import org.springframework.stereotype.Controller;

import com.oy.grpc.GrpcClientPool;

import com.oy.grpc.client.GrpcClient;

import com.oy.utils.UtilFunctions;

@Controller

public class InitController {

    @PreDestroy

    public void destroy() {

        UtilFunctions.log.info("InitController#destroy running...");

        GenericObjectPool<GrpcClient> objectPool = GrpcClientPool.getObjectPool();

        UtilFunctions.log.info("InitController#destroy, total threads created: " + objectPool.getCreatedCount());

        UtilFunctions.log.info("InitController#destroy objectPool is closing...");

        objectPool.close();

    }

}

　　

相关文章