23、集群扩展：发送请求遇到服务不可用，怎么办

protected void checkInvokers(List<Invoker<T>> invokers, Invocation invocation) {
    // 检查传入的 invokers 服务提供者列表，若集合为空，则会抛出无提供者异常
    if (CollectionUtils.isEmpty(invokers)) {
        // 抛出的 RpcException 异常信息中，会有 No provider available 明显的关键字
        throw new RpcException(RpcException.NO_INVOKER_AVAILABLE_AFTER_FILTER, "Failed to invoke the method "
            + invocation.getMethodName() + " in the service " + getInterface().getName()
            + ". No provider available for the service " + getDirectory().getConsumerUrl().getServiceKey()
            + " from registry " + getDirectory().getUrl().getAddress()
            + " on the consumer " + NetUtils.getLocalHost()
            + " using the dubbo version " + Version.getVersion()
            + ". Please check if the providers have been started and registered.");
    }
}

public class TransferClusterInvoker<T> extends FailoverClusterInvoker<T> {
    // 按照父类 FailoverClusterInvoker 要求创建的构造方法
    public TransferClusterInvoker(Directory<T> directory) {
        super(directory);
    }
    // 重写父类 doInvoke 发起远程调用的接口
    @Override
    public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        try {
            // 先完全按照父类的业务逻辑调用处理，无异常则直接将结果返回
            return super.doInvoke(invocation, invokers, loadbalance);
        } catch (RpcException e) {
            // 这里就进入了 RpcException 处理逻辑
            
            // 当调用发现无提供者异常描述信息时则向转发服务发起调用
            if (e.getMessage().toLowerCase().contains("no provider available")){
                // TODO 从 invocation 中拿到所有的参数，然后再处理调用转发服务的逻辑
                return doTransferInvoke(invocation);
            }
            // 如果不是无提供者异常，则不做任何处理，异常该怎么抛就怎么抛
            throw e;
        }
    }
}

public class TransferCluster implements Cluster {
    // 返回自定义的 Invoker 调用器
    @Override
    public <T> Invoker<T> join(Directory<T> directory, boolean buildFilterChain) throws RpcException {
        return new TransferClusterInvoker<T>(directory);
    }
}

dubbo.consumer.cluster=transfer

///////////////////////////////////////////////////                  
// 1、org.apache.dubbo.remoting.exchange.codec.ExchangeCodec#encodeResponse
// 2、将 Response 对象按照 Dubbo 协议格式编码为字节流
// 3、重点关注我在代码中描述的 ①②③④⑤⑥⑦ 几个关键位置
///////////////////////////////////////////////////
// header length.
protected static final int HEADER_LENGTH = 16;
// magic header.
protected static final short MAGIC = (short) 0xdabb;
protected static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0];
protected static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC)[1];
// message flag.
protected static final byte FLAG_REQUEST = (byte) 0x80;
protected static final byte FLAG_TWOWAY = (byte) 0x40;
protected static final byte FLAG_EVENT = (byte) 0x20;
protected static final int SERIALIZATION_MASK = 0x1f;

