Netty优化与源码

黑马程序员Netty笔记合集
注意：由于章节连贯，此套笔记更适合学习《黑马Netty全套课程》的同学参考、复习使用。

一、优化

1.1 扩展序列化算法

序列化，反序列化主要用在消息正文的转换上

• 序列化时，需要将 Java 对象变为要传输的数据（可以是 byte[]，或 json 等，最终都需要变成 byte[]）
• 反序列化时，需要将传入的正文数据还原成 Java 对象，便于处理

消息编解码器 MessageCodeSharable 原序列化和反序列化机制

// 反序列化
byte[] body = new byte[bodyLength];
byteByf.readBytes(body);
ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(body));
Message message = (Message) in.readObject();
message.setSequenceId(sequenceId);

// 序列化
ByteArrayOutputStream out = new ByteArrayOutputStream();
new ObjectOutputStream(out).writeObject(message);
byte[] bytes = out.toByteArray();
1
2
3
4
5
6
7
8
9
10
11

为了支持更多序列化算法，抽象一个 Serializer 接口

public interface Serializer {
    //反序列化
    <T> T deserializer(Class<T> clazz,byte[] bytes);

    //序列化
    <T> byte[] serializer(T object);

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

            @Override
            public <T> byte[] serializer(T object) {
                try {
                    ByteArrayOutputStream baos = new ByteArrayOutputStream();
                    ObjectOutputStream oos=new ObjectOutputStream(baos);
                    oos.writeObject(object);
                    return baos.toByteArray();
                } catch (IOException e) {
                    throw new RuntimeException("序列化失败！",e);
                }
            }
        },
        Json{
            @Override
            public <T> T deserializer(Class<T> clazz, byte[] bytes) {
                return new Gson().fromJson(new String(bytes, StandardCharsets.UTF_8),clazz);
            }

            @Override
            public <T> byte[] serializer(T object) {
                return new Gson().toJson(object).getBytes(StandardCharsets.UTF_8);
            }
        }
        // 需要从协议的字节中得到是哪种序列化算法
    	public static SerializerAlgorithm getByInt(int type) {
        	SerializerAlgorithm[] array = SerializerAlgorithm.values();
        	if (type < 0 || type > array.length - 1) {
            	throw new IllegalArgumentException("超过 SerializerAlgorithm 范围");
        	}
        	return array[type];
    	}
    }
}
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

