1: /**
 2:  * 通道处理器
 3:  */
 4: private final ChannelHandler handler;
 5: /**
 6:  * URL
 7:  */
 8: private volatile URL url;
 9: /**
10:  * 正在关闭
11:  *
12:  * {@link #startClose()}
13:  */
14: // closing closed means the process is being closed and close is finished
15: private volatile boolean closing;
16: /**
17:  * 关闭完成
18:  *
19:  * {@link #close()}
20:  */
21: private volatile boolean closed;
22: 
23: public AbstractPeer(URL url, ChannelHandler handler) {
24:     if (url == null) {
25:         throw new IllegalArgumentException("url == null");
26:     }
27:     if (handler == null) {
28:         throw new IllegalArgumentException("handler == null");
29:     }
30:     this.url = url;
31:     this.handler = handler;
32: }

@Override
public void send(Object message) throws RemotingException {
    send(message, url.getParameter(Constants.SENT_KEY, false));
}

 1: /**
 2:  * 编解码器
 3:  */
 4: private Codec2 codec;
 5: /**
 6:  * 超时时间
 7:  */
 8: private int timeout;
 9: /**
10:  * 连接超时时间
11:  */
12: private int connectTimeout;
13: 
14: public AbstractEndpoint(URL url, ChannelHandler handler) {
15:     super(url, handler);
16:     this.codec = getChannelCodec(url);
17:     this.timeout = url.getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
18:     this.connectTimeout = url.getPositiveParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT);
19: }

protected abstract void doOpen() throws Throwable;
protected abstract void doClose() throws Throwable;

protected abstract void doConnect() throws Throwable;
protected abstract void doDisConnect() throws Throwable;