// 将 Response 对象按照 Dubbo 协议格式编码为字节流
protected void encodeResponse(Channel channel, ChannelBuffer buffer, Response res) throws IOException {
    int savedWriteIndex = buffer.writerIndex();
    try {
        Serialization serialization = getSerialization(channel, res);
        // header.
        // ① 针对 header 字节数组赋值一个 0xdabb 魔术值
        byte[] header = new byte[HEADER_LENGTH];
        // set magic number.
        Bytes.short2bytes(MAGIC, header);
        
        // set request and serialization flag.
        // ② 设置序列化方式，序列化方式从 channel 的 url 取出 serialization 对应的参数值，
        header[2] = serialization.getContentTypeId();
        if (res.isHeartbeat()) {
            // ③ 设置请求类型，根据 mTwoWay、mEvent 来决定是怎样的请求类型
            header[2] |= FLAG_EVENT;
        }
        
        // set response status.
        // ④ 设置响应码，这里因为是响应成功，因此 status = 20
        //
        // status 的值，在源码中有如下这些值：
        // /** ok. 响应成功 */
        // public static final byte OK = 20;
        // /** client side timeout. 消费方超时 */
        // public static final byte CLIENT_TIMEOUT = 30;
        // /** server side timeout. 提供方超时 */
        // public static final byte SERVER_TIMEOUT = 31;
        // /** channel inactive, directly return the unfinished requests. 通道channel被关闭不可用了 */
        // public static final byte CHANNEL_INACTIVE = 35;
        // /** request format error. 请求数据格式错误 */
        // public static final byte BAD_REQUEST = 40;
        // /** response format error. 响应数据格式错误 */
        // public static final byte BAD_RESPONSE = 50;
        // /** service not found. 提供方服务找不到 */
        // public static final byte SERVICE_NOT_FOUND = 60;
        // /** service error. 提供方错误 */
        // public static final byte SERVICE_ERROR = 70;
        // /** internal server error. 提供方内部错误 */
        // public static final byte SERVER_ERROR = 80;
        // /** internal server error. 消费方内部错误 */
        // public static final byte CLIENT_ERROR = 90;
        // /** server side threadpool exhausted and quick return. 线程池满拒绝请求 */
        // public static final byte SERVER_THREADPOOL_EXHAUSTED_ERROR = 100;        
        byte status = res.getStatus();
        header[3] = status;
        
        // ⑤ 设置响应ID，该ID是请求发送过来的ID
        // set request id.
        Bytes.long2bytes(res.getId(), header, 4);
        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
        ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
        
        // ⑥ 构建一个输出流，根据 mEvent 的值来将 mData 进行序列化转为字节数组
        // encode response data or error message.
        if (status == Response.OK) {
            if(res.isHeartbeat()){
                // heartbeat response data is always null
                bos.write(CodecSupport.getNullBytesOf(serialization));
            }else {
                ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
                if (res.isEvent()) {
                    encodeEventData(channel, out, res.getResult());
                } else {
                    encodeResponseData(channel, out, res.getResult(), res.getVersion());
                }
                out.flushBuffer();
                if (out instanceof Cleanable) {
                    ((Cleanable) out).cleanup();
                }
            }
        } else {
            ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
            out.writeUTF(res.getErrorMessage());
            out.flushBuffer();
            if (out instanceof Cleanable) {
                ((Cleanable) out).cleanup();
            }
        }
        bos.flush();
        bos.close();
        int len = bos.writtenBytes();
        checkPayload(channel, len);
        Bytes.int2bytes(len, header, 12);
        // write
        buffer.writerIndex(savedWriteIndex);
        buffer.writeBytes(header); // write header.
        
        // ⑦ 最终将序列化出来的字节数组的长度填充至报文体长度位置
        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
    } catch (Throwable t) {
        // 下面的逻辑都是异常处理，主要想办法赋值 status、errorMessage 字段
        // 将错误信息体现出来，告诉消费方
        // clear buffer
        buffer.writerIndex(savedWriteIndex);
        // send error message to Consumer, otherwise, Consumer will wait till timeout.
        if (!res.isEvent() && res.getStatus() != Response.BAD_RESPONSE) {
            Response r = new Response(res.getId(), res.getVersion());
            r.setStatus(Response.BAD_RESPONSE);
            if (t instanceof ExceedPayloadLimitException) {
                logger.warn(t.getMessage(), t);
                try {
                    r.setErrorMessage(t.getMessage());
                    channel.send(r);
                    return;
                } catch (RemotingException e) {
                    logger.warn("Failed to send bad_response info back: " + t.getMessage() + ", cause: " + e.getMessage(), e);
                }
            } else {
                // FIXME log error message in Codec and handle in caught() of IoHanndler?
                logger.warn("Fail to encode response: " + res + ", send bad_response info instead, cause: " + t.getMessage(), t);
                try {
                    r.setErrorMessage("Failed to send response: " + res + ", cause: " + StringUtils.toString(t));
                    channel.send(r);
                    return;
                } catch (RemotingException e) {
                    logger.warn("Failed to send bad_response info back: " + res + ", cause: " + e.getMessage(), e);
                }
            }
        }
        
        // Rethrow exception
        if (t instanceof IOException) {
            throw (IOException) t;
        } else if (t instanceof RuntimeException) {
            throw (RuntimeException) t;
        } else if (t instanceof Error) {
            throw (Error) t;
        } else {
            throw new RuntimeException(t.getMessage(), t);
        }
    }
}