• Netty——部分优化以及搭建简单RPC框架（笔记）

    文章目录


    代码接前文 Netty——搭建一个聊天室（笔记）

    一、优化

    1.1. 扩展序列化算法

    抽象序列化算法

    package com.yjx23332.netty.test.entity;

import java.io.*;

public interface Serializer {
    //反序列化算法
    <T> T deserialize(Class<T> clazz,byte[] bytes);
    //序列化算法
    <T> byte[] serialize(T object);

    enum Algorithm implements  Serializer{
        Java{
            @Override
            public <T> T deserialize(Class<T> clazz, byte[] bytes) {
                try (ObjectInputStream objectInputStream = new ObjectInputStream(new ByteArrayInputStream(bytes));){
                    return (T) objectInputStream.readObject();
                } catch (IOException | ClassNotFoundException e) {
                    throw new RuntimeException("反序列化失败",e);
                }
            }

            @Override
            public <T> byte[] serialize(T object) {
                try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
                     ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteArrayOutputStream);){
                    objectOutputStream.writeObject(object);
                    byte[] bytes = byteArrayOutputStream.toByteArray();
                    return bytes;
                } catch (IOException e) {
                    throw new RuntimeException("序列化失败",e);
                }
            }
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35

    修改编解码器

    package com.yjx23332.netty.test.protocol;

import com.yjx23332.netty.test.entity.Message;
import com.yjx23332.netty.test.entity.Serializer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.List;

/**
 * 必须确保收到的消息是完整的，才可以使用
 * */
@ChannelHandler.Sharable
@Slf4j
public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf,Message> {
    /***
     * 编码
     */
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> out) throws Exception {
        ByteBuf outTemp = ctx.alloc().buffer();
        //1. 4 字节表示魔数
        outTemp.writeBytes(new byte[]{1,2,3,4});
        //2. 1 字节表示版本
        outTemp.writeByte(1);
        //3. 1 字节表示序列化方式,0 jdk,1 json
        outTemp.writeByte(0);
        //4. 1 字节表示指令类型
        outTemp.writeByte(msg.getMessageType());
        //5. 4 字节表示序列号
        outTemp.writeInt(msg.getSequenceId());

        // 上述和为15，为了满足2^n倍，让内存对齐。填入一个无意义字节
        outTemp.writeByte(0xff);

        //6. 获取内容的字节数组
        byte[] bytes = Serializer.Algorithm.Java.serialize(msg);

        //7. 4字节表示长度
        outTemp.writeInt(bytes.length);
        //8. 写入内容
        outTemp.writeBytes(bytes);
        out.add(outTemp);
    }

    /**
     * 解码
     * */
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        int magicNum = in.readInt();
        byte version = in.readByte();
        byte serializerType = in.readByte();
        byte messageType = in.readByte();
        int sequenceId = in.readInt();
        in.readByte();
        int length = in.readInt();
        byte[] bytes = new byte[length];
        /**
         * 从当前读指针位置开始读
         * */
        in.readBytes(bytes,0,length);
        if(serializerType == 0){
            Message message = Serializer.Algorithm.Java.deserialize(Message.class,bytes);
            log.debug("{},{},{},{},{},{}",magicNum,version,serializerType,messageType,sequenceId,length);
            log.debug("{}",message);
            out.add(message);
        }
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78

    1.2 加入JSON序列化

    引入一个转json的包

    		<dependency>
            <groupId>com.alibaba.fastjson2groupId>
            <artifactId>fastjson2artifactId>
            <version>2.0.13version>
        dependency>

    • 1
    • 2
    • 3
    • 4
    • 5
    	JSON{
            @Override
            public <T> T deserialize(Class<T> clazz, byte[] bytes) {
                String json = new String(bytes,StandardCharsets.UTF_8);
                return JSONObject.parseObject(json,clazz);
            }

            @Override
            public <T> byte[] serialize(T object) {
                String json = JSONObject.toJSONString(object);
                return json.getBytes(StandardCharsets.UTF_8);
            }
        }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

    1.2.1 修改为从配置中读取设置

    修改为读取配置文件模式（也可以直接用@value）