protected abstract Channel getChannel();

 1: /**
 2:  * 重连定时任务执行器
 3:  */
 4: private static final ScheduledThreadPoolExecutor reconnectExecutorService = new ScheduledThreadPoolExecutor(2, new NamedThreadFactory("DubboClientReconnectTimer", true));
 5: /**
 6:  * 发送消息时，若断开，是否重连
 7:  */
 8: private final boolean send_reconnect;
 9: /**
10:  * 重连 warning 的间隔.(waring多少次之后，warning一次) //for test
11:  */
12: // reconnect warning period. Reconnect warning interval (log warning after how many times) //for test
13: private final int reconnect_warning_period;
14: /**
15:  * 关闭超时时间
16:  */
17: private final long shutdown_timeout;
18: /**
19:  * 线程池
20:  *
21:  * 在调用 {@link #wrapChannelHandler(URL, ChannelHandler)} 时，会调用 {@link com.alibaba.dubbo.remoting.transport.dispatcher.WrappedChannelHandler} 创建
22:  */
23: protected volatile ExecutorService executor;
24: 
25: public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
26:     super(url, handler);
27:     // 从 URL 中，获得重连相关配置项
28:     send_reconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, false);
29:     shutdown_timeout = url.getParameter(Constants.SHUTDOWN_TIMEOUT_KEY, Constants.DEFAULT_SHUTDOWN_TIMEOUT);
30:     // The default reconnection interval is 2s, 1800 means warning interval is 1 hour.
31:     reconnect_warning_period = url.getParameter("reconnect.waring.period", 1800);
32: 
33:     // 初始化客户端
34:     try {
35:         doOpen();
36:     } catch (Throwable t) {
37:         close(); // 失败，则关闭
38:         throw new RemotingException(url.toInetSocketAddress(), null,
39:                 "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
40:                         + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
41:     }
42: 
43:     // 连接服务器
44:     try {
45:         // connect.
46:         connect();
47:         if (logger.isInfoEnabled()) {
48:             logger.info("Start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() + " connect to the server " + getRemoteAddress());
49:         }
50:     } catch (RemotingException t) {
51:         if (url.getParameter(Constants.CHECK_KEY, true)) {
52:             close(); // 失败，则关闭
53:             throw t;
54:         } else {
55:             logger.warn("Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
56:                     + " connect to the server " + getRemoteAddress() + " (check == false, ignore and retry later!), cause: " + t.getMessage(), t);
57:         }
58:     } catch (Throwable t) {
59:         close(); // 失败，则关闭
60:         throw new RemotingException(url.toInetSocketAddress(), null,
61:                 "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
62:                         + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
63:     }
64: 
65:     // 获得线程池
66:     executor = (ExecutorService) ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension()
67:             .get(Constants.CONSUMER_SIDE, Integer.toString(url.getPort()));
68:     ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension()
69:             .remove(Constants.CONSUMER_SIDE, Integer.toString(url.getPort()));
70: }

/**
 * 连接锁，用于实现发起连接和断开连接互斥，避免并发。
 */
private final Lock connectLock = new ReentrantLock();

  1: protected void connect() throws RemotingException {
  2:     // 获得锁
  3:     connectLock.lock();
  4:     try {
  5:         // 已连接，
  6:         if (isConnected()) {
  7:             return;
  8:         }
  9:         // 初始化重连线程
 10:         initConnectStatusCheckCommand();
 11:         // 执行连接
 12:         doConnect();
 13:         // 连接失败，抛出异常
 14:         if (!isConnected()) {
 15:             throw new RemotingException(this, "Failed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " "
 16:                     + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion()
 17:                     + ", cause: Connect wait timeout: " + getTimeout() + "ms.");
 18:         // 连接成功，打印日志
 19:         } else {
 20:             if (logger.isInfoEnabled()) {
 21:                 logger.info("Successed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " "
 22:                         + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion()
 23:                         + ", channel is " + this.getChannel());
 24:             }
 25:         }
 26:         // 设置重连次数归零
 27:         reconnect_count.set(0);
 28:         // 设置未打印过错误日志
 29:         reconnect_error_log_flag.set(false);
 30:     } catch (RemotingException e) {
 31:         throw e;
 32:     } catch (Throwable e) {
 33:         throw new RemotingException(this, "Failed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " "
 34:                 + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion()
 35:                 + ", cause: " + e.getMessage(), e);
 36:     } finally {
 37:         // 释放锁
 38:         connectLock.unlock();
 39:     }
 40: }

/**
 * 重连次数
 */
private final AtomicInteger reconnect_count = new AtomicInteger(0);
/**
 * 重连时，是否已经打印过错误日志。
 */
// Reconnection error log has been called before?
private final AtomicBoolean reconnect_error_log_flag = new AtomicBoolean(false);
/**
 * 重连执行任务 Future
 */
private volatile ScheduledFuture<?> reconnectExecutorFuture = null;
/**
 * 最后成功连接时间
 */
// the last successed connected time
private long lastConnectedTime = System.currentTimeMillis();
    
  1: private synchronized void initConnectStatusCheckCommand() {
  2:     //reconnect=false to close reconnect
  3:     // 获得获得重连频率，默认开启。
  4:     int reconnect = getReconnectParam(getUrl());
  5:     // 若开启重连功能，创建重连线程
  6:     if (reconnect > 0 && (reconnectExecutorFuture == null || reconnectExecutorFuture.isCancelled())) {
  7:         // 创建 Runnable 对象
  8:         Runnable connectStatusCheckCommand = new Runnable() {
  9:             public void run() {
 10:                 try {
 11:                     // 未连接，重连
 12:                     if (!isConnected()) {
 13:                         connect();
 14:                     // 已连接，记录最后连接时间
 15:                     } else {
 16:                         lastConnectedTime = System.currentTimeMillis();
 17:                     }
 18:                 } catch (Throwable t) {
 19:                     // 超过一定时间未连接上，才打印异常日志。并且，仅打印一次。默认，15 分钟。
 20:                     String errorMsg = "client reconnect to " + getUrl().getAddress() + " find error . url: " + getUrl();
 21:                     // wait registry sync provider list
 22:                     if (System.currentTimeMillis() - lastConnectedTime > shutdown_timeout) {
 23:                         if (!reconnect_error_log_flag.get()) {
 24:                             reconnect_error_log_flag.set(true);
 25:                             logger.error(errorMsg, t);
 26:                             return;
 27:                         }
 28:                     }
 29:                     // 每一定次发现未重连，才打印告警日志。默认，1800 次，1 小时。
 30:                     if (reconnect_count.getAndIncrement() % reconnect_warning_period == 0) {
 31:                         logger.warn(errorMsg, t);
 32:                     }
 33:                 }
 34:             }
 35:         };
 36:         // 发起定时任务
 37:         reconnectExecutorFuture = reconnectExecutorService.scheduleWithFixedDelay(connectStatusCheckCommand, reconnect, reconnect, TimeUnit.MILLISECONDS);
 38:     }
 39: }

@Override
public void send(Object message, boolean sent) throws RemotingException {
    // 未连接时，开启重连功能，则先发起连接
    if (send_reconnect && !isConnected()) {
        connect();
    }
    // 发送消息
    Channel channel = getChannel();
    //TODO Can the value returned by getChannel() be null? need improvement.
    if (channel == null || !channel.isConnected()) {
        throw new RemotingException(this, "message can not send, because channel is closed . url:" + getUrl());
    }
    channel.send(message, sent);
}

// Constants.java
public static final String DEFAULT_CLIENT_THREADPOOL = "cached";

protected static final String CLIENT_THREAD_POOL_NAME = "DubboClientHandler";

  1: /**
  2:  * 包装通道处理器
  3:  *
  4:  * @param url URL
  5:  * @param handler 被包装的通道处理器
  6:  * @return 包装后的通道处理器
  7:  */
  8: protected static ChannelHandler wrapChannelHandler(URL url, ChannelHandler handler) {
  9:     // 设置线程名
 10:     url = ExecutorUtil.setThreadName(url, CLIENT_THREAD_POOL_NAME);
 11:     // 设置使用的线程池类型
 12:     url = url.addParameterIfAbsent(Constants.THREADPOOL_KEY, Constants.DEFAULT_CLIENT_THREADPOOL);
 13:     // 包装通道处理器
 14:     return ChannelHandlers.wrap(handler, url);
 15: }

protected abstract void doOpen() throws Throwable;

protected abstract void doClose() throws Throwable;

 1: /**
 2:  * 线程池
 3:  */
 4: ExecutorService executor;
 5: /**
 6:  * 服务地址
 7:  */
 8: private InetSocketAddress localAddress;
 9: /**
10:  * 绑定地址
11:  */
12: private InetSocketAddress bindAddress;
13: /**
14:  * 服务器最大可接受连接数
15:  */
16: private int accepts;
17: /**
18:  * 空闲超时时间，单位：毫秒
19:  */
20: private int idleTimeout; //600 seconds
21: 
22: public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
23:     super(url, handler);
24:     // 服务地址
25:     localAddress = getUrl().toInetSocketAddress();
26:     // 绑定地址
27:     String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost());
28:     int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort());
29:     if (url.getParameter(Constants.ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) {
30:         bindIp = NetUtils.ANYHOST;
31:     }
32:     bindAddress = new InetSocketAddress(bindIp, bindPort);
33:     // 服务器最大可接受连接数
34:     this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS);
35:     // 空闲超时时间
36:     this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT);
37: 
38:     // 开启服务器
39:     try {
40:         doOpen();
41:         if (logger.isInfoEnabled()) {
42:             logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
43:         }
44:     } catch (Throwable t) {
45:         throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName()
46:                 + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
47:     }
48: 
49:     // 获得线程池
50:     //fixme replace this with better method
51:     DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
52:     executor = (ExecutorService) dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY, Integer.toString(url.getPort()));
53: }