增加配置类和配置文件

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.SerializerAlgorithm getSerializerAlgorithm() {
        String value = properties.getProperty("serializer.algorithm");
        if(value == null) {
            return Serializer.SerializerAlgorithm.Java;
        } else {
            return Serializer.SerializerAlgorithm.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

配置文件：application.properties

serializer.algorithm=Json
1

修改编解码器

/**
 * 必须和 LengthFieldBasedFrameDecoder 一起使用，确保接到的 ByteBuf 消息是完整的
 */
@Slf4j
@ChannelHandler.Sharable
public class MessageCodeSharable extends MessageToMessageCodec<ByteBuf, Message> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> list) throws Exception {
        ByteBuf byteBuf = ctx.alloc().buffer();
        //1.魔数：4个字节
        byteBuf.writeBytes(new byte[]{1,2,3,4});
        //2.版本号：1个字节
        byteBuf.writeByte(1);
        //3.序列化算法：1个字节，0 jdk ，1 json
        //获取配置文件中指定序的列化算法在序列化枚举数组中的索引下标
        byteBuf.writeByte(Config.getSerializerAlgorithm().ordinal());
        //4.消息类型：1个字节，由消息本身决定
        byteBuf.writeByte(msg.getMessageType());
        //5.请求序号：4个字节，由消息本身自带
        byteBuf.writeInt(msg.getSequenceId());
        byteBuf.writeByte(-1); //填充：无意义字节
        //使用指定的序列化算法转换正文内容
        byte[] bytes = Config.getSerializerAlgorithm().serializer(msg);
        //6.正文长度：4个字节
        byteBuf.writeInt(bytes.length);
        //7.消息正文：
        byteBuf.writeBytes(bytes);
        list.add(byteBuf);
    }

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
        //1.魔数：4个字节
        int magicNum = byteBuf.readInt();
        //2.版本号：1个字节
        byte version=byteBuf.readByte();
        //3.序列化算法：0 jdk ，1 json
        byte serializerType=byteBuf.readByte();
        //4.指令类型：由消息本身决定
        byte messageType=byteBuf.readByte();
        //5.请求序号：由消息本身自带
        int sequenceId=byteBuf.readInt();
        byteBuf.readByte();
        //6.正文长度
        int length=byteBuf.readInt();
        //7.消息正文
        byte[] bytes=new byte[length];
        byteBuf.readBytes(bytes,0,length);
        //获取序列化算法
        Serializer.SerializerAlgorithm serializerAlgorithm = Config.getSerializerAlgorithm();
        //通过获取 Message 实现类对象
        Class<? extends Message> messageClass = Message.getMessageClass(messageType);
        //通过指定的算法将字节数组转换为 对应的Message实现类 对象
        Object message = Config.getSerializerAlgorithm().deserializer(messageClass, bytes);
        log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
        log.debug("{}", message);
        list.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

消息实现类的获取

@Data
public abstract class Message implements Serializable {

    /**
     * 根据消息类型字节，获得对应的消息 class
     * @param messageType 消息类型字节
     * @return 消息 class
     */
    public static Class<? extends Message> getMessageClass(int messageType) {
        return messageClasses.get(messageType);
    }

    private int sequenceId;

    private int messageType;

    public abstract int getMessageType();

    public static final int LoginRequestMessage = 0;
    public static final int LoginResponseMessage = 1;
    public static final int ChatRequestMessage = 2;
    public static final int ChatResponseMessage = 3;
    public static final int GroupCreateRequestMessage = 4;
    public static final int GroupCreateResponseMessage = 5;
    public static final int GroupJoinRequestMessage = 6;
    public static final int GroupJoinResponseMessage = 7;
    public static final int GroupQuitRequestMessage = 8;
    public static final int GroupQuitResponseMessage = 9;
    public static final int GroupChatRequestMessage = 10;
    public static final int GroupChatResponseMessage = 11;
    public static final int GroupMembersRequestMessage = 12;
    public static final int GroupMembersResponseMessage = 13;
    public static final int PingMessage = 14;
    public static final int PongMessage = 15;
    /**
     * 请求类型 byte 值
     */
    public static final int RPC_MESSAGE_TYPE_REQUEST = 101;
    /**
     * 响应类型 byte 值
     */
    public static final int  RPC_MESSAGE_TYPE_RESPONSE = 102;

    private static final Map<Integer, Class<? extends Message>> messageClasses = new HashMap<>();

    static {
        messageClasses.put(LoginRequestMessage, LoginRequestMessage.class);
        messageClasses.put(LoginResponseMessage, LoginResponseMessage.class);
        messageClasses.put(ChatRequestMessage, ChatRequestMessage.class);
        messageClasses.put(ChatResponseMessage, ChatResponseMessage.class);
        messageClasses.put(GroupCreateRequestMessage, GroupCreateRequestMessage.class);
        messageClasses.put(GroupCreateResponseMessage, GroupCreateResponseMessage.class);
        messageClasses.put(GroupJoinRequestMessage, GroupJoinRequestMessage.class);
        messageClasses.put(GroupJoinResponseMessage, GroupJoinResponseMessage.class);
        messageClasses.put(GroupQuitRequestMessage, GroupQuitRequestMessage.class);
        messageClasses.put(GroupQuitResponseMessage, GroupQuitResponseMessage.class);
        messageClasses.put(GroupChatRequestMessage, GroupChatRequestMessage.class);
        messageClasses.put(GroupChatResponseMessage, GroupChatResponseMessage.class);
        messageClasses.put(GroupMembersRequestMessage, GroupMembersRequestMessage.class);
        messageClasses.put(GroupMembersResponseMessage, GroupMembersResponseMessage.class);
        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
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

测试

/**
 * 出站
 */
@Test
public void test(){
    LoggingHandler LOGGING = new LoggingHandler();
    MessageCodeSharable CODEC = new MessageCodeSharable();
    EmbeddedChannel channel=new EmbeddedChannel(LOGGING,CODEC,LOGGING);
    channel.writeOutbound(new ChatRequestMessage("zhangsan","lisi","Hello"));
}
1
2
3
4
5
6
7
8
9
10

• Java

  10:14:09 [DEBUG] [main] i.n.h.l.LoggingHandler : [id: 0xembedded, L:embedded - R:embedded] WRITE: ChatRequestMessage(super=Message(sequenceId=0, messageType=2), content=Hello, to=lisi, from=zhangsan)
  10:14:09 [DEBUG] [main] i.n.h.l.LoggingHandler : [id: 0xembedded, L:embedded - R:embedded] WRITE: 259B
           +-------------------------------------------------+
           |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  +--------+-------------------------------------------------+----------------+
  |00000000| 01 02 03 04 01 00 02 00 00 00 00 ff 00 00 00 f3 |................|
  |00000010| ac ed 00 05 73 72 00 34 6f 72 67 2e 65 78 61 6d |....sr.4org.exam|
  |00000020| 70 6c 65 2e 6a 61 76 61 2e 63 68 61 74 52 6f 6f |ple.java.chatRoo|
  |00000030| 6d 2e 6d 65 73 73 61 67 65 2e 43 68 61 74 52 65 |m.message.ChatRe|
  |00000040| 71 75 65 73 74 4d 65 73 73 61 67 65 78 13 15 e1 |questMessagex...|
  |00000050| fe 0b 6d 55 02 00 03 4c 00 07 63 6f 6e 74 65 6e |..mU...L..conten|
  |00000060| 74 74 00 12 4c 6a 61 76 61 2f 6c 61 6e 67 2f 53 |tt..Ljava/lang/S|
  |00000070| 74 72 69 6e 67 3b 4c 00 04 66 72 6f 6d 71 00 7e |tring;L..fromq.~|
  |00000080| 00 01 4c 00 02 74 6f 71 00 7e 00 01 78 72 00 29 |..L..toq.~..xr.)|
  |00000090| 6f 72 67 2e 65 78 61 6d 70 6c 65 2e 6a 61 76 61 |org.example.java|
  |000000a0| 2e 63 68 61 74 52 6f 6f 6d 2e 6d 65 73 73 61 67 |.chatRoom.messag|
  |000000b0| 65 2e 4d 65 73 73 61 67 65 06 07 90 d6 50 f8 11 |e.Message....P..|
  |000000c0| fa 02 00 02 49 00 0b 6d 65 73 73 61 67 65 54 79 |....I..messageTy|
  |000000d0| 70 65 49 00 0a 73 65 71 75 65 6e 63 65 49 64 78 |peI..sequenceIdx|
  |000000e0| 70 00 00 00 00 00 00 00 00 74 00 05 48 65 6c 6c |p........t..Hell|
  |000000f0| 6f 74 00 08 7a 68 61 6e 67 73 61 6e 74 00 04 6c |ot..zhangsant..l|
  |00000100| 69 73 69                                        |isi             |
  +--------+-------------------------------------------------+----------------+
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23

• Json

  10:15:17 [DEBUG] [main] i.n.h.l.LoggingHandler : [id: 0xembedded, L:embedded - R:embedded] WRITE: ChatRequestMessage(super=Message(sequenceId=0, messageType=2), content=Hello, to=lisi, from=zhangsan)
  10:15:18 [DEBUG] [main] i.n.h.l.LoggingHandler : [id: 0xembedded, L:embedded - R:embedded] WRITE: 96B
           +-------------------------------------------------+
           |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  +--------+-------------------------------------------------+----------------+
  |00000000| 01 02 03 04 01 01 02 00 00 00 00 ff 00 00 00 50 |...............P|
  |00000010| 7b 22 63 6f 6e 74 65 6e 74 22 3a 22 48 65 6c 6c |{"content":"Hell|
  |00000020| 6f 22 2c 22 74 6f 22 3a 22 6c 69 73 69 22 2c 22 |o","to":"lisi","|
  |00000030| 66 72 6f 6d 22 3a 22 7a 68 61 6e 67 73 61 6e 22 |from":"zhangsan"|
  |00000040| 2c 22 73 65 71 75 65 6e 63 65 49 64 22 3a 30 2c |,"sequenceId":0,|
  |00000050| 22 6d 65 73 73 61 67 65 54 79 70 65 22 3a 30 7d |"messageType":0}|
  +--------+-------------------------------------------------+----------------+
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12

1.2 参数调优

1）CONNECT_TIMEOUT_MILLIS

• 属于 SocketChannal 参数

• 用在客户端建立连接时，如果在指定毫秒内无法连接，会抛出 timeout 异常

• SO_TIMEOUT 主要用在阻塞 IO，阻塞 IO 中 accept，read 等都是无限等待的，如果不希望永远阻塞，使用它调整超时时间

@Slf4j
public class TestConnectionTimeout {
    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap()
                    .group(group)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 300)
                    .channel(NioSocketChannel.class)
                    .handler(new LoggingHandler());
            ChannelFuture future = bootstrap.connect("127.0.0.1", 8080);
            future.sync().channel().closeFuture().sync(); // 断点1
        } 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

超时异常

io.netty.channel.ConnectTimeoutException: connection timed out: localhost/127.0.0.1:8080
	at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe$1.run(AbstractNioChannel.java:263)
	at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
	at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:127)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:416)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:515)
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.lang.Thread.run(Thread.java:750)
1
2
3
4
5
6
7
8
9
10
11