    读取Properties
    package com.yjx23332.netty.test.config;

import com.yjx23332.netty.test.entity.Serializer;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

public abstract class Config {
    static Properties properties;
    static{
        try(InputStream in = Config.class.getResourceAsStream("/application.properties")){
            properties = new Properties();
            properties.load(in);
        }catch (IOException e){
            throw new ExceptionInInitializerError(e);
        }
    }
    public static int getServerPort(){
        String value = properties.getProperty("server.port");
        if(value == null){
            return 8080;
        }
        else{
            return Integer.parseInt(value);
        }
    }
    public static Serializer.Algorithm getSerializerAlgorithm(){
        String value = properties.getProperty("serializer.algorithm");
        if(value == null){
            return Serializer.Algorithm.Java;
        }
        else{
            return Serializer.Algorithm.valueOf(value);
        }
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    读取yml
    package com.yjx23332.netty.test.config;

import com.yjx23332.netty.test.entity.Serializer;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.core.io.ClassPathResource;


import java.util.Properties;

public abstract class Config {
    static Properties properties;
    static{
        YamlPropertiesFactoryBean yamlProFb = new YamlPropertiesFactoryBean();
        yamlProFb.setResources(new ClassPathResource("application.yml"));
        properties = yamlProFb.getObject();
    }
	//余下同上

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    笔者用的是yml

    server:
  port: 8080

serializer:
  algorithm: JSON

    • 1
    • 2
    • 3
    • 4
    • 5

    编码是从配置中读取
    解码是从收到的消息中读取

    package com.yjx23332.netty.test.protocol;

import com.yjx23332.netty.test.config.Config;
import com.yjx23332.netty.test.entity.Message;
import com.yjx23332.netty.test.entity.Serializer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.List;

/**
 * 必须确保收到的消息是完整的，才可以使用
 * */
@ChannelHandler.Sharable
@Slf4j
public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf,Message> {
    /***
     * 编码
     */
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> out) throws Exception {
        ByteBuf outTemp = ctx.alloc().buffer();
        //1. 4 字节表示魔数
        outTemp.writeBytes(new byte[]{1,2,3,4});
        //2. 1 字节表示版本
        outTemp.writeByte(1);
        //3. 1 字节表示序列化方式,0 jdk,1 json
        outTemp.writeByte(Config.getSerializerAlgorithm().ordinal());
        //4. 1 字节表示指令类型
        outTemp.writeByte(msg.getMessageType());
        //5. 4 字节表示序列号
        outTemp.writeInt(msg.getSequenceId());

        // 上述和为15，为了满足2^n倍，让内存对齐。填入一个无意义字节
        outTemp.writeByte(0xff);

        //6. 获取内容的字节数组
        byte[] bytes = Config.getSerializerAlgorithm().serialize(msg);

        //7. 4字节表示长度
        outTemp.writeInt(bytes.length);
        //8. 写入内容
        outTemp.writeBytes(bytes);
        out.add(outTemp);
    }

    /**
     * 解码
     * */
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        int magicNum = in.readInt();
        byte version = in.readByte();
        byte serializerType = in.readByte();
        byte messageType = in.readByte();
        int sequenceId = in.readInt();
        in.readByte();
        int length = in.readInt();
        byte[] bytes = new byte[length];
        /**
         * 从当前读指针位置开始读
         * */
        in.readBytes(bytes,0,length);
        in.readBytes(bytes,0,length);
        //找到反序列化算法
        Serializer.Algorithm algorithm = Serializer.Algorithm.values()[serializerType];
        Class<?> messageClass = Message.getMessageClass(messageType);
        Object message = algorithm.deserialize(messageClass,bytes);
        log.debug("{},{},{},{},{},{}",magicNum,version,serializerType,messageType,sequenceId,length);
        log.debug("{}",message);
        out.add(message);
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81

    1.2.2 测试

    在test中创建如下代码，即可测试

    package com.yjx23332.netty.test;

import com.yjx23332.netty.test.config.Config;
import com.yjx23332.netty.test.entity.Message;
import com.yjx23332.netty.test.entity.Serializer;
import com.yjx23332.netty.test.entity.vo.req.LoginRequestMessage;
import com.yjx23332.netty.test.protocol.MessageCodecSharable;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.logging.LoggingHandler;

public class TestSerializer {
    public static void main(String[] args) {
        MessageCodecSharable CODEC = new MessageCodecSharable();
        LoggingHandler LOGGING = new LoggingHandler();
        EmbeddedChannel channel = new EmbeddedChannel(LOGGING,CODEC,LOGGING);
        LoginRequestMessage message = new LoginRequestMessage("zhangsan","123");
        //出站测试
        //channel.writeOutbound(message);
        //入栈测试
        channel.writeInbound(messageToBytes(message));
    }

    public static ByteBuf messageToBytes(Message msg){
        int algorithm = Config.getSerializerAlgorithm().ordinal();
        ByteBuf out = ByteBufAllocator.DEFAULT.buffer();
        out.writeBytes(new byte[]{1,2,3,4});
        out.writeByte(1);
        out.writeByte(algorithm);
        out.writeByte(msg.getMessageType());
        out.writeInt(msg.getSequenceId());
        out.writeByte(0xff);
        byte[] bytes = Serializer.Algorithm.values()[algorithm].serialize(msg);
        out.writeInt(bytes.length);
        out.writeBytes(bytes);
        return out;
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40

    1.2 参数调整

    1.2.1 CONNECT_TIMEOUT_MILLIS 连接超时

    • 属于SocketChannel参数
    • 用在客户端建立连接时，如果过在指定好秒内无法丽娜姐，会抛出timeout异常
    • SO_TIMEOUT主要用在阻塞IO，阻塞IO中accept，read等都是无限等待的，如果不希望永远阻塞，使用它调整时间
    package com.yjx23332.netty.test;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class TestConnectionTimeout {
    public static void main(String[] args) {
        //1. 客户通过 .option() 方法配置参数 给SocketChannel 配置参数

        //2. 服务端
        // new ServerBootstrap().option() //是给 ServerSocketChannel
        // new ServerBootstrap().childOption() //给 SocketChannel
        NioEventLoopGroup group = new NioEventLoopGroup();
        try{
            Bootstrap bootstrap = new Bootstrap()
                    .group(group)
            		//单位毫秒
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,1000)
                    .channel(NioSocketChannel.class)
                    .handler(new LoggingHandler());
            ChannelFuture channelFuture = bootstrap.connect("localhost",8080);
            channelFuture.sync().channel().closeFuture().sync();
        }catch (Exception e){
            e.printStackTrace();
            log.debug("timeout");
        }finally {
            group.shutdownGracefully();
        }

    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    debug

    设置如下断点
    在这里插入图片描述
    在超时报错的抛出异常的地方也加入断点
    在这里插入图片描述

    为channelFuture添加标记
    在这里插入图片描述
    切换至NIO线程
    在这里插入图片描述

    可以看到future来自这里
    在这里插入图片描述
    本质是一个定时线程,到时间后去检测是否成功，没有成功就抛出异常
    通过Promise把结果返给主线程

    				boolean wasActive = isActive();
                if (doConnect(remoteAddress, localAddress)) {
                    fulfillConnectPromise(promise, wasActive);
                } else {
                    connectPromise = promise;
                    requestedRemoteAddress = remoteAddress;

                    // Schedule connect timeout.
                    int connectTimeoutMillis = config().getConnectTimeoutMillis();
                    if (connectTimeoutMillis > 0) {
                        connectTimeoutFuture = eventLoop().schedule(new Runnable() {
                            @Override
                            public void run() {
                                ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                                if (connectPromise != null && !connectPromise.isDone()
                                        && connectPromise.tryFailure(new ConnectTimeoutException(
                                                "connection timed out: " + remoteAddress))) {
                                    close(voidPromise());
                                }
                            }
                        }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
                    }

                    promise.addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if (future.isCancelled()) {
                                if (connectTimeoutFuture != null) {
                                    connectTimeoutFuture.cancel(false);
                                }
                                connectPromise = null;
                                close(voidPromise());
                            }
                        }
                    });
                }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36

    1.2.2 SO_BACKLOG 连接队列

    client server syns queue accept queue bind() listen() connect() 1. SYN SYN_SEND put SYN_RCVD 2. SYN + ACK ESTABLISH 3. ACK put ESTABLISH accept() client server syns queue accept queue
    • syns queue:半连接队列
    • accept queue：全连接队列
    1. linux旧版本（2.2之前）中，backlog大小包括了两个队列的大小，之后分为了如下两个队列
    2. sync queue：通过/proc/sys/net/ipv4/tcp_max_syn_backlog指定，在syncookies启用的情况之下，逻辑上没有最大值限制，因此这个设置可以被忽略
    3. accept queue
      • 通过/proc/sys/net/core/somaxconn指定，在使用listen函数时，内核会根据传入的backlog参数与系统参数，取二者较小值
      • 如果accept queue队列满了，server将发送一个拒绝连接的错误信息到client

    nio中，可以用bind(端口号,backlog参数)设置
    在netty中，可以通过SO_BACKLOG来设置

    package com.yjx23332.netty.test;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LoggingHandler;

public class TestBacklogServer {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .option(ChannelOption.SO_BACKLOG,2) //全连接队列设置
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>(){
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new LoggingHandler());
                    }
                }).bind(8080);
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    debug

    在NioEventLoop方法中，打入如下断点，用来填满队列
    在这里插入图片描述

    随便找一个客户端即可，开启多实例模式
    在这里插入图片描述
    就可以出现报错了。
    在这里插入图片描述
    在此处为backlog的默认值
    在这里插入图片描述
    追踪可以看到，windows平台默认200，其余128
    如果存在配置文件，读出，则以这个数字为准
    在这里插入图片描述

    			// Determine the default somaxconn (server socket backlog) value of the platform.
            // The known defaults:
            // - Windows NT Server 4.0+: 200
            // - Linux and Mac OS X: 128
            int somaxconn = PlatformDependent.isWindows() ? 200 : 128;
            File file = new File("/proc/sys/net/core/somaxconn");
            BufferedReader in = null;
            try {
                // file.exists() may throw a SecurityException if a SecurityManager is used, so execute it in the
                // try / catch block.
                // See https://github.com/netty/netty/issues/4936
                if (file.exists()) {
                    in = new BufferedReader(new FileReader(file));
                    somaxconn = Integer.parseInt(in.readLine());
                    if (logger.isDebugEnabled()) {
                        logger.debug("{}: {}", file, somaxconn);
                    }
                } else {
                    // Try to get from sysctl
                    Integer tmp = null;
                    if (SystemPropertyUtil.getBoolean("io.netty.net.somaxconn.trySysctl", false)) {
                        tmp = sysctlGetInt("kern.ipc.somaxconn");
                        if (tmp == null) {
                            tmp = sysctlGetInt("kern.ipc.soacceptqueue");
                            if (tmp != null) {
                                somaxconn = tmp;
                            }
                        } else {
                            somaxconn = tmp;
                        }
                    }

                    if (tmp == null) {
                        logger.debug("Failed to get SOMAXCONN from sysctl and file {}. Default: {}", file,
                                somaxconn);
                    }
                }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37

    1.2.3 ulimit -n

    • 限制进程能够打开文件的数量
    • 属于操作系统参数，文件设置。

    1.2.4 TCP_NODELAY

    属于socketChannel，对于一些小的数据的发送，TCP可能不会及时发送，而是当累计到了一定量后，一起发出去。通过该参数可以设置是否要延迟发送。
    默认是false。也就是开启了延迟发送（nagle算法）。

    1.2.5 SO_SNDBUF & SO_RCVBUF

    • SO_SNDBUF 属于SocketChannel 参数
    • SO_RCVBUF 既可用于 SocketChannel 参数，也可以用于 ServerSocketChannel 参数（建议设置到 ServerSocketChannel 上）
      现在的TCP协议会在连接时，双方协商一个大小，一般不用自己调整。

    1.2.6 ALLOCATOR分配器配置

    • 属于 SocketChannel 参数
    • 用于分配ByteBuf 与 ctx.alloc()

    在这里插入图片描述
    在这里插入图片描述
    要修改的话如下即可（虚拟机参数）
    在这里插入图片描述

    如下为，是否首选直接内存
    在这里插入图片描述

    在这里插入图片描述

    修改如下
    在这里插入图片描述

    package com.yjx23332.netty.test;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class TestByteBuf {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new LoggingHandler());
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf buf = ctx.alloc().buffer();
                                log.debug("alloc buf {}",buf);
                            }
                        });
                    }
                }).bind(8080);
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35

    运行后，获得消息时，可得下列结果
    在这里插入图片描述

    1.2.7 RCVBUF_ALLOCATOR

    • 属于SocketChannel参数
    • 控制netty接受缓冲区大小
    • 负责入站数据的分配，决定缓冲区大小（可动态调整），
    public class TestByteBuf {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new LoggingHandler());
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf buf = ctx.alloc().buffer();
                                log.debug("alloc buf {}",buf);
                                log.debug("receive buf {}",msg);
                            }
                        });
                    }
                }).bind(8080);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21

    我们可以发现，即使设置了非池化和非直接内存，直接发送的IO消息，仍然是池化+直接内存。
    在这里插入图片描述
    allocHandle是RecvByteBufAllocator的内部类，allocate方法，会创建最初的byteBuf。决定了byteBuf的大小以及是否为直接内存。
    allocator是分配器，决定池化还是非池化
    在这里插入图片描述
    强制使用ioBuffer（直接内存），guess猜测大小，依据于实际数据量

    在这里插入图片描述
    自适应缓冲大小
    在这里插入图片描述

    二、简化的 RPC 框架

    RPC：Remote Produce Call，远程调用请求。
    基于上文聊天项目，搭建RPC请求和响应消息。

    2.1 新增代码

    Message

    为Message类，新增

    @Data
