前言
设计模式是软件设计中常见问题的典型解决方案,你可以通过对其进行定制来解决代码中的特定设计问题。
关于设计模式,网上有很多讲解。但大部分都是Demo示例,看完有可能还是不知道怎么用。
本文笔者将从设计模式入手,看一看在优秀的Java框架/中间件产品中,不同的设计模式应用场景在哪里。
一、单例模式
单例模式是Java中最简单的设计模式之一,它提供了一种创建对象的最佳方式。这种模式涉及到一个单一的类,该类负责创建自己的对象,同时确保只有单个对象被创建。这个类提供了一种访问其唯一的对象的方式,可以直接访问,不需要实例化该类的对象。
单例模式虽然很简单,但它的花样一点都不少,我们一一来看。
1、饿汉式
饿汉式,顾名思义,就是我很饿,迫不及待。不管有没有人用,我先创建了再说。
比如在Dubbo中的这段代码,创建一个配置管理器。
public class ConfigManager { private static final ConfigManager configManager = new ConfigManager(); private ConfigManager() {} public static ConfigManager getInstance() { return configManager; } }
又或者在RocketMQ中,创建一个MQ客户端实例的时候。
public class MQClientManager { private static MQClientManager instance = new MQClientManager(); private MQClientManager() {} public static MQClientManager getInstance() { return instance; } }
2、懒汉式
懒汉式是对应饿汉式而言的。它旨在第一次调用才初始化,避免内存浪费。但为了线程安全和性能,一般都会使用双重检查锁的方式来创建。
我们来看Seata框架中,通过这种方式来创建一个配置类。
public class ConfigurationFactory{ private static volatile Configuration CONFIG_INSTANCE = null; public static Configuration getInstance() { if (CONFIG_INSTANCE == null) { synchronized (Configuration.class) { if (CONFIG_INSTANCE == null) { CONFIG_INSTANCE = buildConfiguration(); } } } return CONFIG_INSTANCE; } }
3、静态内部类
可以看到,通过双重检查锁的方式来创建单例对象,还是比较复杂的。又是加锁,又是判断两次,还需要加volatile修饰的。
使用静态内部类的方式,可以达到双重检查锁相同的功效,但实现上简单了。
在Seata框架中,创建RM事件处理程序器的时候,就使用了静态内部类的方式来创建单例对象。
public class DefaultRMHandler extends AbstractRMHandler{ protected DefaultRMHandler() { initRMHandlers(); } private static class SingletonHolder { private static AbstractRMHandler INSTANCE = new DefaultRMHandler(); } public static AbstractRMHandler get() { return DefaultRMHandler.SingletonHolder.INSTANCE; } }
还有可以通过枚举的方式来创建单例对象,但这种方式并没有被广泛采用,至少笔者在常见的开源框架中没见过,所以就不再列举。
有人说,饿汉式的单例模式不好,不能做到延迟加载,浪费内存。但笔者认为似乎过于吹毛求疵,事实上很多开源框架中,用的最多的就是这种方式。
如果明确希望实现懒加载效果时,可以考虑用静态内部类的方式;如果还有其他特殊的需求,比如创建对象的过程比较繁琐,可以用双重检查锁的方式。
二、工厂模式
工厂模式是Java中最常用的设计模式之一。这种类型的设计模式属于创建型模式,它提供了一种创建对象的最佳方式。
简单来说,在工厂模式中,就是代替new实例化具体类的一种模式。
1、简单工厂
简单工厂,的确比较简单,它的作用就是把对象的创建放到一个工厂类中,通过参数来创建不同的对象。
在分布式事务框架Seata中,如果发生异常,则需要进行二阶段回滚。
它的过程是,通过事务id找到undoLog记录,然后解析里面的数据生成SQL,将一阶段执行的SQL给撤销掉。
问题是SQL的种类包含了比如INSERT、UPDATE、DELETE,所以它们反解析的过程也不一样,就需要不同的执行器去解析。
在Seata中,有一个抽象的撤销执行器,可以生成一条SQL。
public abstract class AbstractUndoExecutor{ //生成撤销SQL protected abstract String buildUndoSQL(); }
然后有一个获取撤销执行器的工厂,根据SQL的类型,创建不同类型的执行器并返回。
public class UndoExecutorFactory { public static AbstractUndoExecutor getUndoExecutor(String dbType, SQLUndoLog sqlUndoLog) { switch (sqlUndoLog.getSqlType()) { case INSERT: return new MySQLUndoInsertExecutor(sqlUndoLog); case UPDATE: return new MySQLUndoUpdateExecutor(sqlUndoLog); case DELETE: return new MySQLUndoDeleteExecutor(sqlUndoLog); default: throw new ShouldNeverHappenException(); } } }
使用的时候,直接通过工厂类获取执行器。
AbstractUndoExecutor undoExecutor = UndoExecutorFactory.getUndoExecutor(dataSourceProxy.getDbType(),sqlUndoLog); undoExecutor.executeOn(conn);
简单工厂模式的优点,想必各位都能领会,我们不再赘述。但它还有个小小的缺点:
一旦有了新的实现类,就需要修改工厂实现,有可能造成工厂逻辑过于复杂,不利于系统的扩展和维护。
2、工厂方法
工厂方法模式解决了上面那个问题。它可以创建一个工厂接口和多个工厂实现类,这样如果增加新的功能,只需要添加新的工厂类就可以,不需要修改之前的代码。
另外,工厂方法模式还可以和模板方法模式结合一起,将他们共同的基础逻辑抽取到父类中,其它的交给子类去实现。
在Dubbo中,有一个关于缓存的设计完美的体现了工厂方法模式+模板方法模式。
首先,有一个缓存的接口,它提供了设置缓存和获取缓存两个方法。
public interface Cache { void put(Object key, Object value); Object get(Object key); }
然后呢,还有一个缓存工厂,它返回一个缓存的实现。
public interface CacheFactory { Cache getCache(URL url, Invocation invocation); }
由于结合了模板方法模式,所以Dubbo又搞了个抽象的缓存工厂类,它实现了缓存工厂的接口。
public abstract class AbstractCacheFactory implements CacheFactory { //具体的缓存实现类 private final ConcurrentMap<String, Cache> caches = new ConcurrentHashMap<String, Cache>(); @Override public Cache getCache(URL url, Invocation invocation) { url = url.addParameter(Constants.METHOD_KEY, invocation.getMethodName()); String key = url.toFullString(); Cache cache = caches.get(key); if (cache == null) { //创建缓存实现类,交给子类实现 caches.put(key, createCache(url)); cache = caches.get(key); } return cache; } //抽象方法,交给子类实现 protected abstract Cache createCache(URL url); }
在这里,公共的逻辑就是通过getCahce()创建缓存实现类,那具体创建什么样的缓存实现类,就由子类去决定。
所以,每个子类都是一个个具体的缓存工厂类,比如包括:
ExpiringCacheFactory、JCacheFactory、LruCacheFactory、ThreadLocalCacheFactory。 这些工厂类,只有一个方法,就是创建具体的缓存实现类。 public class ThreadLocalCacheFactory extends AbstractCacheFactory { @Override protected Cache createCache(URL url) { return new ThreadLocalCache(url); } }
这里的ThreadLocalCache就是具体的缓存实现类,比如它是通过ThreadLocal来实现缓存功能。
public class ThreadLocalCache implements Cache { private final ThreadLocal<Map<Object, Object>> store; public ThreadLocalCache(URL url) { this.store = new ThreadLocal<Map<Object, Object>>() { @Override protected Map<Object, Object> initialValue() { return new HashMap<Object, Object>(); } }; } @Override public void put(Object key, Object value) { store.get().put(key, value); } @Override public Object get(Object key) { return store.get().get(key); } }
那在客户端使用的时候,还是通过工厂来获取缓存对象。
public static void main(String[] args) { URL url = URL.valueOf("http://localhost:8080/cache=jacache&.cache.write.expire=1"); Invocation invocation = new RpcInvocation(); CacheFactory cacheFactory = new ThreadLocalCacheFactory(); Cache cache = cacheFactory.getCache(url, invocation); cache.put("java","java"); System.out.println(cache.get("java")); }
这样做的好处有两点。
第一,如果增加新的缓存实现,只要添加一个新的缓存工厂类就可以,别的都无需改动。
第二,通过模板方法模式,封装不变部分,扩展可变部分。提取公共代码,便于维护。
另外,在Dubbo中,注册中心的获取也是通过工厂方法来实现的。
3、抽象工厂
抽象工厂模式,它能创建一系列相关的对象,而无需指定其具体类。
工厂方法模式和抽象工厂模式,它们之间最大的区别在于:
- 工厂方法模式只有一个抽象产品类,具体工厂类只能创建一个具体产品类的实例;
- 抽象工厂模式有多个抽象产品类,具体工厂类可以创建多个具体产品类的实例。
我们拿上面缓存的例子来继续往下说。
如果我们现在有一个数据访问程序,需要同时操作缓存和数据库,那就需要多个抽象产品和多个具体产品实现。
缓存相关的产品类都已经有了,我们接着来创建数据库相关的产品实现。
首先,有一个数据库接口,它是抽象产品类。
public interface DataBase { void insert(Object tableName, Object record); Object select(Object tableName); }
然后,我们创建两个具体产品类MysqlDataBase和OracleDataBase。
public class MysqlDataBase implements DataBase{ Map<Object,Object> mysqlDb = new HashMap<>(); @Override public void insert(Object tableName, Object record) { mysqlDb.put(tableName,record); } @Override public Object select(Object tableName) { return mysqlDb.get(tableName); } } public class OracleDataBase implements DataBase { Map<Object,Object> oracleDb = new HashMap<>(); @Override public void insert(Object tableName, Object record) { oracleDb.put(tableName,record); } @Override public Object select(Object tableName) { return oracleDb.get(tableName); } }
其次,创建抽象的工厂类,它可以返回一个缓存对象和数据库对象。
public interface DataAccessFactory { Cache getCache(URL url); DataBase getDb(); }
最后是具体的工厂类,可以根据实际的需求,任意组合每一个具体的产品。
比如我们需要一个基于ThreadLocal的缓存实现和基于MySQL的数据库对象。
public class DataAccessFactory1 implements DataAccessFactory { @Override public Cache getCache(URL url) { return new ThreadLocalCache(url); } @Override public DataBase getDb() { return new MysqlDataBase(); } }
如果需要一个基于Lru的缓存实现和基于Oracle的数据库对象。
public class DataAccessFactory2 implements DataAccessFactory { @Override public Cache getCache(URL url) { return new LruCache(url); } @Override public DataBase getDb() { return new OracleDataBase(); } }
可以看到,抽象工厂模式隔离了具体类的生成,使得客户并不需要知道什么被创建。由于这种隔离,更换一个具体工厂就变得相对容易,所有的具体工厂都实现了抽象工厂中定义的那些公共接口,因此只需改变具体工厂的实例,就可以在某种程度上改变整个软件系统的行为。
三、模板方法模式
在模板模式中,一个抽象类公开定义了执行它的方法的方式/模板。它的子类可以按需要重写方法实现,但调用将以抽象类中定义的方式进行。
简单来说,有多个子类共有的方法,且逻辑相同,可以考虑作为模板方法。
在上面Dubbo缓存的例子中,我们已经看到了模板方法模式的应用。但那个是和工厂方法模式结合在一块的,我们再单独找找模板方法模式的应用。
我们知道,当我们的Dubbo应用出现多个服务提供者时,服务消费者需要通过负载均衡算法,选择其中一个服务来进行调用。
首先,有一个LoadBalance接口,返回一个Invoker。
public interface LoadBalance { <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException; }
然后定义一个抽象类,AbstractLoadBalance,实现LoadBalance接口。
public abstract class AbstractLoadBalance implements LoadBalance { @Override public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) { if (invokers == null || invokers.isEmpty()) { return null; } if (invokers.size() == 1) { return invokers.get(0); } return doSelect(invokers, url, invocation); } //抽象方法,由子类选择一个Invoker protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation); }
这里的公共逻辑就是两个判断,判断invokers集合是否为空或者是否只有一个实例。如果是的话,就无需调用子类,直接返回就好了。
具体的负载均衡实现有四个:
- 基于权重随机算法的 RandomLoadBalance
- 基于最少活跃调用数算法的 LeastActiveLoadBalance
- 基于 hash 一致性的 ConsistentHashLoadBalance
- 基于加权轮询算法的 RoundRobinLoadBalance
public class RandomLoadBalance extends AbstractLoadBalance { @Override protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) { //省略逻辑.... return invokers.get(ThreadLocalRandom.current().nextInt(length)); } }
它们根据不同的算法实现,来返回一个具体的Invoker对象。
四、构造器模式
构造器模式使用多个简单的对象一步一步构建成一个复杂的对象。这种类型的设计模式属于创建型模式,它提供了一种创建对象的最佳方式。
这种模式,常见于在构建一个复杂的对象,里面可能包含一些业务逻辑,比如检查,属性转换等。如果都在客户端手动去设置,那么会产生大量的冗余代码。那么这时候,我们就可以考虑使用构造器模式。
比如,在MyBatis中,MappedStatement的创建过程就使用了构造器模式。
我们知道,XML文件中的每一个SQL标签就要生成一个MappedStatement对象,它里面包含很多个属性,我们要构造的对象也是它。
public final class MappedStatement { private String resource; private Configuration configuration; private String id; private SqlSource sqlSource; private ParameterMap parameterMap; private List<ResultMap> resultMaps; //.....省略大部分属性 }
然后有一个内部类Builder,它负责完成MappedStatement对象的构造。
首先,这个Builder类,通过默认的构造函数,先完成对MappedStatement对象,部分的构造。
public static class Builder { private MappedStatement mappedStatement = new MappedStatement(); public Builder(Configuration configuration, String id, SqlSource sqlSource, SqlCommandType sqlCommandType) { mappedStatement.configuration = configuration; mappedStatement.id = id; mappedStatement.sqlSource = sqlSource; mappedStatement.statementType = StatementType.PREPARED; mappedStatement.resultSetType = ResultSetType.DEFAULT; //.....省略大部分过程 } }
然后,通过一系列方法,可以设置特定的属性,并返回这个Builder类,这里的方法适合处理一些业务逻辑。
public static class Builder { public Builder parameterMap(ParameterMap parameterMap) { mappedStatement.parameterMap = parameterMap; return this; } public Builder resultMaps(List<ResultMap> resultMaps) { mappedStatement.resultMaps = resultMaps; for (ResultMap resultMap : resultMaps) { mappedStatement.hasNestedResultMaps = mappedStatement.hasNestedResultMaps || resultMap.hasNestedResultMaps(); } return this; } public Builder statementType(StatementType statementType) { mappedStatement.statementType = statementType; return this; } public Builder resultSetType(ResultSetType resultSetType) { mappedStatement.resultSetType = resultSetType == null ? ResultSetType.DEFAULT : resultSetType; return this; } }
最后呢,就是提供一个build方法,返回构建完成的对象就好了。
public MappedStatement build() { assert mappedStatement.configuration != null; assert mappedStatement.id != null; assert mappedStatement.sqlSource != null; assert mappedStatement.lang != null; mappedStatement.resultMaps = Collections.unmodifiableList(mappedStatement.resultMaps); return mappedStatement; }
在客户端使用的时候,先创建一个Builder,然后链式的调用一堆方法,最后再调用一次build()方法,我们需要的对象就有了,这就是构造器模式的应用。
MappedStatement.Builder statementBuilder = new MappedStatement.Builder(configuration, id, sqlSource, sqlCommandType) .resource(resource) .fetchSize(fetchSize) .timeout(timeout) .statementType(statementType) .keyGenerator(keyGenerator) .keyProperty(keyProperty) .keyColumn(keyColumn) .databaseId(databaseId) .lang(lang) .resultOrdered(resultOrdered) .resultSets(resultSets) .resultMaps(getStatementResultMaps(resultMap, resultType, id)) .resultSetType(resultSetType) .flushCacheRequired(valueOrDefault(flushCache, !isSelect)) .useCache(valueOrDefault(useCache, isSelect)) .cache(currentCache); ParameterMap statementParameterMap = getStatementParameterMap(parameterMap, parameterType, id); MappedStatement statement = statementBuilder.build(); configuration.addMappedStatement(statement); return statement;
五、适配器模式
适配器模式是作为两个不兼容的接口之间的桥梁。这种类型的设计模式属于结构型模式,它结合了两个独立接口的功能。
适配器模式一般用于屏蔽业务逻辑与第三方服务的交互,或者是新老接口之间的差异。
我们知道,在Dubbo中,所有的数据都是通过Netty来负责传输的,然后这就涉及了消息编解码的问题。
所以,首先它有一个编解码器的接口,负责编码和解码。
@SPI public interface Codec2 { @Adaptive({Constants.CODEC_KEY}) void encode(Channel channel, ChannelBuffer buffer, Object message) throws IOException; @Adaptive({Constants.CODEC_KEY}) Object decode(Channel channel, ChannelBuffer buffer) throws IOException; enum DecodeResult { NEED_MORE_INPUT, SKIP_SOME_INPUT } }
然后,有几个实现类,比如DubboCountCodec、DubboCodec、ExchangeCodec等。
但是,当我们打开这些类的时候,就会发现,他们都是Dubbo中普通的类,只是实现了Codec2接口,其实不能直接作用于Netty编解码。
这是因为,Netty编解码需要实现ChannelHandler接口,这样才会被声明成Netty的处理组件。比如像MessageToByteEncoder、ByteToMessageDecoder那样。
鉴于此,Dubbo搞了一个适配器,专门来适配编解码器接口。
final public class NettyCodecAdapter { private final ChannelHandler encoder = new InternalEncoder(); private final ChannelHandler decoder = new InternalDecoder(); private final Codec2 codec; private final URL url; private final org.apache.dubbo.remoting.ChannelHandler handler; public NettyCodecAdapter(Codec2 codec, URL url, org.apache.dubbo.remoting.ChannelHandler handler) { this.codec = codec; this.url = url; this.handler = handler; } public ChannelHandler getEncoder() { return encoder; } public ChannelHandler getDecoder() { return decoder; } private class InternalEncoder extends MessageToByteEncoder { @Override protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception { org.apache.dubbo.remoting.buffer.ChannelBuffer buffer = new NettyBackedChannelBuffer(out); Channel ch = ctx.channel(); NettyChannel channel = NettyChannel.getOrAddChannel(ch, url, handler); codec.encode(channel, buffer, msg); } } private class InternalDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf input, List<Object> out) throws Exception { ChannelBuffer message = new NettyBackedChannelBuffer(input); NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler); //解码对象 codec.decode(channel, message); //省略部分代码... } } }
上面的代码中,我们看到,NettyCodecAdapter类适配的是Codec2接口,通过构造函数传递实现类,然后定义了内部的编码器实现和解码器实现,同时它们都是ChannelHandler。
这样的话,在内部类里面的编码和解码逻辑,真正调用的还是Codec2接口。
最后我们再来看看,该适配器的调用方式。
//通过SPI方式获取编解码器的实现类,比如这里是DubboCountCodec Codec2 codec = ExtensionLoader.getExtensionLoader(Codec2.class).getExtension("dubbo"); URL url = new URL("dubbo", "localhost", 22226); //创建适配器 NettyCodecAdapter adapter = new NettyCodecAdapter(codec, url, NettyClient.this); //向ChannelPipeline中添加编解码处理器 ch.pipeline() .addLast("decoder", adapter.getDecoder()) .addLast("encoder", adapter.getEncoder())
以上,就是Dubbo中关于编解码器对于适配器模式的应用。
六、责任链模式
责任链模式为请求创建了一个接收者对象的链。允许你将请求沿着处理者链进行发送。收到请求后,每个处理者均可对请求进行处理,或将其传递给链上的下个处理者。
我们来看一个Netty中的例子。我们知道,在Netty中服务端处理消息,就要添加一个或多个ChannelHandler。那么,承载这些ChannelHandler的就是ChannelPipeline,它的实现过程就体现了责任链模式的应用。
ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() { protected void initChannel(NioSocketChannel channel) { channel.pipeline() .addLast(new ChannelHandler1()) .addLast(new ChannelHandler2()) .addLast(new ChannelHandler3()); } });
需要知道的是,在Netty整个框架里面,一条连接对应着一个Channel,每一个新创建的Channel都将会被分配一个新的ChannelPipeline。
ChannelPipeline里面保存的是ChannelHandlerContext对象,它是Channel相关的上下文对象,里面包着我们定义的处理器ChannelHandler。
根据事件的起源,IO事件将会被ChannelInboundHandler或者ChannelOutboundHandler处理。随后,通过调用ChannelHandlerContext实现,它将被转发给同一超类型的下一个ChannelHandler。
1、ChannelHandler
首先,我们来看责任处理器接口,Netty中的ChannelHandler,它充当了所有处理入站和出站数据的应用程序逻辑的容器。
public interface ChannelHandler { //当把 ChannelHandler 添加到 ChannelPipeline 中时被调用 void handlerAdded(ChannelHandlerContext ctx) throws Exception; //当从 ChannelPipeline 中移除 ChannelHandler 时被调用 void handlerRemoved(ChannelHandlerContext ctx) throws Exception; //当处理过程中在 ChannelPipeline 中有错误产生时被调用 void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception; }
然后Netty定义了下面两个重要的ChannelHandler子接口:
1、ChannelInboundHandler,处理入站数据以及各种状态变化;
public interface ChannelInboundHandler extends ChannelHandler { //当 Channel 已经注册到它的 EventLoop 并且能够处理 I/O 时被调用 void channelRegistered(ChannelHandlerContext ctx) throws Exception; //当 Channel 从它的 EventLoop 注销并且无法处理任何 I/O 时被调用 void channelUnregistered(ChannelHandlerContext ctx) throws Exception; //当 Channel 处于活动状态时被调用;Channel 已经连接/绑定并且已经就绪 void channelActive(ChannelHandlerContext ctx) throws Exception; //当 Channel 离开活动状态并且不再连接它的远程节点时被调用 void channelInactive(ChannelHandlerContext ctx) throws Exception; 当从 Channel 读取数据时被调用 void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception; //当 Channel上的一个读操作完成时被调用 void channelReadComplete(ChannelHandlerContext ctx) throws Exception; void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception; void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception; void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception; }
2、ChannelOutboundHandler,处理出站数据并且允许拦截所有的操作;
public interface ChannelOutboundHandler extends ChannelHandler { //当请求将 Channel 绑定到本地地址时被调用 void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception; //当请求将 Channel 连接到远程节点时被调用 void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress,SocketAddress localAddress, ChannelPromise promise) throws Exception; //当请求将 Channel 从远程节点断开时被调用 void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception; //当请求关闭 Channel 时被调用 void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception; //当请求将 Channel 从它的 EventLoop 注销时被调用 void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception; //当请求从 Channel 读取更多的数据时被调用 void read(ChannelHandlerContext ctx) throws Exception; //当请求通过 Channel 将数据写到远程节点时被调用 void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception; //当请求通过 Channel 将入队数据冲刷到远程节点时被调用 void flush(ChannelHandlerContext ctx) throws Exception; }
2、ChannelPipeline
既然叫做责任链模式,那就需要有一个“链”,在Netty中就是ChannelPipeline。
ChannelPipeline提供了ChannelHandler链的容器,并定义了用于在该链上传播入站和出站事件流的方法,另外它还具有添加删除责任处理器接口的功能。
public interface ChannelPipeline{ ChannelPipeline addFirst(String name, ChannelHandler handler); ChannelPipeline addLast(String name, ChannelHandler handler); ChannelPipeline addBefore(String baseName, String name, ChannelHandler handler); ChannelPipeline addAfter(String baseName, String name, ChannelHandler handler); ChannelPipeline remove(ChannelHandler handler); ChannelHandler replace(String oldName, String newName, ChannelHandler newHandler); @Override ChannelPipeline fireChannelRegistered(); @Override ChannelPipeline fireChannelActive(); @Override ChannelPipeline fireExceptionCaught(Throwable cause); @Override ChannelPipeline fireUserEventTriggered(Object event); @Override ChannelPipeline fireChannelRead(Object msg); @Override ChannelPipeline flush(); //省略部分方法..... }
然后我们看它的实现,默认有两个节点,头结点和尾结点。并在构造函数中,使它们首尾相连。这就是标准的链式结构。
public class DefaultChannelPipeline implements ChannelPipeline { final AbstractChannelHandlerContext head; final AbstractChannelHandlerContext tail; private final Channel channel; protected DefaultChannelPipeline(Channel channel) { this.channel = ObjectUtil.checkNotNull(channel, "channel"); tail = new TailContext(this); head = new HeadContext(this); head.next = tail; tail.prev = head; } }
当有新的ChannelHandler被添加时,则将其封装为ChannelHandlerContext对象,然后插入到链表中。
private void addLast0(AbstractChannelHandlerContext newCtx) { AbstractChannelHandlerContext prev = tail.prev; newCtx.prev = prev; newCtx.next = tail; prev.next = newCtx; tail.prev = newCtx; }
3、ChannelHandlerContext
ChannelHandlerContext代表了ChannelHandler和ChannelPipeline之间的关联,每当有ChannelHandler添加到ChannelPipeline中时,都会创建ChannelHandlerContext。
ChannelHandlerContext的主要功能是管理它所关联的ChannelHandler和在同一个ChannelPipeline中的其他ChannelHandler之间的交互。
public interface ChannelHandlerContext{ Channel channel(); EventExecutor executor(); ChannelHandler handler(); ChannelPipeline pipeline(); @Override ChannelHandlerContext fireChannelRegistered(); @Override ChannelHandlerContext fireChannelUnregistered(); @Override ChannelHandlerContext fireChannelActive(); @Override ChannelHandlerContext fireChannelRead(Object msg); @Override ChannelHandlerContext read(); @Override ChannelHandlerContext flush(); //省略部分方法…… }
ChannelHandlerContext负责在链上传播责任处理器接口的事件。
它有两个重要的方法,查找Inbound类型和Outbound类型的处理器。
值得注意的是,如果一个入站事件被触发,它将被从ChannelPipeline的头部开始一直被传播到ChannelPipeline的尾端;一个出站事件将从ChannelPipeline的最右边开始,然后向左传播。
abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint { volatile AbstractChannelHandlerContext next; volatile AbstractChannelHandlerContext prev; //查找下一个Inbound类型的处理器,左 > 右 private AbstractChannelHandlerContext findContextInbound(int mask) { AbstractChannelHandlerContext ctx = this; EventExecutor currentExecutor = executor(); do { ctx = ctx.next; } while (skipContext(ctx, currentExecutor, mask, MASK_ONLY_INBOUND)); return ctx; } //查找下一个Outbound类型的处理器,右 > 左 private AbstractChannelHandlerContext findContextOutbound(int mask) { AbstractChannelHandlerContext ctx = this; EventExecutor currentExecutor = executor(); do { ctx = ctx.prev; } while (skipContext(ctx, currentExecutor, mask, MASK_ONLY_OUTBOUND)); return ctx; } }
4、处理流程
当我们向服务端发送消息的时候,将会触发read方法。
public abstract class AbstractNioByteChannel extends AbstractNioChannel { public final void read() { //从Channel中获取对应的ChannelPipeline final ChannelPipeline pipeline = pipeline(); //数据载体 ByteBuf byteBuf = allocHandle.allocate(allocator); //传递数据 pipeline.fireChannelRead(byteBuf); } }
上面的代码中,就会调用到ChannelPipeline,它会从Head节点开始,根据上下文对象依次调用处理器。
public class DefaultChannelPipeline implements ChannelPipeline { public final ChannelPipeline fireChannelRead(Object msg) { AbstractChannelHandlerContext.invokeChannelRead(head, msg); return this; } }
因为第一个节点是默认的头结点HeadContext,所以它是从ChannelHandlerContext开始的。
abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint { //找到下一个ChannelHandler并执行 public ChannelHandlerContext fireChannelRead(final Object msg) { invokeChannelRead(findContextInbound(MASK_CHANNEL_READ), msg); return this; } }
然后在我们自定义的ChannelHandler中,就会被调用到。
public class ChannelHandler1 extends ChannelInboundHandlerAdapter { public void channelRead(ChannelHandlerContext ctx, Object msg){ System.out.println("ChannelHandler1:"+msg); ctx.fireChannelRead(msg); } }
如果消息有多个ChannelHandler,你可以自由选择是否继续往下传递请求。
比如,如果你认为消息已经被处理且不应该继续往下调用,把上面的ctx.fireChannelRead(msg);注释掉就终止了整个责任链。
七、策略模式
该模式定义了一系列算法,并将每个算法封装起来,使它们可以相互替换,且算法的变化不会影响使用算法的客户。
策略模式是一个很常见,而且也很好用的设计模式,如果你的业务代码中有大量的if...else,那么就可以考虑是否可以使用策略模式改造一下。
RocketMQ我们大家都熟悉,是一款优秀的分布式消息中间件。消息中间件,简单来说,就是客户端发送一条消息,服务端存储起来并提供给消费者去消费。
请求消息的类型多种多样,处理过程肯定也不一样,每次都判断一下再处理就落了下乘。在RocketMQ里,它会把所有处理器注册起来,然后根据请求消息的code,让对应的处理器处理请求,这就是策略模式的应用。
首先,它们需要实现同一个接口,在这里就是请求处理器接口。
public interface NettyRequestProcessor { RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)throws Exception; boolean rejectRequest(); }
这个接口只做一件事,就是处理来自客户端的请求。不同类型的请求封装成不同的RemotingCommand对象。
RocketMQ大概有90多种请求类型,都在RequestCode里以code来区分。
然后,定义一系列策略类。我们来看几个。
//默认的消息处理器 public class DefaultRequestProcessor implements NettyRequestProcessor {} //发送消息的处理器 public class SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor {} //拉取消息的处理器 public class PullMessageProcessor implements NettyRequestProcessor {} //查询消息的处理器 public class QueryMessageProcessor implements NettyRequestProcessor {} //消费者端管理的处理器 public class ConsumerManageProcessor implements NettyRequestProcessor {}
接着,将这些策略类封装起来。在RocketMQ中,在启动Broker服务器的时候,注册这些处理器。
public class BrokerController { public void registerProcessor() { SendMessageProcessor sendProcessor = new SendMessageProcessor(this); this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor); this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor); this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE,this.pullMessageProcessor,this.pullMessageExecutor); this.remotingServer.registerProcessor(RequestCode.SEND_REPLY_MESSAGE, replyMessageProcessor, replyMessageExecutor); this.remotingServer.registerProcessor(RequestCode.SEND_REPLY_MESSAGE_V2, replyMessageProcessor, replyMessageExecutor); NettyRequestProcessor queryProcessor = new QueryMessageProcessor(this); this.remotingServer.registerProcessor(RequestCode.QUERY_MESSAGE, queryProcessor, this.queryMessageExecutor); this.remotingServer.registerProcessor(RequestCode.VIEW_MESSAGE_BY_ID, queryProcessor, this.queryMessageExecutor); ClientManageProcessor clientProcessor = new ClientManageProcessor(this); this.remotingServer.registerProcessor(RequestCode.HEART_BEAT, clientProcessor, this.heartbeatExecutor); this.remotingServer.registerProcessor(RequestCode.UNREGISTER_CLIENT, clientProcessor, this.clientManageExecutor); //省略部分注册过程..... } }
最后,在Netty接收到客户端的请求之后,就会根据消息的类型,找到对应的策略类,去处理消息。
public abstract class NettyRemotingAbstract { public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) { //根据请求类型找到对应的策略类 final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode()); //如果没有找到就使用默认的 final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched; //执行策略 final RemotingCommand response = pair.getObject1().processRequest(ctx, cmd); //省略大部分代码...... } }
如果有了新的请求消息类型,RocketMQ也无需修改业务代码,新增策略类并将其注册进来就好了。
八、代理模式
代理模式,为其他对象提供一种代理以控制对这个对象的访问。
在一些开源框架或中间件产品中,代理模式会非常常见。我们使用的时候越简便,框架在背后帮我们做的事就可能越复杂。这里面往往都体现着代理模式的应用,颇有移花接木的味道。
1、Dubbo
Dubbo作为一个RPC框架,其中有一个很重要的功能就是:
提供高性能的基于代理的远程调用能力,服务以接口为粒度,为开发者屏蔽远程调用底层细节。
这里我们关注两个重点:
- 面向接口代理;
- 屏蔽调用底层细节。
比如我们有一个库存服务,它提供一个扣减库存的接口。
public interface StorageDubboService { int decreaseStorage(StorageDTO storage); }
在别的服务里,需要扣减库存的时候,就会通过Dubbo引用这个接口,也比较简单。
@Reference StorageDubboService storageDubboService;
我们使用起来很简单,可StorageDubboService只是一个普通的服务类,并不具备远程调用的能力。
Dubbo就是给这些服务类,创建了代理类。通过ReferenceBean来创建并返回一个代理对象。
public class ReferenceBean<T>{ @Override public Object getObject() { return get(); } public synchronized T get() { if (ref == null) { init(); } return ref; } }
在我们使用的时候,实则调用的是代理对象,代理对象完成复杂的远程调用。比如连接注册中心、负载均衡、集群容错、连接服务器发送消息等功能。
2、MyBatis
还有一个典型的应用,就是我们经常在用的MyBatis。我们在使用的时候,一般只操作Mapper接口,然后MyBatis会找到对应的SQL语句来执行。
public interface UserMapper { List<User> getUserList(); }
如上代码,UserMapper也只是一个普通的接口,它是怎样最终执行到我们的SQL语句的呢?
答案也是代理。当MyBatis扫描到我们定义的Mapper接口时,会将其设置为MapperFactoryBean,并创建返回一个代理对象。
protected T newInstance(SqlSession sqlSession) { final MapperProxy<T> mapperProxy = new MapperProxy<>(sqlSession, mapperInterface, methodCache); return (T) Proxy.newProxyInstance(mapperInterface.getClassLoader(), new Class[] { mapperInterface }, mapperProxy); }
代理对象去通过请求的方法名找到MappedStatement对象,调用执行器,解析SqlSource对象来生成SQL,执行并解析返回结果等。
以上案例具体的实现过程,在这里就不再深入细聊。有兴趣可能翻阅笔者其他文章~
九、装饰器模式
装饰器模式,在不改变现有对象结构的情况下,动态地给该对象增加一些职责(即增加其额外功能)的模式,它属于对象结构型模式。
MyBatis里的缓存设计,就是装饰器模式的典型应用。
首先,我们知道,MyBatis执行器是MyBatis调度的核心,它负责SQL语句的生成和执行。
在创建SqlSession的时候,会创建这个执行器,默认的执行器是SimpleExecutor。
但是为了给执行器增加缓存的职责,就变成了在SimpleExecutor上一层添加了CachingExecutor。
在CachingExecutor中的实际操作还是委托给SimpleExecutor去执行,只是在执行前后增加了缓存的操作。
首先,我们来看看它的装饰过程。
public Executor newExecutor(Transaction transaction, ExecutorType executorType) { //默认的执行器 executorType = executorType == null ? defaultExecutorType : executorType; executorType = executorType == null ? ExecutorType.SIMPLE : executorType; Executor executor; if (ExecutorType.BATCH == executorType) { executor = new BatchExecutor(this, transaction); } else if (ExecutorType.REUSE == executorType) { executor = new ReuseExecutor(this, transaction); } else { executor = new SimpleExecutor(this, transaction); } //使用缓存执行器来装饰 if (cacheEnabled) { executor = new CachingExecutor(executor); } executor = (Executor) interceptorChain.pluginAll(executor); return executor; }
当SqlSession执行方法的时候,则会先调用到CachingExecutor,我们来看查询方法。
public class CachingExecutor implements Executor { @Override public <E> List<E> query()throws SQLException { Cache cache = ms.getCache(); if (cache != null) { flushCacheIfRequired(ms); if (ms.isUseCache() && resultHandler == null) { List<E> list = (List<E>) tcm.getObject(cache, key); if (list == null) { list = delegate.query(ms, parameterObject, rowBounds, resultHandler, key, boundSql); tcm.putObject(cache, key, list); // issue #578 and #116 } return list; } } return delegate.query(ms, parameterObject, rowBounds, resultHandler, key, boundSql); } }
这里的代码,如果开启了缓存,则先从缓存中获取结果。如果没有开启缓存或者缓存中没有结果,则再调用SimpleExecutor执行器去数据库中查询。
十、观察者模式
观察者模式,定义对象间的一种一对多的依赖关系,当一个对象的状态发生改变时,所有依赖于它的对象都得到通知并被自动更新。
在Spring或者SpringBoot项目中,有时候我们需要在Spring容器启动并加载完之后,做一些系统初始化的事情。这时候,我们可以配置一个观察者ApplicationListener,来达到这一目的。这就是观察者模式的实践。
@Component public class ApplicationStartup implements ApplicationListener<ContextRefreshedEvent> { @Override public void onApplicationEvent(ContextRefreshedEvent event) { System.out.println("干一些系统初始化的事情...."); ApplicationContext context = event.getApplicationContext(); String[] names = context.getBeanDefinitionNames(); for (String beanName:names){ System.out.println("----------"+beanName+"---------"); } } }
首先,我们知道,ApplicationContext是 Spring 中的核心容器。
public abstract class AbstractApplicationContext extends DefaultResourceLoader implements ConfigurableApplicationContext { //观察者容器 private final Set<ApplicationListener<?>> applicationListeners = new LinkedHashSet<>(); //被观察者 private ApplicationEventMulticaster applicationEventMulticaster; }
在ApplicationContext容器刷新的时候,会初始化一个被观察者,并注册到Spring容器中。
然后,注册各种观察者到被观察者中,形成一对多的依赖。
public abstract class AbstractApplicationContext{ protected void registerListeners() { for (ApplicationListener<?> listener : getApplicationListeners()) { getApplicationEventMulticaster().addApplicationListener(listener); } String[] listenerBeanNames = getBeanNamesForType(ApplicationListener.class, true, false); for (String listenerBeanName : listenerBeanNames) { getApplicationEventMulticaster().addApplicationListenerBean(listenerBeanName); } Set<ApplicationEvent> earlyEventsToProcess = this.earlyApplicationEvents; this.earlyApplicationEvents = null; if (earlyEventsToProcess != null) { for (ApplicationEvent earlyEvent : earlyEventsToProcess) { getApplicationEventMulticaster().multicastEvent(earlyEvent); } } } }
这时候,我们自定义的观察者对象也被注册到了applicationEventMulticaster里面。
最后,当ApplicationContext完成刷新后,则发布ContextRefreshedEvent事件。
protected void finishRefresh() { publishEvent(new ContextRefreshedEvent(this)); }
通知观察者,调用ApplicationListener.onApplicationEvent()。
private void doInvokeListener(ApplicationListener listener, ApplicationEvent event) { listener.onApplicationEvent(event); }
接下来我们再看看在Dubbo是如何应用这一机制的。
Dubbo服务导出过程始于 Spring 容器发布刷新事件,Dubbo 在接收到事件后,会立即执行服务导出逻辑。
public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent>, BeanNameAware, ApplicationEventPublisherAware { @Override public void onApplicationEvent(ContextRefreshedEvent event) { if (!isExported() && !isUnexported()) { if (logger.isInfoEnabled()) { logger.info("The service ready on spring started. service: " + getInterface()); } export(); } } }
我们看到,Dubbo中的ServiceBean也实现了ApplicationListener接口,在Spring容器发布刷新事件之后就会执行导出方法。我们重点关注,在Dubbo执行完导出之后,它也发布了一个事件。
public class ServiceBean<T>{ public void export() { super.export(); publishExportEvent(); } private void publishExportEvent() { ServiceBeanExportedEvent exportEvent = new ServiceBeanExportedEvent(this); applicationEventPublisher.publishEvent(exportEvent); } }
ServiceBeanExportedEvent,服务导出事件,需要继承Spring中的事件对象ApplicationEvent。
public class ServiceBeanExportedEvent extends ApplicationEvent { public ServiceBeanExportedEvent(ServiceBean serviceBean) { super(serviceBean); } public ServiceBean getServiceBean() { return (ServiceBean) super.getSource(); } }
然后我们自定义一个ApplicationListener,也就是观察者,就可以监听到Dubbo服务接口导出事件了。
@Component public class ServiceBeanListener implements ApplicationListener<ServiceBeanExportedEvent> { @Override public void onApplicationEvent(ServiceBeanExportedEvent event) { ServiceBean serviceBean = event.getServiceBean(); String beanName = serviceBean.getBeanName(); Service service = serviceBean.getService(); System.out.println(beanName+":"+service); } }
十一、命令模式
命令模式是一种行为设计模式,它可将请求转换为一个包含与请求相关的所有信息的独立对象。该转换让你能根据不同的请求将方法参数化、延迟请求执行或将其放入队列中,且能实现可撤销操作。
Hystrix是Netflix开源的一款容错框架,具有自我保护能力。可以阻止故障的连锁反应,快速失败和优雅降级。
它用一个HystrixCommand或者HystrixObservableCommand包装所有对外部系统/依赖的调用,每个命令在单独线程中/信号授权下执行。这正是命令模式的典型应用。
我们来看一个Hystrix应用的例子。
首先,我们需要创建一个具体的命令类,通过构造函数传递接收者对象。
public class OrderServiceHystrixCommand extends HystrixCommand<Object> { //接收者,处理业务逻辑 private OrderService orderService; public OrderServiceHystrixCommand(OrderService orderService) { super(setter()); this.orderService = orderService; } //设置Hystrix相关参数 public static Setter setter() { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("orderGroup"); HystrixCommandKey commandKey = HystrixCommandKey.Factory.asKey("orderService"); HystrixThreadPoolProperties.Setter threadPoolProperties = HystrixThreadPoolProperties.Setter().withCoreSize(1) .withQueueSizeRejectionThreshold(1); HystrixCommandProperties.Setter commandProperties = HystrixCommandProperties.Setter(); return Setter.withGroupKey(groupKey) .andCommandKey(commandKey) .andThreadPoolPropertiesDefaults(threadPoolProperties) .andCommandPropertiesDefaults(commandProperties); } @Override protected Object run() throws InterruptedException { Thread.sleep(500); return orderService.orders(); } @Override protected Object getFallback() { System.out.println("-------------------------------"); return new ArrayList(); } }
然后,在客户端调用的时候,创建这个命令类并执行即可。
@RestController public class OrderController { @Autowired OrderService orderService; @RequestMapping("/orders") public Object orders(){ OrderServiceHystrixCommand command = new OrderServiceHystrixCommand(orderService); return command.execute(); } }
看上去,命令模式和策略模式有些相像,它们都可以通过某些行为来参数化对象。但它们的思想有很大区别。
比如说我们可以使用命令来将任何操作转换为对象,操作的参数将成为对象的成员变量。同样的,我们也可以对请求做任何操作,比如延迟执行,记录日志,保存历史命令等。
而策略模式侧重点在于描述完成某件事的不同方式,让你能够在同一个上下文类中切换算法。
总结
本文重点介绍了设计模式在不同框架中的实现,以期让大家更好地理解模式背后的思想和应用场景。欢迎有不同想法的朋友,留言探讨~