另外源码部分 io.netty.channel.nio.AbstractNioChannel.AbstractNioUnsafe#connect

/**
 * 1.线程间使用 Promise 进行通信
 * 2.根据超时时间设置 定时抛异常任务 。如果定时任务未被取消，则说明连接超时
 */
@Override
public final void connect(
        final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
    // ...
    // Schedule connect timeout.
    int connectTimeoutMillis = config().getConnectTimeoutMillis();
    if (connectTimeoutMillis > 0) {
        connectTimeoutFuture = eventLoop().schedule(new Runnable() {
            @Override
            public void run() {                
                ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                ConnectTimeoutException cause =
                    new ConnectTimeoutException("connection timed out: " + remoteAddress); // 断点2
                if (connectPromise != null && connectPromise.tryFailure(cause)) {
                    close(voidPromise());
                }
            }
        }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
    }
	// ...
}
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

2）SO_BACKLOG

• 属于 ServerSocketChannal 参数

1. 第一次握手，client 发送 SYN 到 server，状态修改为 SYN_SEND，server 收到，状态改变为 SYN_REVD，并将该请求放入 sync queue 队列
2. 第二次握手，server 回复 SYN + ACK 给 client，client 收到，状态改变为 ESTABLISHED，并发送 ACK 给 server
3. 第三次握手，server 收到 ACK，状态改变为 ESTABLISHED，将该请求从 sync queue 放入 accept queue

其中

• 在 linux 2.2 之前，backlog 大小包括了两个队列的大小，在 2.2 之后，分别用下面两个参数来控制

• sync queue - 半连接队列

  • 大小通过 /proc/sys/net/ipv4/tcp_max_syn_backlog 指定，在 syncookies 启用的情况下，逻辑上没有最大值限制，这个设置便被忽略

• accept queue - 全连接队列

  • 其大小通过 /proc/sys/net/core/somaxconn 指定，在使用 listen 函数时，内核会根据传入的 backlog 参数与系统参数，取二者的较小值
  • 如果 accpet queue 队列满了，server 将发送一个拒绝连接的错误信息到 client

netty 中控制全连接

可以通过 option(ChannelOption.SO_BACKLOG, 值) 来设置大小

测试

• 调试关键断点为：io.netty.channel.nio.NioEventLoop#processSelectedKey

/**
 * 服务器
 */
public static void main(String[] args) {
    NioEventLoopGroup group = new NioEventLoopGroup(2);
    try {
        Channel channel = new ServerBootstrap()
                .channel(NioServerSocketChannel.class)
                //设置全连接队列大小：2
                .option(ChannelOption.SO_BACKLOG,2)
                .group(group)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline().addLast(new LoggingHandler());
                    }
                })
                .bind().sync().channel();
        channel.closeFuture().sync();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } 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