public abstract  class Message implements Serializable {
	...
	public static final int RPC_MESSAGE_TYPE_REQUEST = 101;

    public static final int RPC_MESSAGE_TYPE_RESPONSE = 102;

	static{
		...
		messageClasses.put(RPC_MESSAGE_TYPE_REQUEST,RpcRequestMessage.class);
        messageClasses.put(RPC_MESSAGE_TYPE_RESPONSE,RpcResponseMessage.class);
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

    RPCMessage

    package com.yjx23332.netty.test.entity.vo.req;

import com.yjx23332.netty.test.entity.Message;

@Data
@ToString(callSuper = true)
public class RpcRequestMessage extends Message {
    /**
     * 调用的接口全限定名，服务端根据它找到实现
     * */
    private String interfaceName;
    /**
     * 调用接口中的方法名
     * */
    private String methodName;
    /**
     * 方法返回类型
     * */
    private Class<?> returnType;
    /**
     * 方法参数类型数组
     * */
    private Class[] parameterTypes;
    /**
     * 方法参数值数组
     * */
    private Object[] parameterValue;

    public RpcRequestMessage(int sequenceId,String interfaceName,String methodName,Class<?> returnType,Class[] parameterTypes,Object[] parameterValue){
        super.setSequenceId(sequenceId);
        this.interfaceName = interfaceName;
        this.methodName = methodName;
        this.returnType = returnType;
        this.parameterTypes = parameterTypes;
        this.parameterValue = parameterValue;
    }



    @Override
    public int getMessageType() {
        return RPC_MESSAGE_TYPE_REQUEST;
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    package com.yjx23332.netty.test.entity.vo.resp;

import com.yjx23332.netty.test.entity.Message;
import com.yjx23332.netty.test.entity.vo.req.RpcRequestMessage;
import lombok.Data;
import lombok.ToString;

@Data
@ToString(callSuper = true)
public class RpcResponseMessage extends Message {
    /**
     * 返回值
     * */
    private Object returnValue;

    /**
     * 异常值
     * */
    private Exception exceptionValue;

    @Override
    public int getMessageType() {
        return RPC_MESSAGE_TYPE_RESPONSE;
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26

    RPCServer

    package com.yjx23332.netty.test.server;


import com.yjx23332.netty.test.handler.RpcRequestMessageHandler;
import com.yjx23332.netty.test.protocol.MessageCodecSharable;
import com.yjx23332.netty.test.protocol.ProcotolFrameDecoder;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class RPCServer {
    public static void main(String[] args) {
        NioEventLoopGroup boss = new NioEventLoopGroup();
        NioEventLoopGroup worker = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler();
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();

        // rpc 请求消息处理
        RpcRequestMessageHandler RPC_HANDLER = new RpcRequestMessageHandler();
        try{
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap
                    .channel(NioServerSocketChannel.class)
                    .group(boss,worker)
                    .childHandler(new ChannelInitializer<SocketChannel>(){
                        @Override
                        protected void initChannel(SocketChannel ch){
                            ch.pipeline().addLast(new ProcotolFrameDecoder());
                            ch.pipeline().addLast(LOGGING_HANDLER);
                            ch.pipeline().addLast(MESSAGE_CODEC);
                            ch.pipeline().addLast(RPC_HANDLER);
                        }
                    });
            ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
            channelFuture.channel().closeFuture().sync();
        }catch (InterruptedException e){
            log.error("server error {}",e);
        }finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}



    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51

    RPCClient

    package com.yjx23332.netty.test.client;

import com.yjx23332.netty.test.handler.RpcResponseMessageHandler;
import com.yjx23332.netty.test.protocol.MessageCodecSharable;
import com.yjx23332.netty.test.protocol.ProcotolFrameDecoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;


@Slf4j
public class RPCClient {
    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();

        //rpc响应消息处理器
        RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
        try{
            Bootstrap bootstrap = new Bootstrap();
            bootstrap
                    .channel(NioSocketChannel.class)
                    .group(group)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new ProcotolFrameDecoder());
                            ch.pipeline().addLast(LOGGING_HANDLER);
                            ch.pipeline().addLast(MESSAGE_CODEC);
                            ch.pipeline().addLast(RPC_HANDLER);
                        }
                    });
            ChannelFuture channelFuture = bootstrap.connect("localhost",8080).sync();
            channelFuture.channel().writeAndFlush(new RpcRequestMessage(
                    1,
                    "com.yjx23332.netty.test.server.service.HelloService",
                    "sayHello",
                    String.class,
                    new  Class[]{String.class},
                    new Object[]{"张三"}
            ));
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.debug("client error {}",e);
        }finally {
            group.shutdownGracefully();
        }
    }
}



    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58

    HelloService

    这是我们远程调用的服务

    package com.yjx23332.netty.test.server.service;

public interface HelloService {
    String sayHello(String userName);
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    package com.yjx23332.netty.test.server.service.Impl;

import com.yjx23332.netty.test.server.service.HelloService;

public class HelloServiceImpl implements HelloService {
    @Override
    public String sayHello(String userName) {
        return "你好," + userName;
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    ServicesFactory

    package com.yjx23332.netty.test.server.factory;

import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.core.io.ClassPathResource;

import java.lang.reflect.InvocationTargetException;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public abstract class ServicesFactory{
    static Properties properties;
    static Map<Class<?>,Object> map = new ConcurrentHashMap<>();
    static{
        try{
            YamlPropertiesFactoryBean yamlProFb = new YamlPropertiesFactoryBean();
            yamlProFb.setResources(new ClassPathResource("application.yml"));
            properties = yamlProFb.getObject();
            Set<String> names = properties.stringPropertyNames();
            for(String name : names){
                if(name.endsWith("Service")){
                    Class<?> interfaceClass = Class.forName(name);
                    Class<?> instanceClass = Class.forName(properties.getProperty(name));
                    map.put(interfaceClass,instanceClass.getDeclaredConstructor().newInstance());
                }
            }
        } catch (ClassNotFoundException | InstantiationError|InstantiationException|IllegalAccessException|InvocationTargetException|NoSuchMethodException e) {
            throw new RuntimeException(e);
        }
    }
    public static <T> T getService(Class<T> interfaceClass){ return (T) map.get(interfaceClass);}
}




    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    server:
  port: 8080

serializer:
  algorithm: JSON

com:
  yjx23332:
    netty:
      test:
        server:
          service:
            HelloService: com.yjx23332.netty.test.server.service.Impl.HelloServiceImpl

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

    Handler

    package com.yjx23332.netty.test.handler;

import com.yjx23332.netty.test.entity.vo.req.RpcRequestMessage;
import com.yjx23332.netty.test.entity.vo.resp.RpcResponseMessage;
import com.yjx23332.netty.test.server.factory.ServicesFactory;
import com.yjx23332.netty.test.server.service.HelloService;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;


@Slf4j
@ChannelHandler.Sharable
public class RpcRequestMessageHandler extends SimpleChannelInboundHandler<RpcRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequestMessage msg) throws Exception {
    	RpcResponseMessage rpcResponseMessage = new RpcResponseMessage();
    	rpcResponseMessage.setSequenceId(msg.getSequenceId());
        try {
            HelloService helloService = (HelloService) ServicesFactory.getService(Class.forName(msg.getInterfaceName()));
            Method method = helloService.getClass().getMethod(msg.getMethodName(),msg.getParameterTypes());
            Object invoke = method.invoke(helloService,msg.getParameterValue());
            rpcResponseMessage.setReturnValue(invoke);
        }catch (Exception e) {
            log.debug("{}", e);
            rpcResponseMessage.setExceptionValue(new Exception("远程调用出错" + e.getCause().getMessage()));
        }finally {
            ctx.writeAndFlush(rpcResponseMessage);
        }
    }
    /**
     * 测试用
     * */
    public static void main(String[] args) throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, IllegalAccessException {
        RpcRequestMessage rpcRequestMessage = new RpcRequestMessage(
                1,
                "com.yjx23332.netty.test.server.service.HelloService",
                "sayHello",
                String.class,
                new  Class[]{String.class},
                new Object[]{"张三"}
        );
        HelloService helloService = (HelloService)
                ServicesFactory.getService(Class.forName(rpcRequestMessage.getInterfaceName()));
        Method method = helloService.getClass().getMethod(rpcRequestMessage.getMethodName(),rpcRequestMessage.getParameterTypes());
        Object invoke = method.invoke(helloService,rpcRequestMessage.getParameterValue());
        System.out.println(invoke);
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    package com.yjx23332.netty.test.handler;

import com.yjx23332.netty.test.entity.vo.resp.RpcResponseMessage;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@ChannelHandler.Sharable
public class RpcResponseMessageHandler extends SimpleChannelInboundHandler<RpcResponseMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage msg) throws Exception {
        log.debug("{}",msg);
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    修改序列化方法

    我们这个地方传的是className，而我们原来的代码需要的是类的反射。因此我们要么设置一个转换器，让它自动处理class的名字与反射内容之间的转换。要么 ，就使用支持类名模式。
    笔者图方便，使用了支持类名模式。

    package com.yjx23332.netty.test.entity;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSONReader;

import java.io.*;
import java.nio.charset.StandardCharsets;

public interface Serializer {
    //反序列化算法
    <T> T deserialize(Class<T> clazz,byte[] bytes);
    //序列化算法
    <T> byte[] serialize(T object);

    enum Algorithm implements  Serializer{
        Java{
            @Override
            public <T> T deserialize(Class<T> clazz, byte[] bytes) {
                try (ObjectInputStream objectInputStream = new ObjectInputStream(new ByteArrayInputStream(bytes));){
                    return (T) objectInputStream.readObject();
                } catch (IOException | ClassNotFoundException e) {
                    throw new RuntimeException("反序列化失败",e);
                }
            }

            @Override
            public <T> byte[] serialize(T object) {
                try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
                     ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteArrayOutputStream);){
                    objectOutputStream.writeObject(object);
                    byte[] bytes = byteArrayOutputStream.toByteArray();
                    return bytes;
                } catch (IOException e) {
                    throw new RuntimeException("序列化失败",e);
                }
            }
        },
        JSON{
            @Override
            public <T> T deserialize(Class<T> clazz, byte[] bytes) {
                String json = new String(bytes,StandardCharsets.UTF_8);
                return JSONObject.parseObject(json,clazz,JSONReader.Feature.SupportClassForName);
            }

            @Override
            public <T> byte[] serialize(T object) {
                String json = JSONObject.toJSONString(object);
                return json.getBytes(StandardCharsets.UTF_8);
            }
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52

    在这里插入图片描述

    2.2 让它更符合使用习惯

    改造RPCResponseHandler

    package com.yjx23332.netty.test.handler;

import com.yjx23332.netty.test.entity.vo.resp.RpcResponseMessage;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.concurrent.Promise;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

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

@Slf4j
@Data
@ChannelHandler.Sharable
public class RpcResponseMessageHandler extends SimpleChannelInboundHandler<RpcResponseMessage> {
    /***
     * Integer - 序列号
     * Promise - 接收结果
     */
    public static final Map<Integer, Promise<Object>> PROMISE_MAP = new ConcurrentHashMap<>();
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage msg) throws Exception {
        log.debug("{}",msg);
        Promise<Object> promise = PROMISE_MAP.remove(msg.getSequenceId());
        if(promise != null){
            Object returnValue = msg.getReturnValue();
            Exception exceptionValue = msg.getExceptionValue();
            if(exceptionValue != null){
                promise.setFailure(exceptionValue);
            }
            else{
                promise.setSuccess(returnValue);
            }
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38

    改造客户端

    package com.yjx23332.netty.test.client;

import com.yjx23332.netty.test.entity.vo.req.RpcRequestMessage;
import com.yjx23332.netty.test.handler.RpcResponseMessageHandler;
import com.yjx23332.netty.test.protocol.MessageCodecSharable;
import com.yjx23332.netty.test.protocol.ProcotolFrameDecoder;
import com.yjx23332.netty.test.server.service.HelloService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Proxy;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;


@Slf4j
public class RPCClientManager {
    private static volatile Channel channel = null;
    /**
     * 计数器
     * */
    private  static  final AtomicInteger ID = new AtomicInteger();
    /**
     * 锁
     * */
    private static final Object LOCK = new Object();
    /**
     * 初始化channel
     * */
    private static void initChannel(){
        NioEventLoopGroup group = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();

        //rpc响应消息处理器
        RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap
                .channel(NioSocketChannel.class)
                .group(group)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new ProcotolFrameDecoder());
                        ch.pipeline().addLast(LOGGING_HANDLER);
                        ch.pipeline().addLast(MESSAGE_CODEC);
                        ch.pipeline().addLast(RPC_HANDLER);
                    }
                });
        try{
            channel = bootstrap.connect("localhost",8080).sync().channel();
            channel.closeFuture().addListener(promise->{
                group.shutdownGracefully();
            });
        } catch (InterruptedException e) {
            log.debug("client error {}",e);
        }
    }

     public static Channel getChannel(){
        if(channel == null){
            synchronized (LOCK){
                if(channel == null){
                    initChannel();
                }
            }
        }
        return channel;
     }

    /**
     * 代理类，将方法调用转换为消息
     * 在这里我们就不调用原方法，而是直接替换为发送消息的逻辑
     * 然后，我们在后面等待结果即可
     * */
    public static <T> T getProxyService(Class<T> serviceClass){
        ClassLoader loader = serviceClass.getClassLoader();
        Class<?>[] interfaces = new Class[]{serviceClass};
        int sequenceId = ID.getAndAdd(1);
        Object o = Proxy.newProxyInstance(loader,interfaces,(proxy,method,args)->{
            //1. 将方法的调用，转换成消息对象
            RpcRequestMessage rpcRequestMessage = new RpcRequestMessage(
                    sequenceId,
                    serviceClass.getName(),
                    method.getName(),
                    method.getReturnType(),
                    method.getParameterTypes(),
                    args
            );

            //2. 发送消息对象
            getChannel().writeAndFlush(rpcRequestMessage);

            //3. 准备一个空 Promise 对象 ，来接受结果
            //4. 用 eventLoop 线程来异步接收结果，当我们addListener时，就由它来接收
            //此处我们时同步接收
            DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());
            RpcResponseMessageHandler.PROMISE_MAP.put(sequenceId,promise);
            promise.await();
            if(promise.isSuccess()){
                return promise.getNow();
            }
            else{
                throw new RuntimeException(promise.cause());
            }
        });
        return (T) o;
    }

   public static void main(String[] args) {
        HelloService service = getProxyService(HelloService.class);
        System.out.println(service.sayHello("zhangsan"));
        System.out.println(service.sayHello("lisi"));
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124

    笔记中的工厂和代理都是自己实现，而不是用Spring的方式，实际开发中，可以用SpringBoot来帮助我们快速开发。

    参考文献

    [1]黑马程序员Netty全套教程

  • 相关阅读:
    132. SAP UI5 Simple Form 控件的使用方法介绍
    [游戏开发][Unity]编辑器Assets操作API大全
    计算机毕业设计Java网上鲜花交易平台（源码+系统+mysql数据库+Lw文档）
    golang slice/array无重复取随机内容
    【算法刷题第二题】牛客题:旋转数组的最小数字
    Spring源码里开天辟地的五个Bean，再介绍一个学习方法
    2023年Vue开发中的8个最佳工具
    用HTML+CSS+JS做一个漂亮简单的公司网站（JavaScript期末大作业）
    YGG 与《元素突击》的创造者游戏平台 G4AL 达成合作
    Synchronized
  • 原文地址：https://blog.csdn.net/weixin_46949627/article/details/126824605