Java开源框架中的设计模式以及应用场景
前言
一,单例模式
public class ConfigManager {
private static final ConfigManager configManager = new ConfigManager();
private ConfigManager() {}
public static ConfigManager getInstance() {
return configManager;
}
}
public class MQClientManager {
private static MQClientManager instance = new MQClientManager();
private MQClientManager() {}
public static MQClientManager getInstance() {
return instance;
}
}
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;
}
}
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;
}
}
二,工厂模式
public abstract class AbstractUndoExecutor{
//生成撤销SQL
protected abstract String buildUndoSQL();
}
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);
public interface Cache {
void put(Object key, Object value);
Object get(Object key);
}
public interface CacheFactory {
Cache getCache(URL url, Invocation invocation);
}
public abstract class AbstractCacheFactory implements CacheFactory {
//具体的缓存实现类
private final ConcurrentMapcaches = new ConcurrentHashMap ();
@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);
}
ExpiringCacheFactory、JCacheFactory、LruCacheFactory、ThreadLocalCacheFactory。
这些工厂类,只有一个方法,就是创建具体的缓存实现类。
public class ThreadLocalCacheFactory extends AbstractCacheFactory {
@Override
protected Cache createCache(URL url) {
return new ThreadLocalCache(url);
}
}
public class ThreadLocalCache implements Cache {
private final ThreadLocal
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"));
}
工厂方法模式只有一个抽象产品类,具体工厂类只能创建一个具体产品类的实例;
抽象工厂模式有多个抽象产品类,具体工厂类可以创建多个具体产品类的实例。
public interface DataBase {
void insert(Object tableName, Object record);
Object select(Object tableName);
}
public class MysqlDataBase implements DataBase{
MapmysqlDb = 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 {
MaporacleDb = 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();
}
public class DataAccessFactory1 implements DataAccessFactory {
@Override
public Cache getCache(URL url) {
return new ThreadLocalCache(url);
}
@Override
public DataBase getDb() {
return new MysqlDataBase();
}
}
public class DataAccessFactory2 implements DataAccessFactory {
@Override
public Cache getCache(URL url) {
return new LruCache(url);
}
@Override
public DataBase getDb() {
return new OracleDataBase();
}
}
三,模板方式模式
public interface LoadBalance {
Invoker select(List > invokers, URL url, Invocation invocation) throws RpcException;
}
public abstract class AbstractLoadBalance implements LoadBalance {
@Override
publicInvoker select(List > 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 abstractInvoker doSelect(List > invokers, URL url, Invocation invocation);
}
基于权重随机算法的 RandomLoadBalance
基于最少活跃调用数算法的 LeastActiveLoadBalance
基于 hash 一致性的 ConsistentHashLoadBalance
基于加权轮询算法的 RoundRobinLoadBalance
public class RandomLoadBalance extends AbstractLoadBalance {
@Override
protectedInvoker doSelect(List > invokers, URL url, Invocation invocation) {
//省略逻辑....
return invokers.get(ThreadLocalRandom.current().nextInt(length));
}
}
四,构造器模式
public final class MappedStatement {
private String resource;
private Configuration configuration;
private String id;
private SqlSource sqlSource;
private ParameterMap parameterMap;
private ListresultMaps;
//.....省略大部分属性
}
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;
//.....省略大部分过程
}
}
public static class Builder {
public Builder parameterMap(ParameterMap parameterMap) {
mappedStatement.parameterMap = parameterMap;
return this;
}
public Builder resultMaps(ListresultMaps) {
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;
}
}
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;
}
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;
五,适配器模式
@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
}
}
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, Listout) 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() {
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 Pairmatched = this.processorTable.get(cmd.getCode());
//如果没有找到就使用默认的
final Pairpair =
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 {
@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 {
ListgetUserList();
}如上代码,UserMapper也只是一个普通的接口,它是怎样最终执行到我们的SQL语句的呢? 答案也是代理。当MyBatis扫描到我们定义的Mapper接口时,会将其设置为MapperFactoryBean,并创建返回一个代理对象。 protected T newInstance(SqlSession sqlSession) {
final MapperProxymapperProxy = 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
publicList query()throws SQLException {
Cache cache = ms.getCache();
if (cache != null) {
flushCacheIfRequired(ms);
if (ms.isUseCache() && resultHandler == null) {
Listlist = (List ) 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{
@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> 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);
}
SetearlyEventsToProcess = 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
extends ServiceConfig implements InitializingBean, DisposableBean,
ApplicationContextAware, ApplicationListener, 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
{
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{
@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
{
//接收者,处理业务逻辑
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();
}
}看上去,命令模式和策略模式有些相像,它们都可以通过某些行为来参数化对象。但它们的思想有很大区别。 比如说我们可以使用命令来将任何操作转换为对象,操作的参数将成为对象的成员变量。同样的,我们也可以对请求做任何操作,比如延迟执行,记录日志,保存历史命令等。 而策略模式侧重点在于描述完成某件事的不同方式,让你能够在同一个上下文类中切换算法。
总结
本文重点介绍了设计模式在不同框架中的实现,以期让大家更好地理解模式背后的思想和应用场景。欢迎有不同想法的朋友,留言探讨~ 原文链接:https://juejin.im/post/6859160350692999181
往期推荐
浏览 35 后台回复 学习资料 领取学习视频
如有收获,点个在看,诚挚感谢
评论