/**
 * 客户端
 */
public static void main(String[] args) {
    NioEventLoopGroup group = new NioEventLoopGroup(2);
    try {
        Channel channel = new Bootstrap()
                .channel(NioSocketChannel.class)
                .group(group)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline().addLast(new LoggingHandler());
                    }
                })
                .connect(new InetSocketAddress("127.0.0.1", 8080)).sync().channel();
        channel.closeFuture().sync();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } 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

3）ulimit -n 数字

• 属于操作系统参数
• 属于临时调整，建议放在启动脚本里
• 限制一个进程能够同时打开的文件描述符的数量

4）TCP_NODELAY

• 属于 SocketChannal 参数
• Nagle 算法设置，默认为 true

5）SO_SNDBUF & SO_RCVBUF

• SO_SNDBUF 属于 SocketChannal 参数
• SO_RCVBUF 既可用于 SocketChannal 参数，也可以用于 ServerSocketChannal 参数（建议设置到 ServerSocketChannal 上）
• 现在的操作系统比较智能，会自动调整。不建议设置

6）ALLOCATOR

• 属于 SocketChannal 参数。ByteBuf 分配器
• 用来分配 ByteBuf， ctx.alloc()

7）RCVBUF_ALLOCATOR

• 属于 SocketChannal 参数
• 控制 netty 接收缓冲区大小
• 负责入站数据的分配，决定入站缓冲区的大小（并可动态调整），统一采用 direct 直接内存，具体池化还是非池化由 allocator 决定

