RPC 进阶 - 线程池

发表于 更新于

在 Rpc Introduction 中简单写了一个 RPC Framework。在本节中，对 Edith 新增采用线程池支持多线程的特性。

背景

实际开发中，我们的应用程序是不能盲目的开辟线程的，这样会造成资源的极大浪费。若采用线程池，可以减少创建和销毁线程的次数，让每个线程可以多次使用，并且可以根据系统的承受能力，调整线程池中工作线程的数量，方式消耗过多的内存。

如有需要可以阅读 Dubbo 中关于线程池的设计（位于 org.apache.dubbo.common.threadpool 包）。

开发

依赖

1
2
3
4
<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
</dependency>

Code

本文代码可以从这里获取。

新增 ThreadUtils.java

用于创建 ExecutorService

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public class ThreadUtils {

    /**
     * 用于对线程分配一个名字，方便排查问题
     * @param format format
     * @return Edith-1
     */
    private static ThreadFactory buildThreadFactory(String format) {
        return new ThreadFactoryBuilder().setNameFormat(format).build();
    }

    public static ExecutorService newCachedThreadPool(int maximumPoolSize) {
        return new ThreadPoolExecutor(THREAD_CORE_POOL_SIZE, maximumPoolSize > 0 ? maximumPoolSize : THREAD_MAXIMUM_POOL_SIZE,
                Long.MAX_VALUE, TimeUnit.MICROSECONDS, new SynchronousQueue<>(), buildThreadFactory(THREAD_NAME), new ThreadPoolExecutor.AbortPolicy());
    }
}

修改 RpcFramework.java

将原代码中 new Thread() 部分替换为线程池。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private static final ExecutorService executorService = ThreadUtils.newCachedThreadPool(0);

public static void export(final Object service, int port) throws Exception {
    Objects.requireNonNull(service, "Service instance is null!");

    if (port <= PORT_MIN || port > PORT_MAX) {
        throw new IllegalArgumentException("Invalid port " + port);
    }
    System.out.println("Export service " + service.getClass().getName() + " on port " + port);
    ServerSocket server = new ServerSocket(port);
    for (; ; ) {
        System.out.println("application is running");
        Socket socket = server.accept();
        executorService.execute(() -> {
            // do something ...
        });
    }
}

经过以上修改，我们在例子中的 HelloServiceImpl 打印一下线程名称

1
2
3
4
5
6
@Override
public String hello(String name) {
    String result = "Hello " + name;
    System.out.println(Thread.currentThread().getName() + " " + result);
    return result;
}

启动 provider，然后用 consumer 调用一次，观察 provider 的控制台输出

1
Edith-0 Hello Peter Parker

Done.