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);

}

巴特西