手撸RPC框架 完善消费者发送消息基础功能

大家好，我是小趴菜，接下来我会从0到1手写一个RPC框架，该专题包括以下专题，有兴趣的小伙伴就跟着我一起学习吧

本章源码地址：gitee.com/baojh123/se…

自定义注解 -> opt-01
服务提供者收发消息基础实现 -> opt-01
自定义网络传输协议的实现 -> opt-02
自定义编解码实现 -> opt-03
服务提供者调用真实方法实现 -> opt-04
完善服务消费者发送消息基础功能 -> opt-05
注册中心基础功能实现 -> opt-06
服务提供者整合注册中心 -> opt-07
服务消费者整合注册中心 -> opt-08
完善服务消费者接收响应结果 -> opt-09
服务消费者，服务提供者整合SpringBoot -> opt-10
动态代理屏蔽RPC服务调用底层细节 -> opt-10
SPI机制基础功能实现 -> opt-11
SPI机制扩展随机负载均衡策略 -> opt-12
SPI机制扩展轮询负载均衡策略 -> opt-13
SPI机制扩展JDK序列化 -> opt-14
SPI机制扩展JSON序列化 -> opt-15
SPI机制扩展protustuff序列化 -> opt-16

前言

在之前的章节中，我们已经实现了服务提供者的收发消息功能，并且服务提供者能够调用真实方法并返回响应结果。

但是我们发现，发送消息是在服务消费者注册的时候，发送了一条消息，这显然不符合我们的需求。

这一章我们完善下服务消费者发送消息的基础功能

实现

修改服务消费者：com.xpc.rpc.consumer.RpcConsumer

package com.xpc.rpc.consumer;

import com.xpc.rpc.codec.RpcDecoder;
import com.xpc.rpc.codec.RpcEncoder;
import com.xpc.rpc.consumer.handler.RpcConsumerHandler;
import com.xpc.rpc.protocol.ProtocolMessage;
import com.xpc.rpc.protocol.request.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class RpcConsumer {

     private static final Logger LOGGER = LoggerFactory.getLogger(RpcConsumer.class);

     private Bootstrap bootstrap;

     private EventLoopGroup eventLoopGroup;

     private Map handlerMap = new ConcurrentHashMap();

     public RpcConsumer() {
         eventLoopGroup = new NioEventLoopGroup();
         bootstrap = new Bootstrap();

         bootstrap.group(eventLoopGroup)
                 .channel(NioSocketChannel.class)
                 .handler(new ChannelInitializer() {
                     @Override
                     protected void initChannel(SocketChannel channel) throws Exception {
                         ChannelPipeline pipeline = channel.pipeline();
                         //我们自己的编解码器
                         pipeline.addLast(new RpcDecoder());
                         pipeline.addLast(new RpcEncoder());
                         pipeline.addLast(new RpcConsumerHandler());
                     }
                 });
     }

     public void sendRequest(ProtocolMessage requestProtocolMessage) {
         //先写死，后续会通过注册中心获取
         String host = "127.0.0.1";
         int port = 21778;
         RpcRequest request = requestProtocolMessage.getT();
         String key = buildHandlerMapKey(request.getClassName(), request.getMethodName());
         RpcConsumerHandler consumerHandler;
         if(handlerMap.containsKey(key)) {
             consumerHandler = handlerMap.get(key);
             Channel channel = consumerHandler.getChannel();
             if(!channel.isOpen() || !channel.isActive()) {
                 consumerHandler = getConsumerHandler(key, host, port);
                 handlerMap.put(buildHandlerMapKey(request.getClassName(),request.getMethodName()),consumerHandler);
             }
         }else {
             consumerHandler = getConsumerHandler(key, host, port);
             if(consumerHandler == null) {
                 throw new RuntimeException("");
             }
             handlerMap.put(buildHandlerMapKey(request.getClassName(),request.getMethodName()),consumerHandler);
         }
         //发送消息
         consumerHandler.sendRequest(requestProtocolMessage);
     }

    private RpcConsumerHandler getConsumerHandler(String key,String host, int port) {
        try {
            ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
            channelFuture.addListener(new GenericFutureListener