Kotlin 1.9 空安全进阶5个真实项目中的NPE陷阱与解决方案在Android开发领域Kotlin凭借其出色的空安全设计赢得了开发者的青睐。然而随着项目复杂度提升和协程、Flow等高级特性的引入空指针异常NPE仍可能以意想不到的方式出现。本文将深入剖析5个源自真实项目的NPE陷阱并提供基于Kotlin 1.9特性的防御性编程方案。1. 协程作用域取消导致的空引用在异步编程# 1. 概述本文分享Dubbo 的线程池策略。在 《精尽 Dubbo 源码分析 —— 线程池》 一文中我们已经看到了多种线程池和线程队列那么 Dubbo 是如何使用它们的呢本文涉及的类如下图所示2. Dispatchercom.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 namedubbo dispatcherall threadpoolfixed threads100 /dispatcher为all和threadpool为fixed为默认配置。2.1 AllDispatchercom.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 DirectDispatchercom.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 MessageOnlyDispatchercom.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 ExecutionDispatchercom.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 ConnectionOrderedDispatchercom.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 WrappedChannelHandlercom.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 ChannelEventRunnablecom.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 ChannelHandlerscom.alibaba.dubbo.remoting.transport.ChannelHandlersChannelHandler 工厂类。代码如下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. DataStorecom.alibaba.dubbo.common.store.DataStore数据存储接口。代码如下public interface DataStore { /** * 获取数据 * * param component 组件 * param type 类型 * return 数据 */ MapString, 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 SimpleDataStorecom.alibaba.dubbo.common.store.support.SimpleDataStore实现 DataStore 接口简单的数据存储实现类。代码如下public class SimpleDataStore implements DataStore { /** * 数据集合 * key1: 组件 * key2: 类型 * value: 数据 */ private ConcurrentMapString, ConcurrentMapString, Object data new ConcurrentHashMapString, ConcurrentMapString, Object(); Override public MapString, Object get(String component) { ConcurrentMapString, Object value data.get(component); if (value null) { return new HashMapString, Object(); } return new HashMapString, Object(value); } Override public Object get(String component, String type) { if (component null || type null) { return null; } ConcurrentMapString, 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; } // 获得组件的数据集合 ConcurrentMapString, Object componentData data.get(component); if (componentData null) { data.putIfAbsent(component, new ConcurrentHashMapString, 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; } // 获得组件的数据集合 ConcurrentMapString, 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. 彩蛋美滋滋