Kotlin 1.9 空安全进阶：5个真实项目中的NPE陷阱与解决方案-创锋一号

Kotlin 1.9 空安全进阶：5个真实项目中的NPE陷阱与解决方案

在Android开发领域，Kotlin凭借其出色的空安全设计赢得了开发者的青睐。然而，随着项目复杂度提升和协程、Flow等高级特性的引入，空指针异常（NPE）仍可能以意想不到的方式出现。本文将深入剖析5个源自真实项目的NPE陷阱，并提供基于Kotlin 1.9特性的防御性编程方案。

1. 协程作用域取消导致的空引用

在异步编程# 1. 概述

本文分享Dubbo 的线程池策略。在《精尽 Dubbo 源码分析 —— 线程池》一文中，我们已经看到了多种线程池和线程队列，那么 Dubbo 是如何使用它们的呢？

本文涉及的类，如下图所示：

2. Dispatcher

com.alibaba.dubbo.common.threadpool.support.Dispatcher，线程池调度器接口。代码如下：

@SPI(AllDispatcher.NAME) public interface Dispatcher { /** * dispatch the message to threadpool. * * 调度消息到线程池 * * @param handler 处理器 * @param url URL * @return ChannelHandler */ @Adaptive({"dispatcher", "dispather", "channel.handler"}) // 后两个参数，可参见 http://dubbo.apache.org/zh-cn/docs/user/demos/thread-model.html 文档 ChannelHandler dispatch(ChannelHandler handler, URL url); }

@SPI(AllDispatcher.NAME)注解，Dubbo SPI拓展点，默认为"all"。
@Adaptive({"dispatcher", "dispather", "channel.handler"})注解，基于 Dubbo SPI Adaptive 机制，加载对应的线程池调度器实现，使用URL.dispatcher或URL.dispather或URL.channel.handler属性。
为什么会有"dispather"呢？因为，Dispatcher 在 Dubbo 的文档里，称为线程模型，配置方式如下：
```
<dubbo:protocol name="dubbo" dispatcher="all" threadpool="fixed" threads="100" />
```
- dispatcher为"all"和threadpool为"fixed"为默认配置。

2.1 AllDispatcher

com.alibaba.dubbo.common.threadpool.support.all.AllDispatcher，实现 Dispatcher 接口，所有消息都派发到线程池，包括请求，响应，连接事件，断开事件等。代码如下：

public class AllDispatcher implements Dispatcher { public static final String NAME = "all"; @Override public ChannelHandler dispatch(ChannelHandler handler, URL url) { return new AllChannelHandler(handler, url); } }

2.2 DirectDispatcher

com.alibaba.dubbo.common.threadpool.support.direct.DirectDispatcher，实现 Dispatcher 接口，所有消息都不派发到线程池，全部在 IO 线程上直接执行。代码如下：

public class DirectDispatcher implements Dispatcher { public static final String NAME = "direct"; @Override public ChannelHandler dispatch(ChannelHandler handler, URL url) { return new DirectChannelHandler(handler, url); } }

2.3 MessageOnlyDispatcher

com.alibaba.dubbo.common.threadpool.support.messagedispatcher.MessageOnlyDispatcher，实现 Dispatcher 接口，只有请求响应消息派发到线程池，其它连接断开事件，心跳等消息，直接在 IO 线程上执行。代码如下：

public class MessageOnlyDispatcher implements Dispatcher { public static final String NAME = "message"; @Override public ChannelHandler dispatch(ChannelHandler handler, URL url) { return new MessageOnlyChannelHandler(handler, url); } }

2.4 ExecutionDispatcher

com.alibaba.dubbo.common.threadpool.support.execution.ExecutionDispatcher，实现 Dispatcher 接口，只有请求消息派发到线程池，不含响应。其它连接断开事件，心跳等消息，直接在 IO 线程上执行。代码如下：

public class ExecutionDispatcher implements Dispatcher { public static final String NAME = "execution"; @Override public ChannelHandler dispatch(ChannelHandler handler, URL url) { return new ExecutionChannelHandler(handler, url); } }

2.5 ConnectionOrderedDispatcher

com.alibaba.dubbo.common.threadpool.support.connection.ConnectionOrderedDispatcher，实现 Dispatcher 接口，在 IO 线程上，将连接断开事件放入队列，有序逐个执行，其它消息派发到线程池。代码如下：

public class ConnectionOrderedDispatcher implements Dispatcher { public static final String NAME = "connection"; @Override public ChannelHandler dispatch(ChannelHandler handler, URL url) { return new ConnectionOrderedChannelHandler(handler, url); } }

3. ChannelHandler

在《精尽 Dubbo 源码分析 —— NIO 服务器》一文中，我们已经详细分享了 ChannelHandler 的代码实现。在本小节中，我们主要分享Dubbo 给 ChannelHandler 的调度实现。

3.1 WrappedChannelHandler

com.alibaba.dubbo.remoting.transport.dispatcher.WrappedChannelHandler，实现 ChannelHandler 接口，装饰器，将不同的消息派发到线程池，在 IO 线程上执行，进行功能增强。代码如下：

public class WrappedChannelHandler implements ChannelHandler { protected static final Logger logger = LoggerFactory.getLogger(WrappedChannelHandler.class); protected final ChannelHandler handler; protected final URL url; /** * 线程池 */ protected final ExecutorService executor; public WrappedChannelHandler(ChannelHandler handler, URL url) { this.handler = handler; this.url = url; // 创建线程池 executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url); // 添加线程池到 DataStore 中 String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY; if (Constants.CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(Constants.SIDE_KEY))) { componentKey = Constants.CONSUMER_SIDE; } DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension(); dataStore.put(componentKey, Integer.toString(url.getPort()), executor); } @Override public void connected(Channel channel) throws RemotingException { handler.connected(channel); } @Override public void disconnected(Channel channel) throws RemotingException { handler.disconnected(channel); } @Override public void sent(Channel channel, Object message) throws RemotingException { handler.sent(channel, message); } @Override public void received(Channel channel, Object message) throws RemotingException { handler.received(channel, message); } @Override public void caught(Channel channel, Throwable exception) throws RemotingException { handler.caught(channel, exception); } public void close() { try { if (executor != null) { executor.shutdown(); } } catch (Throwable t) { logger.warn("fail to destroy thread pool of server: " + t.getMessage(), t); } } public ExecutorService getExecutor() { return executor; } public ChannelHandler getHandler() { return handler; } public URL getUrl() { return url; } }

executor属性，线程池。在构造方法中，调用ExtensionLoader#getExecutor(url)方法，基于 Dubbo SPI Adaptive 机制，获得对应的线程池实现。代码如下：

// ThreadPool$Adaptive#getExecutor public java.util.concurrent.Executor getExecutor(com.alibaba.dubbo.common.URL arg0) { if (arg0 == null) { throw new IllegalArgumentException("url == null"); } com.alibaba.dubbo.common.URL url = arg0; // 线程池类型 String extName = url.getParameter("threadpool", "fixed"); if (extName == null) { throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.common.threadpool.ThreadPool) name from url(" + url.toString() + ") use keys([threadpool])"); } // 加载线程池实现 com.alibaba.dubbo.common.threadpool.ThreadPool extension = (com.alibaba.dubbo.common.threadpool.ThreadPool) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.common.threadpool.ThreadPool.class).getExtension(extName); return extension.getExecutor(arg0); }

默认情况下，使用"fixed"类型，即 FixedThreadPool 。
在《精尽 Dubbo 源码分析 —— 线程池》一文中，我们已经详细解析。

#close()方法，关闭线程池。

3.1.1 子类

WrappedChannelHandler 的子类，如下图所示：

每个子类，对应不同的 Dispatcher 实现类。以 AllChannelHandler 举例子，代码如下：

public class AllChannelHandler extends WrappedChannelHandler { public AllChannelHandler(ChannelHandler handler, URL url) { super(handler, url); } /** 处理连接事件 */ @Override public void connected(Channel channel) throws RemotingException { // 获得线程池 ExecutorService cexecutor = getExecutorService(); try { // 提交线程池执行 cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED)); } catch (Throwable t) { throw new ExecutionException("connect event", channel, getClass() + " error when process connected event .", t); } } /** 处理断开事件 */ @Override public void disconnected(Channel channel) throws RemotingException { ExecutorService cexecutor = getExecutorService(); try { cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.DISCONNECTED)); } catch (Throwable t) { throw new ExecutionException("disconnect event", channel, getClass() + " error when process disconnected event .", t); } } /** 处理发送事件 */ @Override public void sent(Channel channel, Object message) throws RemotingException { ExecutorService cexecutor = getExecutorService(); try { cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.SENT, message)); } catch (Throwable t) { throw new ExecutionException("send event", channel, getClass() + " error when process sent event .", t); } } /** 处理接收事件 */ @Override public void received(Channel channel, Object message) throws RemotingException { ExecutorService cexecutor = getExecutorService(); try { cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message)); } catch (Throwable t) { //TODO A temporary solution to the problem that the exception information can not be sent to the opposite end after the thread pool is full. Need a refactoring //fix The thread pool is full, refuses to call, does not return, and causes the consumer to wait for time out if (message instanceof Request && t instanceof RejectedExecutionException) { Request request = (Request) message; if (request.isTwoWay()) { String msg = "Server side(" + url.getIp() + "," + url.getPort() + ") threadpool is exhausted ,detail msg:" + t.getMessage(); Response response = new Response(request.getId(), request.getVersion()); response.setStatus(Response.SERVER_THREADPOOL_EXHAUSTED_ERROR); response.setErrorMessage(msg); channel.send(response); return; } } throw new ExecutionException(message, channel, getClass() + " error when process received event .", t); } } /** 处理异常事件 */ @Override public void caught(Channel channel, Throwable exception) throws RemotingException { ExecutorService cexecutor = getExecutorService(); try { cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CAUGHT, exception)); } catch (Throwable t) { throw new ExecutionException("caught event", channel, getClass() + " error when process caught event .", t); } } /** * 获得线程池 * * @return 线程池 */ private ExecutorService getExecutorService() { ExecutorService cexecutor = executor; if (cexecutor == null || cexecutor.isShutdown()) { // 无可用，则创建 cexecutor = SHARED_EXECUTOR; } return cexecutor; } }

处理所有的事件，提交到线程池中，进行执行。
以#received(channel, message)方法举例子，在提交到线程池，被拒绝时，返回Response.SERVER_THREADPOOL_EXHAUSTED_ERROR响应。代码如下：
```
// Response.java /** * 服务端线程池满 * * 200 - 299 为业务错误。 */ public static final byte SERVER_THREADPOOL_EXHAUSTED_ERROR = 101;
```
- 在dubbo-remoting-api模块的com.alibaba.dubbo.remoting.exchange.support.DefaultFuture#returnFromResponse()方法中，会判断返回的是该状态，打印WARN日志。代码如下：
```
if (res.getStatus() == Response.SERVER_THREADPOOL_EXHAUSTED_ERROR) { logger.warn("..." ); }
```
- 在dubbo-rpc-default模块的com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol中，会判断返回的是该状态，打印ERROR日志。代码如下：
```
} else if (res.getStatus() == Response.SERVER_THREADPOOL_EXHAUSTED_ERROR) { logger.error("..." ); }
```
其他 ChannelHandler 实现类，和 AllChannelHandler 类似，笔者就不详细解析。胖友自己查看噢。

3.1.2 ChannelEventRunnable

com.alibaba.dubbo.remoting.transport.dispatcher.ChannelEventRunnable，实现 Runnable 接口，通道事件任务。代码如下：

public class ChannelEventRunnable implements Runnable { /** * 通道 */ private final Channel channel; /** * 通道处理器 */ private final ChannelHandler handler; /** * 通道状态 */ private final ChannelState state; /** * 异常 */ private final Throwable exception; /** * 消息 */ private final Object message; public ChannelEventRunnable(Channel channel, ChannelHandler handler, ChannelState state) { this(channel, handler, state, null); } public ChannelEventRunnable(Channel channel, ChannelHandler handler, ChannelState state, Object message) { this(channel, handler, state, message, null); } public ChannelEventRunnable(Channel channel, ChannelHandler handler, ChannelState state, Throwable t) { this(channel, handler, state, null, t); } public ChannelEventRunnable(Channel channel, ChannelHandler handler, ChannelState state, Object message, Throwable exception) { this.channel = channel; this.handler = handler; this.state = state; this.message = message; this.exception = exception; } @Override public void run() { // 执行 switch (state) { case CONNECTED: try { handler.connected(channel); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e); } break; case DISCONNECTED: try { handler.disconnected(channel); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e); } break; case SENT: try { handler.sent(channel, message); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e); } break; case RECEIVED: try { handler.received(channel, message); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e); } break; case CAUGHT: try { handler.caught(channel, exception); } catch (Exception e) { logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e); } break; default: logger.warn("unknown state: " + state + ", message is " + message); } } }

根据不同的通道状态，调用不同的 ChannelHandler 的方法，处理对应的事件。

3.2 ChannelHandlers

com.alibaba.dubbo.remoting.transport.ChannelHandlers，ChannelHandler 工厂类。代码如下：

public class ChannelHandlers { private static ChannelHandlers INSTANCE = new ChannelHandlers(); protected ChannelHandlers() { } public static ChannelHandler wrap(ChannelHandler handler, URL url) { return ChannelHandlers.getInstance().wrapInternal(handler, url); } protected static ChannelHandlers getInstance() { return INSTANCE; } static void setTestingChannelHandlers(ChannelHandlers instance) { INSTANCE = instance; } protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) { // 创建 Dispatcher 实现类对象 return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class) .getAdaptiveExtension().dispatch(handler, url))); } }

调用Dispatcher#dispatch(handler, url)方法，创建对应的 ChannelHandler 实现类。例如，AllDispatcher 会创建 AllChannelHandler 。
在创建完对应的 ChannelHandler 实现类后，再包装 MultiMessageHandler + HeartbeatHandler 处理器。

4. DataStore

com.alibaba.dubbo.common.store.DataStore，数据存储接口。代码如下：

public interface DataStore { /** * 获取数据 * * @param component 组件 * @param type 类型 * @return 数据 */ Map<String, Object> get(String component); /** * 获取数据 * * @param component 组件 * @param type 类型 * @return 数据 */ Object get(String component, String type); /** * 存储数据 * * @param component 组件 * @param type 类型 * @param value 数据 * @return 数据 */ Object put(String component, String type, Object value); /** * 移除数据 * * @param component 组件 * @param type 类型 * @return 数据 */ Object remove(String component, String type); }

4.1 SimpleDataStore

com.alibaba.dubbo.common.store.support.SimpleDataStore，实现 DataStore 接口，简单的数据存储实现类。代码如下：

public class SimpleDataStore implements DataStore { /** * 数据集合 * key1: 组件 * key2: 类型 * value: 数据 */ private ConcurrentMap<String, ConcurrentMap<String, Object>> data = new ConcurrentHashMap<String, ConcurrentMap<String, Object>>(); @Override public Map<String, Object> get(String component) { ConcurrentMap<String, Object> value = data.get(component); if (value == null) { return new HashMap<String, Object>(); } return new HashMap<String, Object>(value); } @Override public Object get(String component, String type) { if (component == null || type == null) { return null; } ConcurrentMap<String, Object> componentData = data.get(component); if (componentData == null) { return null; } return componentData.get(type); } @Override public Object put(String component, String type, Object value) { if (component == null || type == null) { return null; } // 获得组件的数据集合 ConcurrentMap<String, Object> componentData = data.get(component); if (componentData == null) { data.putIfAbsent(component, new ConcurrentHashMap<String, Object>()); componentData = data.get(component); } // 存储数据 return componentData.put(type, value); } @Override public Object remove(String component, String type) { if (component == null || type == null) { return null; } // 获得组件的数据集合 ConcurrentMap<String, Object> componentData = data.get(component); if (componentData == null) { return null; } // 移除数据 return componentData.remove(type); } }

代码比较简单，胖友自己看注释。

4.1.1 初始化

在WrappedChannelHandler的构造方法中，会添加线程池到 DataStore 中。代码如下：

// WrappedChannelHandler.java executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url); // 添加线程池到 DataStore 中 String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY; if (Constants.CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(Constants.SIDE_KEY))) { componentKey = Constants.CONSUMER_SIDE; } DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension(); dataStore.put(componentKey, Integer.toString(url.getPort()), executor);

在"dubbo"组件下，添加线程池。

4.1.2 使用

在dubbo-admin项目中，使用 SimpleDataStore 存储的线程池，如下：

5. 日志增强

在dubbo-remoting-api模块的com.alibaba.dubbo.remoting.transport.netty.NettyHelper中，有 Netty 的日志工厂，可以设置 Dubbo Logger 使用 Netty 的 Logger 打印日志。代码如下：

public final class NettyHelper { public static void setNettyLoggerFactory() { InternalLoggerFactory factory = InternalLoggerFactory.getDefaultFactory(); if (factory == null || !(factory instanceof DubboLoggerFactory)) { InternalLoggerFactory.setDefaultFactory(new DubboLoggerFactory()); } } }

在com.alibaba.dubbo.remoting.transport.netty.NettyTransporter#bind(URL, ChannelHandler)方法中，会调用该方法，设置 Dubbo Logger 使用 Netty 的 Logger 打印日志。这样，Netty 的日志会和 Dubbo 的日志体系一致。

666. 彩蛋

美滋滋！

企业官网建设流程全解析