@Override
public void connected(Channel ch) throws RemotingException {
    // If the server has entered the shutdown process, reject any new connection
    if (this.isClosing() || this.isClosed()) {
        logger.warn("Close new channel " + ch + ", cause: server is closing or has been closed. For example, receive a new connect request while in shutdown process.");
        ch.close();
        return;
    }

    // 超过上限，关闭新的链接
    Collection<Channel> channels = getChannels();
    if (accepts > 0 && channels.size() > accepts) {
        logger.error("Close channel " + ch + ", cause: The server " + ch.getLocalAddress() + " connections greater than max config " + accepts);
        ch.close(); // 关闭新的链接
        return;
    }
    // 连接
    super.connected(ch);
}

@Override
public void send(Object message, boolean sent) throws RemotingException {
    // 获得所有的客户端的通道
    Collection<Channel> channels = getChannels();
    // 群发消息
    for (Channel channel : channels) {
        if (channel.isConnected()) {
            channel.send(message, sent);
        }
    }
}

@Override
public void send(Object message, boolean sent) throws RemotingException {
    if (isClosed()) {
        throw new RemotingException(this, "Failed to send message "
                + (message == null ? "" : message.getClass().getName()) + ":" + message
                + ", cause: Channel closed. channel: " + getLocalAddress() + " -> " + getRemoteAddress());
    }
}

