Netty中使用的设计模式

创建型

简单工厂

public class DefaultThreadFactory implements ThreadFactory {
    @Override
    public Thread newThread(Runnable r) {
        Thread t = newThread(FastThreadLocalRunnable.wrap(r), prefix + nextId.incrementAndGet());
        try {
            if (t.isDaemon() != daemon) {
                t.setDaemon(daemon);
            }

            if (t.getPriority() != priority) {
                t.setPriority(priority);
            }
        } catch (Exception ignored) {
            // Doesn't matter even if failed to set.
        }
        return t;
    }

    protected Thread newThread(Runnable r, String name) {
        return new FastThreadLocalThread(threadGroup, r, name);
    }
}

建造者模式

public final class ClientCookieDecoder extends CookieDecoder {

    ...
    if (cookieBuilder == null) {
        cookieBuilder = new CookieBuilder(cookie, header);
    } else {
        cookieBuilder.appendAttribute(nameBegin, nameEnd, valueBegin, valueEnd);
    }
    
    private static class CookieBuilder {
        void appendAttribute(int keyStart, int keyEnd, int valueStart, int valueEnd) {}
        
        private void parse4(int nameStart, int valueStart, int valueEnd) {
            ...
        }

        private void parse6(int nameStart, int valueStart, int valueEnd) {
            ...
        }
    } 
}

单例模式

构造函数private修饰

public final class ReadTimeoutException extends TimeoutException {

    private static final long serialVersionUID = 169287984113283421L;

    public static final ReadTimeoutException INSTANCE = PlatformDependent.javaVersion() >= 7 ?
            new ReadTimeoutException(true) : new ReadTimeoutException();

    ReadTimeoutException() { }

    private ReadTimeoutException(boolean shared) {
        super(shared);
    }
}

public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {

    public static final MqttEncoder INSTANCE = new MqttEncoder();

    private MqttEncoder() { }
}

结构型

策略模式

在新连接接入的时候，EventExcutorChooser会根据EventExcutor的长度来进行选择。因为默认构造函数是创建CPU两倍的线程，也可以用户自定义线程个数。那么这里将接入的新连接交给哪个线程，就用到了这种策略模式。

public final class DefaultEventExecutorChooserFactory implements EventExecutorChooserFactory {

    @Override
    public EventExecutorChooser newChooser(EventExecutor[] executors) {
        if (isPowerOfTwo(executors.length)) {
            return new PowerOfTwoEventExecutorChooser(executors);
        } else {
            return new GenericEventExecutorChooser(executors);
        }
    }
    
    private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {
        private final AtomicInteger idx = new AtomicInteger();
        private final EventExecutor[] executors;

        PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
        }

        @Override
        public EventExecutor next() {
            return executors[idx.getAndIncrement() & executors.length - 1];
        }
    }

    private static final class GenericEventExecutorChooser implements EventExecutorChooser {
        // Use a 'long' counter to avoid non-round-robin behaviour at the 32-bit overflow boundary.
        // The 64-bit long solves this by placing the overflow so far into the future, that no system
        // will encounter this in practice.
        private final AtomicLong idx = new AtomicLong();
        private final EventExecutor[] executors;

        GenericEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
        }

        @Override
        public EventExecutor next() {
            return executors[(int) Math.abs(idx.getAndIncrement() % executors.length)];
        }
    }
}

装饰模式

WrappedByteBuf、UnreleasableByteBuf、SimpleLeakAwareByteBuf。第一个类就相当于装饰者父类，后两个就是装饰类，而ByteBuf就是原型。

class WrappedByteBuf extends ByteBuf {
    @Override
    public boolean release(int decrement) {
        return buf.release(decrement);
    }
}

class SimpleLeakAwareByteBuf extends WrappedByteBuf {
@Override
public boolean release(int decrement) {
if (super.release(decrement)) {
closeLeak();
return true;
}
return false;
}

private void closeLeak() {
    // Close the ResourceLeakTracker with the tracked ByteBuf as argument. This must be the same that was used when
    // calling DefaultResourceLeak.track(...).
    boolean closed = leak.close(trackedByteBuf);
    assert closed;
}

}

行为型

适配器模式

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {}

public abstract class ChannelHandlerAdapter implements ChannelHandler {}

public interface ChannelHandler {
    void handlerAdded(ChannelHandlerContext ctx) throws Exception;
}

代理模式

public abstract class ProxyHandler extends ChannelDuplexHandler {
    private final SocketAddress proxyAddress;
    
    protected ProxyHandler(SocketAddress proxyAddress) {
        this.proxyAddress = ObjectUtil.checkNotNull(proxyAddress, "proxyAddress");
    }
    
    @Override
    public final void connect(
            ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
            ChannelPromise promise) throws Exception {

        if (destinationAddress != null) {
            promise.setFailure(new ConnectionPendingException());
            return;
        }

        destinationAddress = remoteAddress;
        ctx.connect(proxyAddress, localAddress, promise);
    }
}

观察者模式

ChannelFuture中有个重要的addListener方法，会添加特定的监听器到future中，这些监听器会在future isDone返回true的时候立刻被通知

public interface ChannelFuture extends Future<Void> {

    @Override
    ChannelFuture addListener(GenericFutureListener<? extends Future<? super Void>> listener);
}

迭代器模式

public abstract class AbstractByteBuf extends ByteBuf {
    @Override
    public int forEachByte(ByteProcessor processor) {
        ensureAccessible();
        try {
            return forEachByteAsc0(readerIndex, writerIndex, processor);
        } catch (Exception e) {
            PlatformDependent.throwException(e);
            return -1;
        }
    }
    
    int forEachByteAsc0(int start, int end, ByteProcessor processor) throws Exception {
        for (; start < end; ++start) {
            if (!processor.process(_getByte(start))) {
                return start;
            }
        }

        return -1;
    }
}

责任链模式

责任链模式（Chain of Responsibility Pattern）为请求创建了一个接收者对象的链

public interface ChannelPipeline
        extends ChannelInboundInvoker, ChannelOutboundInvoker, Iterable<Entry<String, ChannelHandler>> {
        
    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);
}