1.3 RPC 框架

1）准备工作

这些代码可以认为是现成的，无需从头编写练习

为了简化起见，在原来聊天项目的基础上新增 Rpc 请求和响应消息

增加 Rpc 消息类型：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
14
15

Rpc请求消息：RpcRequestMessage

@Getter
@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

Rpc响应消息：RpcResponseMessage

@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

服务器架子

@Slf4j
public class RpcServer {
    public static void main(String[] args) {
        NioEventLoopGroup Boss=new NioEventLoopGroup(1);
        NioEventLoopGroup Worker=new NioEventLoopGroup(2);
        LoggingHandler LOGGING_HANDLER=new LoggingHandler();
        MessageCodeSharable MESSAGE_CODEC=new MessageCodeSharable();
        
        //1.RPC消息发送处理器
        RpcRequestMessageHandler RPC_REQUEST_HANDLER=new RpcRequestMessageHandler();
        try {
            ServerBootstrap serverBootstrap=new ServerBootstrap();
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.group(Boss,Worker);
            serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new ProcotolFrameDecoder()); //处理粘包半包
                    ch.pipeline().addLast(LOGGING_HANDLER); //记录日志
                    ch.pipeline().addLast(MESSAGE_CODEC); //消息协议 编码、解码

                    //1.RPC消息发送处理器
                    ch.pipeline().addLast(RPC_REQUEST_HANDLER);
                }
            });
            Channel channel = serverBootstrap.bind(8080).sync().channel();
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            log.debug("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

客户端架子

@Slf4j
public class RpcClient {
    public static void main(String[] args) {
        NioEventLoopGroup group=new NioEventLoopGroup(2);
        LoggingHandler LOGGING_HANDLER=new LoggingHandler();
        MessageCodeSharable MESSAGE_CODEC=new MessageCodeSharable();

        //1.RPC消息响应处理器
        RpcResponseMessageHandler RPC_RESPONSE_HANDLER=new RpcResponseMessageHandler();
        try {
            Bootstrap bootstrap=new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(group);
            bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new ProcotolFrameDecoder()); //处理粘包半包
                    ch.pipeline().addLast(LOGGING_HANDLER); //记录日志
                    ch.pipeline().addLast(MESSAGE_CODEC); //消息协议 编码、解码

                    //1.RPC消息响应处理器
                    ch.pipeline().addLast(RPC_RESPONSE_HANDLER);
                }
            });
            Channel channel = bootstrap.connect(new InetSocketAddress("127.0.0.1", 8080)).sync().channel();
            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

