• Netty入门-Channel

    目录

    Channel详解

    Channel的特点

    Channel接口方法

    ChannelOutboundInvoker接口

    AttributeMap接口

    ChannelHandler接口

    ChannelInboundHandler接口

    ChannelOutboundHandler接口

    ChannelHandlerAdapter

    ChannelInboundHandlerAdapor

    ChannelOutboundHandlerAdapter

    适配器的作用

    ChannelPipeline接口

    创建ChannelPipeline

    ChannelHandlerContext接口

    Channel详解

            Channel代表网络socket或能够进行IO操作的组件的连接关系。这些IO操作包括读、写、连接和绑定。Netty中的Channel提供了如下功能：

    • 查询当前Channel的状态。例如，是打开还是已连接状态等。
    • 提供Channel的参数配置。如接收缓冲区大小。
    • 提供支持的IO操作。如读、写连接和绑定
    • 提供ChannelPipeline。ChannelPipelin用于处理所有与Channel关联的IO事件和请求。

    Channel的特点

    1. 所有IO操作都是异步的

    IO调用将立即返回，返回一个ChannelFuture实例。

    2. Channel是分层的

    3. 向下转型以下访问特定于传输的操作

    4. 释放资源

    Channel接口方法

    1. public interface Channel extends AttributeMap, ChannelOutboundInvoker, Comparable {
    2.     ChannelId id();
    3.     EventLoop eventLoop();
    4.     Channel parent();
    5.     ChannelConfig config();
    6.     boolean isOpen();
    7.     boolean isRegistered();
    8.     boolean isActive();
    9.     ChannelMetadata metadata();
    10.     SocketAddress localAddress();
    11.     SocketAddress remoteAddress();
    12.     ChannelFuture closeFuture();
    13.     boolean isWritable();
    14.     long bytesBeforeUnwritable();
    15.     long bytesBeforeWritable();
    16.     Channel.Unsafe unsafe();
    17.     ChannelPipeline pipeline();
    18.     ByteBufAllocator alloc();
    19.     Channel read();
    20.     Channel flush();
    21.     public interface Unsafe {
    22.         Handle recvBufAllocHandle();
    23.         SocketAddress localAddress();
    24.         SocketAddress remoteAddress();
    25.         void register(EventLoop var1, ChannelPromise var2);
    26.         void bind(SocketAddress var1, ChannelPromise var2);
    27.         void connect(SocketAddress var1, SocketAddress var2, ChannelPromise var3);
    28.         void disconnect(ChannelPromise var1);
    29.         void close(ChannelPromise var1);
    30.         void closeForcibly();
    31.         void deregister(ChannelPromise var1);
    32.         void beginRead();
    33.         void write(Object var1, ChannelPromise var2);
    34.         void flush();
    35.         ChannelPromise voidPromise();
    36.         ChannelOutboundBuffer outboundBuffer();
    37.     }
    38. }
    • id()方法返回全局唯一的ChannelId
    • eventLoop()方法返回分配给该Channel的EventLoop，一个EventLoop就是一个线程，用来处理连接的生命周期中所发生的事件
    • parent()方法返回该Channel的父Channel
    • config()方法返回该Channel的ChannelConfig，其中包含了该Channel的所有配置设置支持热更新
    • pipeline()方法返回该Channel对应的ChannelPipeline
    • alloc方法返回分配给该Channel的ByteBufAllocator，可以用来分配ByteBuf

    ChannelOutboundInvoker接口

    声明了所有出站的网络操作:

    1. package io.netty.channel;
    2.  
    3. import java.net.SocketAddress;
    4.  
    5. public interface ChannelOutboundInvoker {
    6.     ChannelFuture bind(SocketAddress var1);
    7.     ChannelFuture connect(SocketAddress var1);
    8.     ChannelFuture connect(SocketAddress var1, SocketAddress var2);
    9.     ChannelFuture disconnect();
    10.     ChannelFuture close();
    11.     ChannelFuture deregister();
    12.     ChannelFuture bind(SocketAddress var1, ChannelPromise var2);
    13.     ChannelFuture connect(SocketAddress var1, ChannelPromise var2);
    14.     ChannelFuture connect(SocketAddress var1, SocketAddress var2, ChannelPromise var3);
    15.     ChannelFuture disconnect(ChannelPromise var1);
    16.     ChannelFuture close(ChannelPromise var1);
    17.     ChannelFuture deregister(ChannelPromise var1);
    18.     ChannelOutboundInvoker read();
    19.     ChannelFuture write(Object var1);
    20.     ChannelFuture write(Object var1, ChannelPromise var2);
    21.     ChannelOutboundInvoker flush();
    22.     ChannelFuture writeAndFlush(Object var1, ChannelPromise var2);
    23.     ChannelFuture writeAndFlush(Object var1);
    24.     ChannelPromise newPromise();
    25.     ChannelProgressivePromise newProgressivePromise();
    26.     ChannelFuture newSucceededFuture();
    27.     ChannelFuture newFailedFuture(Throwable var1);
    28.     ChannelPromise voidPromise();
    29. }

    ChannelFuture用于获取异步的结果，ChannelPromise是对ChannelFuture的扩展，支持写的操作。ChannelPromise也被称为可写的ChannelFuture。

    AttributeMap接口

    1. package io.netty.util;
    2. public interface AttributeMap {
    3.      Attribute attr(AttributeKey var1);
    4.      boolean hasAttr(AttributeKey var1);
    5. }

    AttributeMap就是类似于Map的键值对，键就是AttributeKey类型，值是Attribute类型。

    Netty提供了AttributeMap的默认实现类DefaultAttributeMap，与JDK中的ConcurrentHashMap相比，在高并发下DefaultAttributeMap可以更加节省内存。

    1. package io.netty.util;
    2.  
    3. import io.netty.util.internal.ObjectUtil;
    4.  
    5. import java.util.Arrays;
    6. import java.util.concurrent.atomic.AtomicReference;
    7. import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
    8.  
    9. /**
    10.  * Default {@link AttributeMap} implementation which not exibit any blocking behaviour on attribute lookup while using a
    11.  * copy-on-write approach on the modify path.
       Attributes lookup and remove exibit {@code O(logn)} time worst-case
    12.  * complexity, hence {@code attribute::set(null)} is to be preferred to {@code remove}.
    13.  */
    14. public class DefaultAttributeMap implements AttributeMap {
    15.  
    16.     private static final AtomicReferenceFieldUpdater ATTRIBUTES_UPDATER =
    17.             AtomicReferenceFieldUpdater.newUpdater(DefaultAttributeMap.class, DefaultAttribute[].class, "attributes");
    18.     private static final DefaultAttribute[] EMPTY_ATTRIBUTES = new DefaultAttribute[0];
    19.  
    20.     /**
    21.      * Similarly to {@code Arrays::binarySearch} it perform a binary search optimized for this use case, in order to
    22.      * save polymorphic calls (on comparator side) and unnecessary class checks.
    23.      */
    24.     private static int searchAttributeByKey(DefaultAttribute[] sortedAttributes, AttributeKey key) {
    25.         int low = 0;
    26.         int high = sortedAttributes.length - 1;
    27.  
    28.         while (low <= high) {
    29.             int mid = low + high >>> 1;
    30.             DefaultAttribute midVal = sortedAttributes[mid];
    31.             AttributeKey midValKey = midVal.key;
    32.             if (midValKey == key) {
    33.                 return mid;
    34.             }
    35.             int midValKeyId = midValKey.id();
    36.             int keyId = key.id();
    37.             assert midValKeyId != keyId;
    38.             boolean searchRight = midValKeyId < keyId;
    39.             if (searchRight) {
    40.                 low = mid + 1;
    41.             } else {
    42.                 high = mid - 1;
    43.             }
    44.         }
    45.  
    46.         return -(low + 1);
    47.     }
    48.  
    49.     private static void orderedCopyOnInsert(DefaultAttribute[] sortedSrc, int srcLength, DefaultAttribute[] copy,
    50.                                             DefaultAttribute toInsert) {
    51.         // let's walk backward, because as a rule of thumb, toInsert.key.id() tends to be higher for new keys
    52.         final int id = toInsert.key.id();
    53.         int i;
    54.         for (i = srcLength - 1; i >= 0; i--) {
    55.             DefaultAttribute attribute = sortedSrc[i];
    56.             assert attribute.key.id() != id;
    57.             if (attribute.key.id() < id) {
    58.                 break;
    59.             }
    60.             copy[i + 1] = sortedSrc[i];
    61.         }
    62.         copy[i + 1] = toInsert;
    63.         final int toCopy = i + 1;
    64.         if (toCopy > 0) {
    65.             System.arraycopy(sortedSrc, 0, copy, 0, toCopy);
    66.         }
    67.     }
    68.  
    69.     private volatile DefaultAttribute[] attributes = EMPTY_ATTRIBUTES;
    70.  
    71.     @SuppressWarnings("unchecked")
    72.     @Override
    73.     public  Attribute attr(AttributeKey key) {
    74.         ObjectUtil.checkNotNull(key, "key");
    75.         DefaultAttribute newAttribute = null;
    76.         for (;;) {
    77.             final DefaultAttribute[] attributes = this.attributes;
    78.             final int index = searchAttributeByKey(attributes, key);
    79.             final DefaultAttribute[] newAttributes;
    80.             if (index >= 0) {
    81.                 final DefaultAttribute attribute = attributes[index];
    82.                 assert attribute.key() == key;
    83.                 if (!attribute.isRemoved()) {
    84.                     return attribute;
    85.                 }
    86.                 // let's try replace the removed attribute with a new one
    87.                 if (newAttribute == null) {
    88.                     newAttribute = new DefaultAttribute(this, key);
    89.                 }
    90.                 final int count = attributes.length;
    91.                 newAttributes = Arrays.copyOf(attributes, count);
    92.                 newAttributes[index] = newAttribute;
    93.             } else {
    94.                 if (newAttribute == null) {
    95.                     newAttribute = new DefaultAttribute(this, key);
    96.                 }
    97.                 final int count = attributes.length;
    98.                 newAttributes = new DefaultAttribute[count + 1];
    99.                 orderedCopyOnInsert(attributes, count, newAttributes, newAttribute);
    100.             }
    101.             if (ATTRIBUTES_UPDATER.compareAndSet(this, attributes, newAttributes)) {
    102.                 return newAttribute;
    103.             }
    104.         }
    105.     }
    106.  
    107.     @Override
    108.     public  boolean hasAttr(AttributeKey key) {
    109.         ObjectUtil.checkNotNull(key, "key");
    110.         return searchAttributeByKey(attributes, key) >= 0;
    111.     }
    112.  
    113.     private  void removeAttributeIfMatch(AttributeKey key, DefaultAttribute value) {
    114.         for (;;) {
    115.             final DefaultAttribute[] attributes = this.attributes;
    116.             final int index = searchAttributeByKey(attributes, key);
    117.             if (index < 0) {
    118.                 return;
    119.             }
    120.             final DefaultAttribute attribute = attributes[index];
    121.             assert attribute.key() == key;
    122.             if (attribute != value) {
    123.                 return;
    124.             }
    125.             final int count = attributes.length;
    126.             final int newCount = count - 1;
    127.             final DefaultAttribute[] newAttributes =
    128.                     newCount == 0? EMPTY_ATTRIBUTES : new DefaultAttribute[newCount];
    129.             // perform 2 bulk copies
    130.             System.arraycopy(attributes, 0, newAttributes, 0, index);
    131.             final int remaining = count - index - 1;
    132.             if (remaining > 0) {
    133.                 System.arraycopy(attributes, index + 1, newAttributes, index, remaining);
    134.             }
    135.             if (ATTRIBUTES_UPDATER.compareAndSet(this, attributes, newAttributes)) {
    136.                 return;
    137.             }
    138.         }
    139.     }
    140.  
    141.     @SuppressWarnings("serial")
    142.     private static final class DefaultAttribute extends AtomicReference implements Attribute {
    143.  
    144.         private static final AtomicReferenceFieldUpdater MAP_UPDATER =
    145.                 AtomicReferenceFieldUpdater.newUpdater(DefaultAttribute.class,
    146.                                                        DefaultAttributeMap.class, "attributeMap");
    147.         private static final long serialVersionUID = -2661411462200283011L;
    148.  
    149.         private volatile DefaultAttributeMap attributeMap;
    150.         private final AttributeKey key;
    151.  
    152.         DefaultAttribute(DefaultAttributeMap attributeMap, AttributeKey key) {
    153.             this.attributeMap = attributeMap;
    154.             this.key = key;
    155.         }
    156.  
    157.         @Override
    158.         public AttributeKey key() {
    159.             return key;
    160.         }
    161.  
    162.         private boolean isRemoved() {
    163.             return attributeMap == null;
    164.         }
    165.  
    166.         @Override
    167.         public T setIfAbsent(T value) {
    168.             while (!compareAndSet(null, value)) {
    169.                 T old = get();
    170.                 if (old != null) {
    171.                     return old;
    172.                 }
    173.             }
    174.             return null;
    175.         }
    176.  
    177.         @Override
    178.         public T getAndRemove() {
    179.             final DefaultAttributeMap attributeMap = this.attributeMap;
    180.             final boolean removed = attributeMap != null && MAP_UPDATER.compareAndSet(this, attributeMap, null);
    181.             T oldValue = getAndSet(null);
    182.             if (removed) {
    183.                 attributeMap.removeAttributeIfMatch(key, this);
    184.             }
    185.             return oldValue;
    186.         }
    187.  
    188.         @Override
    189.         public void remove() {
    190.             final DefaultAttributeMap attributeMap = this.attributeMap;
    191.             final boolean removed = attributeMap != null && MAP_UPDATER.compareAndSet(this, attributeMap, null);
    192.             set(null);
    193.             if (removed) {
    194.                 attributeMap.removeAttributeIfMatch(key, this);
    195.             }
    196.         }
    197.     }
    198. }

    ChannelHandler接口

    1. package io.netty.channel;
    2.  
    3. import io.netty.util.Attribute;
    4. import io.netty.util.AttributeKey;
    5.  
    6. import java.lang.annotation.Documented;
    7. import java.lang.annotation.ElementType;
    8. import java.lang.annotation.Inherited;
    9. import java.lang.annotation.Retention;
    10. import java.lang.annotation.RetentionPolicy;
    11. import java.lang.annotation.Target;
    12.  
    13.  
    14. public interface ChannelHandler {
    15.  
    16.     /**
    17.      * Gets called after the {@link ChannelHandler} was added to the actual context and it's ready to handle events.
    18.      */
    19.     void handlerAdded(ChannelHandlerContext ctx) throws Exception;
    20.  
    21.     /**
    22.      * Gets called after the {@link ChannelHandler} was removed from the actual context and it doesn't handle events
    23.      * anymore.
    24.      */
    25.     void handlerRemoved(ChannelHandlerContext ctx) throws Exception;
    26.  
    27.     /**
    28.      * Gets called if a {@link Throwable} was thrown.
    29.      *
    30.      * @deprecated if you want to handle this event you should implement {@link ChannelInboundHandler} and
    31.      * implement the method there.
    32.      */
    33.     @Deprecated
    34.     void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
    35.  
    36.     /**
    37.      * Indicates that the same instance of the annotated {@link ChannelHandler}
    38.      * can be added to one or more {@link ChannelPipeline}s multiple times
    39.      * without a race condition.
    40.      * 
    41.      * If this annotation is not specified, you have to create a new handler
    42.      * instance every time you add it to a pipeline because it has unshared
    43.      * state such as member variables.
    44.      * 
    45.      * This annotation is provided for documentation purpose, just like
    46.      * the JCIP annotations.
    47.      */
    48.     @Inherited
    49.     @Documented
    50.     @Target(ElementType.TYPE)
    51.     @Retention(RetentionPolicy.RUNTIME)
    52.     @interface Sharable {
    53.         // no value
    54.     }
    55. }


     Handles an I/O event or intercepts an I/O operation, and forwards it to its next handler in
     its ChannelPipeline

    ChannelHandler本身没有提供什么方法，可以使用其子类：

    • ChannelInboundHandler：处理入站IO事件
    • ChannelOutboundHandler：处理出站IO事件
    • ChannelHandlerAdapter：采用适配器模式的ChannelHandler适配器

    ChannelInboundHandler接口

    1. package io.netty.channel;
    2.  
    3. /**
    4.  * {@link ChannelHandler} which adds callbacks for state changes. This allows the user
    5.  * to hook in to state changes easily.
    6.  */
    7. public interface ChannelInboundHandler extends ChannelHandler {
    8.  
    9.     /**
    10.      * The {@link Channel} of the {@link ChannelHandlerContext} was registered with its {@link EventLoop}
    11.      */
    12.     void channelRegistered(ChannelHandlerContext ctx) throws Exception;
    13.  
    14.     /**
    15.      * The {@link Channel} of the {@link ChannelHandlerContext} was unregistered from its {@link EventLoop}
    16.      */
    17.     void channelUnregistered(ChannelHandlerContext ctx) throws Exception;
    18.  
    19.     /**
    20.      * The {@link Channel} of the {@link ChannelHandlerContext} is now active
    21.      */
    22.     void channelActive(ChannelHandlerContext ctx) throws Exception;
    23.  
    24.     /**
    25.      * The {@link Channel} of the {@link ChannelHandlerContext} was registered is now inactive and reached its
    26.      * end of lifetime.
    27.      */
    28.     void channelInactive(ChannelHandlerContext ctx) throws Exception;
    29.  
    30.     /**
    31.      * Invoked when the current {@link Channel} has read a message from the peer.
    32.      */
    33.     void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception;
    34.  
    35.     /**
    36.      * Invoked when the last message read by the current read operation has been consumed by
    37.      * {@link #channelRead(ChannelHandlerContext, Object)}.  If {@link ChannelOption#AUTO_READ} is off, no further
    38.      * attempt to read an inbound data from the current {@link Channel} will be made until
    39.      * {@link ChannelHandlerContext#read()} is called.
    40.      */
    41.     void channelReadComplete(ChannelHandlerContext ctx) throws Exception;
    42.  
    43.     /**
    44.      * Gets called if an user event was triggered.
    45.      */
    46.     void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception;
    47.  
    48.     /**
    49.      * Gets called once the writable state of a {@link Channel} changed. You can check the state with
    50.      * {@link Channel#isWritable()}.
    51.      */
    52.     void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception;
    53.  
    54.     /**
    55.      * Gets called if a {@link Throwable} was thrown.
    56.      */
    57.     @Override
    58.     @SuppressWarnings("deprecation")
    59.     void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
    60. }

    ChannelOutboundHandler接口

    1. package io.netty.channel;
    2.  
    3. import java.net.SocketAddress;
    4.  
    5.  
    6. public interface ChannelOutboundHandler extends ChannelHandler {
    7.     /**
    8.      * Called once a bind operation is made.
    9.      *
    10.      * @param ctx           the {@link ChannelHandlerContext} for which the bind operation is made
    11.      * @param localAddress  the {@link SocketAddress} to which it should bound
    12.      * @param promise       the {@link ChannelPromise} to notify once the operation completes
    13.      * @throws Exception    thrown if an error occurs
    14.      */
    15.     void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception;
    16.  
    17.     /**
    18.      * Called once a connect operation is made.
    19.      *
    20.      * @param ctx               the {@link ChannelHandlerContext} for which the connect operation is made
    21.      * @param remoteAddress     the {@link SocketAddress} to which it should connect
    22.      * @param localAddress      the {@link SocketAddress} which is used as source on connect
    23.      * @param promise           the {@link ChannelPromise} to notify once the operation completes
    24.      * @throws Exception        thrown if an error occurs
    25.      */
    26.     void connect(
    27.             ChannelHandlerContext ctx, SocketAddress remoteAddress,
    28.             SocketAddress localAddress, ChannelPromise promise) throws Exception;
    29.  
    30.     /**
    31.      * Called once a disconnect operation is made.
    32.      *
    33.      * @param ctx               the {@link ChannelHandlerContext} for which the disconnect operation is made
    34.      * @param promise           the {@link ChannelPromise} to notify once the operation completes
    35.      * @throws Exception        thrown if an error occurs
    36.      */
    37.     void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;
    38.  
    39.     /**
    40.      * Called once a close operation is made.
    41.      *
    42.      * @param ctx               the {@link ChannelHandlerContext} for which the close operation is made
    43.      * @param promise           the {@link ChannelPromise} to notify once the operation completes
    44.      * @throws Exception        thrown if an error occurs
    45.      */
    46.     void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;
    47.  
    48.     /**
    49.      * Called once a deregister operation is made from the current registered {@link EventLoop}.
    50.      *
    51.      * @param ctx               the {@link ChannelHandlerContext} for which the close operation is made
    52.      * @param promise           the {@link ChannelPromise} to notify once the operation completes
    53.      * @throws Exception        thrown if an error occurs
    54.      */
    55.     void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;
    56.  
    57.     /**
    58.      * Intercepts {@link ChannelHandlerContext#read()}.
    59.      */
    60.     void read(ChannelHandlerContext ctx) throws Exception;
    61.  
    62.     /**
    63.     * Called once a write operation is made. The write operation will write the messages through the
    64.      * {@link ChannelPipeline}. Those are then ready to be flushed to the actual {@link Channel} once
    65.      * {@link Channel#flush()} is called
    66.      *
    67.      * @param ctx               the {@link ChannelHandlerContext} for which the write operation is made
    68.      * @param msg               the message to write
    69.      * @param promise           the {@link ChannelPromise} to notify once the operation completes
    70.      * @throws Exception        thrown if an error occurs
    71.      */
    72.     void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception;
    73.  
    74.     /**
    75.      * Called once a flush operation is made. The flush operation will try to flush out all previous written messages
    76.      * that are pending.
    77.      *
    78.      * @param ctx               the {@link ChannelHandlerContext} for which the flush operation is made
    79.      * @throws Exception        thrown if an error occurs
    80.      */
    81.     void flush(ChannelHandlerContext ctx) throws Exception;
    82. }

    ChannelHandlerAdapter

    1. package io.netty.channel;
    2.  
    3. import io.netty.channel.ChannelHandler.Sharable;
    4. import io.netty.channel.ChannelHandlerMask.Skip;
    5. import io.netty.util.internal.InternalThreadLocalMap;
    6. import java.util.Map;
    7.  
    8. public abstract class ChannelHandlerAdapter implements ChannelHandler {
    9.     boolean added;
    10.     public ChannelHandlerAdapter() {
    11.     }
    12.     protected void ensureNotSharable() {
    13.         if (this.isSharable()) {
    14.             throw new IllegalStateException("ChannelHandler " + this.getClass().getName() + " is not allowed to be shared");
    15.         }
    16.     }
    17.     public boolean isSharable() {
    18.         Class clazz = this.getClass();
    19.         Map, Boolean> cache = InternalThreadLocalMap.get().handlerSharableCache();
    20.         Boolean sharable = (Boolean)cache.get(clazz);
    21.         if (sharable == null) {
    22.             sharable = clazz.isAnnotationPresent(Sharable.class);
    23.             cache.put(clazz, sharable);
    24.         }
    25.         return sharable;
    26.     }
    27.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    28.     }
    29.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
    30.     }
    31.     /** @deprecated */
    32.     @Skip
    33.     @Deprecated
    34.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    35.         ctx.fireExceptionCaught(cause);
    36.     }
    37. }

    ChannelHandlerAdaptor常用的两个子类，分别是ChannelInboundHandlerAdapor、ChannelOutboundHandlerAdatper

    ChannelInboundHandlerAdapor

    1. package io.netty.channel;
    2.  
    3. import io.netty.channel.ChannelHandlerMask.Skip;
    4.  
    5. public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {
    6.     public ChannelInboundHandlerAdapter() {
    7.     }
    8.  
    9.     @Skip
    10.     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
    11.         ctx.fireChannelRegistered();
    12.     }
    13.  
    14.     @Skip
    15.     public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
    16.         ctx.fireChannelUnregistered();
    17.     }
    18.  
    19.     @Skip
    20.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
    21.         ctx.fireChannelActive();
    22.     }
    23.  
    24.     @Skip
    25.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    26.         ctx.fireChannelInactive();
    27.     }
    28.  
    29.     @Skip
    30.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    31.         ctx.fireChannelRead(msg);
    32.     }
    33.  
    34.     @Skip
    35.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    36.         ctx.fireChannelReadComplete();
    37.     }
    38.  
    39.     @Skip
    40.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    41.         ctx.fireUserEventTriggered(evt);
    42.     }
    43.  
    44.     @Skip
    45.     public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
    46.         ctx.fireChannelWritabilityChanged();
    47.     }
    48.  
    49.     @Skip
    50.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    51.         ctx.fireExceptionCaught(cause);
    52.     }
    53. }

    ChannelOutboundHandlerAdapter

    1. package io.netty.channel;
    2.  
    3. import io.netty.channel.ChannelHandlerMask.Skip;
    4. import java.net.SocketAddress;
    5.  
    6. public class ChannelOutboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelOutboundHandler {
    7.     public ChannelOutboundHandlerAdapter() {
    8.     }
    9.     @Skip
    10.     public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {
    11.         ctx.bind(localAddress, promise);
    12.     }
    13.     @Skip
    14.     public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception {
    15.         ctx.connect(remoteAddress, localAddress, promise);
    16.     }
    17.     @Skip
    18.     public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
    19.         ctx.disconnect(promise);
    20.     }
    21.     @Skip
    22.     public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
    23.         ctx.close(promise);
    24.     }
    25.     @Skip
    26.     public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
    27.         ctx.deregister(promise);
    28.     }
    29.     @Skip
    30.     public void read(ChannelHandlerContext ctx) throws Exception {
    31.         ctx.read();
    32.     }
    33.     @Skip
    34.     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    35.         ctx.write(msg, promise);
    36.     }
    37.     @Skip
    38.     public void flush(ChannelHandlerContext ctx) throws Exception {
    39.         ctx.flush();
    40.     }
    41. }

    适配器的作用

    使用适配器是因为适配器的子类不需要实现父类中的所有方法，按需覆盖适配器的方法即可。

    ChannelPipeline接口

    ChannelPipeline接口设计采用了责任链模式，底层采用双向链表的数据结构，将链上个各个处理器串联起来。

    1.  *                                                 I/O Request
    2.  *                                            via {@link Channel} or
    3.  *                                        {@link ChannelHandlerContext}
    4.  *                                                      |
    5.  *  +---------------------------------------------------+---------------+
    6.  *  |                           ChannelPipeline         |               |
    7.  *  |                                                  \|/              |
    8.  *  |    +---------------------+            +-----------+----------+    |
    9.  *  |    | Inbound Handler  N  |            | Outbound Handler  1  |    |
    10.  *  |    +----------+----------+            +-----------+----------+    |
    11.  *  |              /|\                                  |               |
    12.  *  |               |                                  \|/              |
    13.  *  |    +----------+----------+            +-----------+----------+    |
    14.  *  |    | Inbound Handler N-1 |            | Outbound Handler  2  |    |
    15.  *  |    +----------+----------+            +-----------+----------+    |
    16.  *  |              /|\                                  .               |
    17.  *  |               .                                   .               |
    18.  *  | ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
    19.  *  |        [ method call]                       [method call]         |
    20.  *  |               .                                   .               |
    21.  *  |               .                                  \|/              |
    22.  *  |    +----------+----------+            +-----------+----------+    |
    23.  *  |    | Inbound Handler  2  |            | Outbound Handler M-1 |    |
    24.  *  |    +----------+----------+            +-----------+----------+    |
    25.  *  |              /|\                                  |               |
    26.  *  |               |                                  \|/              |
    27.  *  |    +----------+----------+            +-----------+----------+    |
    28.  *  |    | Inbound Handler  1  |            | Outbound Handler  M  |    |
    29.  *  |    +----------+----------+            +-----------+----------+    |
    30.  *  |              /|\                                  |               |
    31.  *  +---------------+-----------------------------------+---------------+
    32.  *                  |                                  \|/
    33.  *  +---------------+-----------------------------------+---------------+
    34.  *  |               |                                   |               |
    35.  *  |       [ Socket.read() ]                    [ Socket.write() ]     |
    36.  *  |                                                                   |
    37.  *  |  Netty Internal I/O Threads (Transport Implementation)            |
    38.  *  +-------------------------------------------------------------------+
    1. public interface ChannelPipeline
    2.         extends ChannelInboundInvoker, ChannelOutboundInvoker, Iterable> {
    3.  
    4.     ChannelPipeline addFirst(String name, ChannelHandler handler);
    5.  
    6.     ChannelPipeline addFirst(EventExecutorGroup group, String name, ChannelHandler handler);
    7.  
    8.     ChannelPipeline addLast(String name, ChannelHandler handler);
    9.  
    10.     ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler);
    11.  
    12.     ChannelPipeline addBefore(String baseName, String name, ChannelHandler handler);
    13.  
    14.     ChannelPipeline addBefore(EventExecutorGroup group, String baseName, String name, ChannelHandler handler);
    15.  
    16.     ChannelPipeline addAfter(String baseName, String name, ChannelHandler handler);
    17.  
    18.     ChannelPipeline addAfter(EventExecutorGroup group, String baseName, String name, ChannelHandler handler);
    19.  
    20.     ChannelPipeline addFirst(ChannelHandler... handlers);
    21.  
    22.     ChannelPipeline addFirst(EventExecutorGroup group, ChannelHandler... handlers);
    23.  
    24.     ChannelPipeline addLast(ChannelHandler... handlers);
    25.  
    26.     ChannelPipeline addLast(EventExecutorGroup group, ChannelHandler... handlers);
    27.  
    28.     ChannelPipeline remove(ChannelHandler handler);
    29.  
    30.     ChannelHandler remove(String name);
    31.  
    32.     extends ChannelHandler> T remove(Class handlerType);
    33.  
    34.     ChannelHandler removeFirst();
    35.  
    36.     ChannelHandler removeLast();
    37.  
    38.     ChannelPipeline replace(ChannelHandler oldHandler, String newName, ChannelHandler newHandler);
    39.  
    40.     ChannelHandler replace(String oldName, String newName, ChannelHandler newHandler);
    41.  
    42.     extends ChannelHandler> T replace(Class oldHandlerType, String newName,
    43.                                          ChannelHandler newHandler);
    44.  
    45.     ChannelHandler first();
    46.  
    47.     ChannelHandlerContext firstContext();
    48.  
    49.     ChannelHandler last();
    50.  
    51.     ChannelHandlerContext lastContext();
    52.  
    53.     ChannelHandler get(String name);
    54.  
    55.     extends ChannelHandler> T get(Class handlerType);
    56.  
    57.     ChannelHandlerContext context(ChannelHandler handler);
    58.  
    59.     ChannelHandlerContext context(String name);
    60.  
    61.     ChannelHandlerContext context(Class handlerType);
    62.  
    63.     Channel channel();
    64.  
    65.     List names();
    66.  
    67.     Map toMap();
    68.  
    69.     @Override
    70.     ChannelPipeline fireChannelRegistered();
    71.  
    72.     @Override
    73.     ChannelPipeline fireChannelUnregistered();
    74.  
    75.     @Override
    76.     ChannelPipeline fireChannelActive();
    77.  
    78.     @Override
    79.     ChannelPipeline fireChannelInactive();
    80.  
    81.     @Override
    82.     ChannelPipeline fireExceptionCaught(Throwable cause);
    83.  
    84.     @Override
    85.     ChannelPipeline fireUserEventTriggered(Object event);
    86.  
    87.     @Override
    88.     ChannelPipeline fireChannelRead(Object msg);
    89.  
    90.     @Override
    91.     ChannelPipeline fireChannelReadComplete();
    92.  
    93.     @Override
    94.     ChannelPipeline fireChannelWritabilityChanged();
    95.  
    96.     @Override
    97.     ChannelPipeline flush();
    98. }

    创建ChannelPipeline

    ChannelPipeline数据管道是与Channel通道绑定的，一个Channel通道对应一个ChannelPipeline，ChannelPipeline是在Channel初始化时被创建的。

    ChannelHandlerContext接口

    ChannelHandlerContext接口是联系ChannelHandler与其ChannelPipeline之间的纽带。

    每当有ChannelHandler添加到ChannelPipeline中时，都会常见ChannelHandlerContext。ChannelHandlerContext的主要功能是管理它所关联的ChannelHandler和在同一个ChannelPipeline中的其他ChannelHandler之间的交互。例如，ChannelHandlerContext可以通知ChannelPipeline中的下一个ChannelHandler开始执行及动态修改其所属的ChannelPipeline。

  • 相关阅读:
    苹果平板可以用别的电容笔吗？电容笔和Apple pencil区别
    用友GRP-U8 SQL注入漏洞复现
    【最新版配置conda环境】新版pycharm导入新版anaconda环境
    kafka系列七、kafka核心配置（转）
    一小时极速掌握HybridCLR热更新
    绝地求生：想玩以前的老地图
    MD中 面料的物理属性参数
    keras实现深度神经网络,keras实现卷积神经网络
    python使用SMTP发送邮件
    什么是Rebex Total Pack for.NET?
  • 原文地址：https://blog.csdn.net/lovelovelovelovelo/article/details/127724712