自己動手從0開始實現(xiàn)一個分布式RPC框架

+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+ |  BYTE  |        |        |        |        |        |        |             ........ +--------------------------------------------+--------+-----------------+--------+--------+--------+--------+--------+--------+-----------------+ |  magic | version|  type  |           content lenth           |                   content byte[]                                        |        | +--------+-----------------------------------------------------------------------------------------+--------------------------------------------+

    @Override    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory)        throws BeansException {        this.beanFactory = beanFactory;        postProcessRpcConsumerBeanFactory(beanFactory, (BeanDefinitionRegistry)beanFactory);    }
    private void postProcessRpcConsumerBeanFactory(ConfigurableListableBeanFactory beanFactory, BeanDefinitionRegistry beanDefinitionRegistry) {        String[] beanDefinitionNames = beanFactory.getBeanDefinitionNames();        int len = beanDefinitionNames.length;        for (int i = 0; i < len; i++) {            String beanDefinitionName = beanDefinitionNames[i];            BeanDefinition beanDefinition = beanFactory.getBeanDefinition(beanDefinitionName);            String beanClassName = beanDefinition.getBeanClassName();            if (beanClassName != null) {                Class<?> clazz = ClassUtils.resolveClassName(beanClassName, classLoader);                ReflectionUtils.doWithFields(clazz, new FieldCallback() {                    @Override                    public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException {                        parseField(field);                    }                });            }
        }
        Iterator<Entry<String, BeanDefinition>> it = beanDefinitions.entrySet().iterator();        while (it.hasNext()) {            Entry<String, BeanDefinition> entry = it.next();            if (context.containsBean(entry.getKey())) {                throw new IllegalArgumentException("Spring context already has a bean named " + entry.getKey());            }            beanDefinitionRegistry.registerBeanDefinition(entry.getKey(), entry.getValue());            log.info("register OrcRpcConsumerBean definition: {}", entry.getKey());        }
    }
    private void parseField(Field field) {        // 獲取所有OrcRpcConsumer注解        OrcRpcConsumer orcRpcConsumer = field.getAnnotation(OrcRpcConsumer.class);        if (orcRpcConsumer != null) {            // 使用field的類型和OrcRpcConsumer注解一起生成BeanDefinition            OrcRpcConsumerBeanDefinitionBuilder beanDefinitionBuilder = new OrcRpcConsumerBeanDefinitionBuilder(field.getType(), orcRpcConsumer);            BeanDefinition beanDefinition = beanDefinitionBuilder.build();            beanDefinitions.put(field.getName(), beanDefinition);        }    }

public class JdkProxyFactory implements ProxyFactory{
    @Override    public Object getProxy(ServiceMetadata serviceMetadata) {        return Proxy            .newProxyInstance(serviceMetadata.getClazz().getClassLoader(), new Class[] {serviceMetadata.getClazz()},                new ClientInvocationHandler(serviceMetadata));    }
    private class ClientInvocationHandler implements InvocationHandler {
        private ServiceMetadata serviceMetadata;
        public ClientInvocationHandler(ServiceMetadata serviceMetadata) {            this.serviceMetadata = serviceMetadata;        }
        @Override        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {            String serviceId = ServiceUtils.getServiceId(serviceMetadata);            // 通過負載均衡器選取一個服務提供方地址            ServiceURL service = InvocationServiceSelector.select(serviceMetadata);
            OrcRpcRequest request = new OrcRpcRequest();            request.setMethod(method.getName());            request.setParameterTypes(method.getParameterTypes());            request.setParameters(args);            request.setRequestId(UUID.randomUUID().toString());            request.setServiceId(serviceId);
            OrcRpcResponse response = InvocationClientContainer.getInvocationClient(service.getServerNet()).invoke(request, service);            if (response.getStatus() == RpcStatusEnum.SUCCESS) {                return response.getData();            } else if (response.getException() != null) {                throw new OrcRpcException(response.getException().getMessage());            } else {                throw new OrcRpcException(response.getStatus().name());            }        }    }}

+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+ |  BYTE  |        |        |        |        |        |        |             ........ +--------------------------------------------+--------+-----------------+--------+--------+--------+--------+--------+--------+-----------------+ |  magic | version|  type  |           content lenth           |                   content byte[]                                        |        | +--------+-----------------------------------------------------------------------------------------+--------------------------------------------+

@Overrideprotected void encode(ChannelHandlerContext channelHandlerContext, ProtocolMsg protocolMsg, ByteBuf byteBuf)    throws Exception {    // 寫入?yún)f(xié)議頭    byteBuf.writeByte(ProtocolConstant.MAGIC);    // 寫入版本    byteBuf.writeByte(ProtocolConstant.DEFAULT_VERSION);    // 寫入請求類型    byteBuf.writeByte(protocolMsg.getMsgType());    // 寫入消息長度    byteBuf.writeInt(protocolMsg.getContent().length);    // 寫入消息內(nèi)容    byteBuf.writeBytes(protocolMsg.getContent());}

/** * 協(xié)議開始的標志 magic + version + type + length 占據(jù)7個字節(jié) */public final int BASE_LENGTH = 7;
@Overrideprotected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list)    throws Exception {    // 可讀字節(jié)小于基本長度，無法解析出payload長度，返回    if (byteBuf.readableBytes() < BASE_LENGTH) {        return;    }    // 記錄包頭開始的index    int beginIndex;    while (true) {        // 記錄包頭開始的index        beginIndex = byteBuf.readerIndex();        // 標記包頭開始的index        byteBuf.markReaderIndex();        // 讀到了協(xié)議頭魔數(shù)，結束循環(huán)        if (byteBuf.readByte() == ProtocolConstant.MAGIC) {            break;        }        // 未讀到包頭，略過一個字節(jié)        // 每次略過一個字節(jié)，去讀取包頭信息的開始標記        byteBuf.resetReaderIndex();        byteBuf.readByte();
        /**         * 當略過，一個字節(jié)之后，數(shù)據(jù)包的長度，又變得不滿足         * 此時結束。等待后面的數(shù)據(jù)到達         */        if (byteBuf.readableBytes() < BASE_LENGTH) {            return;        }    }    // 讀取版本號    byte version = byteBuf.readByte();    // 讀取消息類型    byte type = byteBuf.readByte();    // 讀取消息長度    int length = byteBuf.readInt();    // 判斷本包是否完整    if (byteBuf.readableBytes() < length) {        // 還原讀指針        byteBuf.readerIndex(beginIndex);        return;    }    byte[] data = new byte[length];    byteBuf.readBytes(data);
    ProtocolMsg msg = new ProtocolMsg();    msg.setMsgType(type);    msg.setContent(data);    list.add(msg);}