相关服务

• HelloService

  public interface HelloService {
      String sayHello(String msg);
  }
  1
  2
  3

• HelloServiceImpl

  //有所修改
  public class HelloServiceImpl implements HelloService{
      @Override
      public String sayHello(String msg) {
          System.out.println("你好!"+msg);
          return "你好!"+msg;
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8

• ServiceFactory：通过 Class对象 获取 该类的实例对象

  public abstract class ServiceFactory {
      static Properties properties;
      static Map<Class<?>,Object> map =new ConcurrentHashMap<>();
  
      static{
          try (InputStream in = Config.class.getResourceAsStream("/application.properties")){
              properties =new Properties();
              properties.load(in);
              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.newInstance());
                  }
              }
          }catch (IOException | ClassNotFoundException | InstantiationException | IllegalAccessException e){
              throw new ExceptionInInitializerError(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

• application.properties

  serializer.algorithm=Json
  org.example.java.chatRoom.server.service.HelloService=org.example.java.chatRoom.server.service.HelloServiceImpl
  1
  2

2）客户端发送 rpc请求

@Slf4j
public class RpcClient {
    public static void main(String[] args) {
        NioEventLoopGroup group=new NioEventLoopGroup(2);
        LoggingHandler LOGGING_HANDLER=new LoggingHandler();
        MessageCodeSharable MESSAGE_CODEC=new MessageCodeSharable();

        //2.RPC消息响应处理器
        RpcResponseMessageHandler RPC_RESPONSE_HANDLER=new RpcResponseMessageHandler();
        try {
            Bootstrap bootstrap=new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(group);
            bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new ProcotolFrameDecoder()); //处理粘包半包
                    ch.pipeline().addLast(LOGGING_HANDLER); //记录日志
                    ch.pipeline().addLast(MESSAGE_CODEC); //消息协议 编码、解码

                    //2.RPC消息响应处理器
                    ch.pipeline().addLast(RPC_RESPONSE_HANDLER);
                }
            });
            Channel channel = bootstrap.connect(new InetSocketAddress("127.0.0.1", 8080)).sync().channel();
            //1.连接完成后发送 Rpc请求消息
            ChannelFuture channelFuture = channel.writeAndFlush(new RpcRequestMessage(
                    0,
                    "org.example.java.chatRoom.server.service.HelloService",
                    "sayHello",
                    String.class,
                    new Class[]{String.class},
                    new Object[]{"张三"}));
            channelFuture.addListener(promise->{
                if(!promise.isSuccess()){
                    Throwable cause=promise.cause();
                    log.debug("error,{}",cause);
                }
            });
            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

3）服务器接收并响应

@Slf4j
@ChannelHandler.Sharable
public class RpcRequestMessageHandler extends SimpleChannelInboundHandler<RpcRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequestMessage message) throws Exception {
        RpcResponseMessage response = new RpcResponseMessage();
        response.setSequenceId(message.getSequenceId());
        try {
            //1.获取实现对象
            String interfaceName = message.getInterfaceName();
            HelloService service = (HelloService) ServiceFactory.getService(Class.forName(interfaceName));
            //2.调用方法
            String methodName = message.getMethodName();
            Class[] parameterTypes = message.getParameterTypes();
            Object[] parameterValue = message.getParameterValue();
            Method method = service.getClass().getMethod(methodName, parameterTypes);
            Object result = method.invoke(service, parameterValue);
            //3.设置返回值
            response.setReturnValue(result);
        } catch (ClassNotFoundException e) {
            log.debug("RpcRequestHandler erro,{}",e);
            response.setExceptionValue(e);
        } finally {
        }
        ctx.writeAndFlush(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
27

4）客户端简单处理响应

@Slf4j
@ChannelHandler.Sharable
public class RpcResponseMessageHandler extends SimpleChannelInboundHandler<RpcResponseMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage message) throws Exception {
        log.debug("{}", message);
    }
}
1
2
3
4
5
6
7
8

5）客户端2.0：管理Channel

包括 channel 管理，代理，接收结果

@Slf4j
public class RpcClientManager {
    public static void main(String[] args) {
        HelloService service=getProxyService(HelloService.class);
        //方法调用暂时没有考虑返回值
        service.sayHello("小红");
    }

    //创建代理类对象：代理调用服务端的方法
    public static <T> T getProxyService(Class<T> serviceClass){
        //类加载器
        ClassLoader classLoader = serviceClass.getClassLoader();
        //代理类的接口数组
        Class<?>[] classes = {serviceClass};
        //方法处理器
        Object o= Proxy.newProxyInstance(classLoader,classes,(proxy, method, args)->{
            //1.生成 rpc消息对象
            int serquenceId= SequenceIdGenerator.nextId();
            RpcRequestMessage message = new RpcRequestMessage(
                    serquenceId,
                    serviceClass.getName(),
                    method.getName(),
                    method.getReturnType(),
                    method.getParameterTypes(),
                    args);
            //2.发送消息
            getChannel().writeAndFlush(message);
            //3.暂时返回null
            return null;
        });
        return (T) o;
    }


    private static Channel channel=null;
    //获取单例channel
    public static Channel getChannel(){
        if(channel!=null) return channel;
        synchronized (RpcClientManager.class){
            if (channel!=null) return channel;
            initChannel();
            return channel;
        }
    }
    //初始化Channel
    private static void initChannel() {
        NioEventLoopGroup group=new NioEventLoopGroup(2);
        LoggingHandler LOGGING_HANDLER=new LoggingHandler();
        MessageCodeSharable MESSAGE_CODEC=new MessageCodeSharable();

        //RPC消息响应处理器
        RpcResponseMessageHandler RPC_RESPONSE_HANDLER=new RpcResponseMessageHandler();
        Bootstrap bootstrap=new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.group(group);
        bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ch.pipeline().addLast(new ProcotolFrameDecoder()); //处理粘包半包
                ch.pipeline().addLast(LOGGING_HANDLER); //记录日志
                ch.pipeline().addLast(MESSAGE_CODEC); //消息协议 编码、解码

                //RPC消息响应处理器
                ch.pipeline().addLast(RPC_RESPONSE_HANDLER);
            }
        });
        try {
            channel = bootstrap.connect(new InetSocketAddress("127.0.0.1", 8080)).sync().channel();
            channel.closeFuture().addListener(future -> {
                group.shutdownGracefully();
            });
        } catch (InterruptedException e) {
            log.debug("Client error,{}",e);
            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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

• SequenceIdGenerator

  public class SequenceIdGenerator {
      private static final AtomicInteger id=new AtomicInteger();
      public static int nextId(){
          return id.incrementAndGet();
      }
  }
  1
  2
  3
  4
  5
  6

6）客户端3.0：等待响应

发送 rpc请求 的线程等待响应结果

@Slf4j
public class RpcClientManager {
    public static void main(String[] args) {
        HelloService service=getProxyService(HelloService.class);
        System.out.println(service.sayHello("小红"));
        System.out.println(service.sayHello("小明"));
        System.out.println(service.sayHello("小紫"));
    }

    //创建代理类对象
    public static <T> T getProxyService(Class<T> serviceClass){
        //类加载器
        ClassLoader classLoader = serviceClass.getClassLoader();
        //代理类的接口数组
        Class<?>[] classes = {serviceClass};
        //方法处理器
        Object o= Proxy.newProxyInstance(classLoader,classes,(proxy, method, args)->{
            //1.生成 rpc消息对象
            int serquenceId= SequenceIdGenerator.nextId();
            RpcRequestMessage message = new RpcRequestMessage(
                    serquenceId,
                    serviceClass.getName(),
                    method.getName(),
                    method.getReturnType(),
                    method.getParameterTypes(),
                    args);
            //2.发送消息
            getChannel().writeAndFlush(message);
            //3.创建 promise容器 接收返回消息：并指定 promise对象 异步接收结果线程
            DefaultPromise<Object> promise=new DefaultPromise<>(getChannel().eventLoop());
            RpcResponseMessageHandler.PROMISES.put(serquenceId,promise);
            //4.等待结果返回
            promise.await();
            //5.返回结果
            if (promise.isSuccess()) {
                return promise.getNow();
            }else {
                throw new RuntimeException(promise.cause());
            }
        });
        return (T) o;
    }


    private static Channel channel=null;

    public static Channel getChannel(){
        if(channel!=null) return channel;
        synchronized (RpcClientManager.class){
            if (channel!=null) return channel;
            initChannel();
            return channel;
        }
    }

    private static void initChannel() {
        NioEventLoopGroup group=new NioEventLoopGroup(2);
        LoggingHandler LOGGING_HANDLER=new LoggingHandler();
        MessageCodeSharable MESSAGE_CODEC=new MessageCodeSharable();

        //RPC消息响应处理器
        RpcResponseMessageHandler RPC_RESPONSE_HANDLER=new RpcResponseMessageHandler();
        Bootstrap bootstrap=new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.group(group);
        bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ch.pipeline().addLast(new ProcotolFrameDecoder()); //处理粘包半包
                ch.pipeline().addLast(LOGGING_HANDLER); //记录日志
                ch.pipeline().addLast(MESSAGE_CODEC); //消息协议 编码、解码

                //RPC消息响应处理器
                ch.pipeline().addLast(RPC_RESPONSE_HANDLER);
            }
        });
        try {
            channel = bootstrap.connect(new InetSocketAddress("127.0.0.1", 8080)).sync().channel();
            channel.closeFuture().addListener(future -> {
                group.shutdownGracefully();
            });
        } catch (InterruptedException e) {
            log.debug("Client error,{}",e);
            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
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

处理响应线程设置返回结果

@Slf4j
@ChannelHandler.Sharable
public class RpcResponseMessageHandler extends SimpleChannelInboundHandler<RpcResponseMessage> {
    //响应后将返回结果与对应的请求进行设置
    public static final Map<Integer, Promise<Object>> PROMISES=new ConcurrentHashMap<>();

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage message) throws Exception {
        //将响应结果返回给调用者
        Promise<Object> promise = PROMISES.remove(message.getSequenceId());
        Object returnValue = message.getReturnValue();
        Exception exceptionValue = message.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

7）几个问题

• RpcResponseMessageHandler 中 PROMISES集合中的泛型问题：Promise