@Override public void connected(Channel channel) throws RemotingException { }

@Override public void disconnected(Channel channel) throws RemotingException { }

@Override public void sent(Channel channel, Object message) { }

@Override public void received(Channel channel, Object message) throws RemotingException { }

@Override public void caught(Channel channel, Throwable exception) throws RemotingException { }

public void received(Channel channel, Object message) {
    for (ChannelHandler listener : channelHandlers) {
        try {
            listener.received(channel, message);
        } catch (Throwable t) {
            logger.error(t.getMessage(), t);
        }
    }
}

ChannelHandler getHandler();

 1: @Override
 2: public void received(Channel channel, Object message) throws RemotingException {
 3:     if (message instanceof Decodeable) {
 4:         decode(message);
 5:     }
 6: 
 7:     if (message instanceof Request) {
 8:         decode(((Request) message).getData());
 9:     }
10: 
11:     if (message instanceof Response) {
12:         decode(((Response) message).getResult());
13:     }
14: 
15:     handler.received(channel, message);
16: }

 1: private void decode(Object message) {
 2:     if (message != null && message instanceof Decodeable) {
 3:         try {
 4:             ((Decodeable) message).decode(); // 解析消息
 5:             if (log.isDebugEnabled()) {
 6:                 log.debug(new StringBuilder(32).append("Decode decodeable message ").append(message.getClass().getName()).toString());
 7:             }
 8:         } catch (Throwable e) {
 9:             if (log.isWarnEnabled()) {
10:                 log.warn(new StringBuilder(32).append("Call Decodeable.decode failed: ").append(e.getMessage()).toString(), e);
11:             }
12:         } // ~ end of catch
13:     } // ~ end of if
14: } // ~ end of method decode

 1: @Override
 2: public void received(Channel channel, Object message) throws RemotingException {
 3:     if (message instanceof MultiMessage) { // 多消息
 4:         MultiMessage list = (MultiMessage) message;
 5:         for (Object obj : list) {
 6:             handler.received(channel, obj);
 7:         }
 8:     } else {
 9:         handler.received(channel, message);
10:     }
11: }

executor.execute(new Runnable() {
    handler.received(channel, message)
});

 1: /**
 2:  * 单例
 3:  */
 4: private static ChannelHandlers INSTANCE = new ChannelHandlers();
 5: 
 6: public static ChannelHandler wrap(ChannelHandler handler, URL url) {
 7:     return ChannelHandlers.getInstance().wrapInternal(handler, url);
 8: }
 9: 
10: protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
11:     return new MultiMessageHandler(
12:             new HeartbeatHandler(
13:                     ExtensionLoader.getExtensionLoader(Dispatcher.class).getAdaptiveExtension().dispatch(handler, url)
14:             )
15:     );
16: }

public class AllDispatcher implements Dispatcher {

    public static final String NAME = "all";

    public ChannelHandler dispatch(ChannelHandler handler, URL url) {
        return new AllChannelHandler(handler, url);
    }

}

@Override
public void connected(Channel channel) throws RemotingException {
    ExecutorService cexecutor = getExecutorService();
    try {
        cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED));
    } catch (Throwable t) {
        throw new ExecutionException("connect event", channel, getClass() + " error when process connected event .", t);
    }
}

 1: /**
 2:  * 线程池
 3:  */
 4: protected final ExecutorService executor;
 5: /**
 6:  * 通道处理器
 7:  */
 8: protected final ChannelHandler handler;
 9: /**
10:  * URL
11:  */
12: protected final URL url;
13: 
14: public WrappedChannelHandler(ChannelHandler handler, URL url) {
15:     this.handler = handler;
16:     this.url = url;
17: 
18:     // 创建线程池
19:     executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);
20: 
21:     // 添加线程池到 DataStore 中
22:     String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY;
23:     if (Constants.CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(Constants.SIDE_KEY))) {
24:         componentKey = Constants.CONSUMER_SIDE;
25:     }
26:     DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
27:     dataStore.put(componentKey, Integer.toString(url.getPort()), executor);
28: }

protected static final ExecutorService SHARED_EXECUTOR = Executors.newCachedThreadPool(new NamedThreadFactory("DubboSharedHandler", true));

/**
 * 序列化对象集合
 * key：序列化类型编号 {@link Serialization#getContentTypeId()}
 */
private static Map<Byte, Serialization> ID_SERIALIZATION_MAP = new HashMap<Byte, Serialization>();
/**
 * 序列化名集合
 * key：序列化类型编号 {@link Serialization#getContentTypeId()}
 * value: 序列化拓展名
 */
private static Map<Byte, String> ID_SERIALIZATIONNAME_MAP = new HashMap<Byte, String>();

static {
    // 基于 Dubbo SPI ，初始化
    Set<String> supportedExtensions = ExtensionLoader.getExtensionLoader(Serialization.class).getSupportedExtensions();
    for (String name : supportedExtensions) {
        Serialization serialization = ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(name);
        byte idByte = serialization.getContentTypeId();
        if (ID_SERIALIZATION_MAP.containsKey(idByte)) {
            logger.error("Serialization extension " + serialization.getClass().getName()
                    + " has duplicate id to Serialization extension "
                    + ID_SERIALIZATION_MAP.get(idByte).getClass().getName()
                    + ", ignore this Serialization extension");
            continue;
        }
        ID_SERIALIZATION_MAP.put(idByte, serialization);
        ID_SERIALIZATIONNAME_MAP.put(idByte, name);
    }
}

public static Serialization getSerialization(URL url, Byte id) throws IOException {
    Serialization serialization = getSerializationById(id);
    String serializationName = url.getParameter(Constants.SERIALIZATION_KEY, Constants.DEFAULT_REMOTING_SERIALIZATION); // 默认，hessian2
    // 出于安全的目的，针对 JDK 的序列化方式（对应编号为 3、4、7），检查连接到服务器的 URL 和实际传输的数据，协议是否一致。
    // https://github.com/apache/incubator-dubbo/issues/1138
    // Check if "serialization id" passed from network matches the id on this side(only take effect for JDK serialization), for security purpose.
    if (serialization == null
            || ((id == 3 || id == 7 || id == 4) && !(serializationName.equals(ID_SERIALIZATIONNAME_MAP.get(id))))) {
        throw new IOException("Unexpected serialization id:" + id + " received from network, please check if the peer send the right id.");
    }
    return serialization;
}

1: protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException {
2:     // check magic number.
3:     if (readable > 0 && header[0] != MAGIC_HIGH
4:             || readable > 1 && header[1] != MAGIC_LOW) {
5:         // ... 省略
6:         return super.decode(channel, buffer, readable, header);
7:     }
8:    
9:      // ... 省略
10:    
11:     return decodeBody(channel, is, header);
12: }

 1: @Override
 2: public void encode(Channel channel, ChannelBuffer buffer, Object message) throws IOException {
 3:     // 获得反序列化的 ObjectOutput 对象
 4:     OutputStream output = new ChannelBufferOutputStream(buffer);
 5:     ObjectOutput objectOutput = getSerialization(channel).serialize(channel.getUrl(), output);
 6:     // 写入 ObjectOutput
 7:     encodeData(channel, objectOutput, message);
 8:     objectOutput.flushBuffer();
 9:     // 释放
10:     if (objectOutput instanceof Cleanable) {
11:         ((Cleanable) objectOutput).cleanup();
12:     }
13: }

// Netty.java
new ChannelInitializer<NioSocketChannel>() {
    @Override
    protected void initChannel(NioSocketChannel ch) {
        NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
        ch.pipeline().addLast("decoder", adapter.getDecoder()) // 解码
                    .addLast("encoder", adapter.getEncoder())  // 解码
                    .addLast("handler", nettyServerHandler); // 处理器
    }
}