Dubbo 系列之传输层(一)

一码到底 2020年09月14日 19次浏览

在讲解dubboTCP端的设计时,先了解下一些类的关系图。它们是如何组织在一起的,每个功能又是什么,接着在进一步深入了解其内涵。

类简介

1、Exchangers(交换器工具类) 用来创建TCP服务(bind)和建立客户端连接(connect)辅助类

2、Transporters(数据流传输工具类)用来创建TCP服务(bind)和建立客户端连接(connect)辅助类,Exchangers的底层内容依赖于Transporters,并且Transporters会根据SPI扩展,来适配合适的tcp通讯框架,比如netty,mina等。

3、Exchanger(交换器) 用来创建TCP链接,通过工具类Exchangers完成,该接口是一个SPI扩展,目前唯一仅有就是HeaderExchanger。从名字的含义可以得到,该协议是具有自定义协议头的交换器,所以取名HeaderExchanger。

4、Transporter(数据传输层) 用来创建TCP连接,通过工具类Transporters完成。它也是一个SPI扩展,比如NettyTransporter,MinaTransporter。

5、ExchangeClient (交换器客户端),Exchanger的connect()方法返回,即建立了TCP连接后,返回的客户端,接着就是通过该客户端与服务端通信,实例有HeaderExchangeClient、LazyConnectExchangeClient、ReferenceCountExchangeClient。之后分别讲解这3个,Exchangers工具类建立的连接客户端是HeaderExchangeClient。

6、ExchangeServer (交换器服务端端) Exchanger的bind()方法返回,即服务端监听的服务端实例,它监听这某个具体的tcp端口。默认实现是HeaderExchangeServer。

7、RemotingServer(远程的TCP服务端),ExchangeServer类也实现了该接口,代表其也是一个远程服务器,具体的实现有NettyServer,由Transporter的bind()方法返回,具体的Transporter返回相应的远程服务端。比如NettyTransporter#bind()返回NettyServer。

8、Client(TCP客户端),ExchangeClient类也实现了该接口,代表其也是一个TCP客户端,具体实现有NettyClient,由Transporter的connect()方法返回,具体的Transporter返回相应的TCP客户端。比如NettyTransporter#connect()返回NettyClient。

9、Channel (通信通道) ,每建立一个TCP链接就相应创建一个Channel。比如Netty建立连接后,就有一个Channel。这里的Channel指的是dubbo自己定义的一个channel。它与netty的channel建立关联,通过NettyChannel类,框架操作的是NettyChannel,而NettyChannel内部持有一个netty的channel对象。

10、HeaderExchangeChannel(交换器Channel,ExchangeChannel属于交换器Channel),它被HeaderExchangeClient客户端所持有,客户端就是通过HeaderExchangeChannel进行通信的,HeaderExchangeChannel内部持有一个具体的Channel。

11、ChannelHandler (通道处理器) 用来处理建立连接、发送请求、结束请求等操作的具体抽象。

12、ChannelHandlers(通道处理器工具类) 主要用来包裹封装具体的Channel,它的作用是通过消息类型,根据Dispatcher返回不同的

13、Dispatcher(消息派发器)

类型DispatcherChannelhandler作用
AllAllDispatcherAllChannelHandler所有的消息类型全部通过业务线程池处理
ConnectionConnectionOrderedDispatcherConnectionOrderedChannelHandler连接、断开消息单独通过一个线程池池来处理,其他的读写等消息通过业务线程池处理
DirectDirectDispatcherDirectChannelHandler所有的消息都通过IO线程池处理,不放到业务线程池中
ExecutionExecutionDispatcherExecutionChannelHandler请求消息在业务线程池处理,其他消息在IO线程池。
MessageMessageOnlyDispatcherMessageOnlyChannelHandler请求和响应消息在业务线程池处理,其他心跳,连接等消息在IO线程池处理

类关系图

bind.jpg

connect.jpg

试一把,Netty操作--客户端多线程,单链路(TCP)

1、定义传输消息

@Data
@ToString
public class SampleMessage {

    private String threadName;
    
    private String id;

    private String desc;
}

2、编写编码器

public class SampleEncoder extends MessageToByteEncoder<SampleMessage> {

    protected void encode(ChannelHandlerContext channelHandlerContext, SampleMessage sampleMessage, ByteBuf byteBuf) throws Exception {

        String threadName = sampleMessage.getThreadName();
        String id = sampleMessage.getId();
        String desc = sampleMessage.getDesc();

        byteBuf.writeInt(threadName.getBytes().length);
        byteBuf.writeBytes(threadName.getBytes());

        byteBuf.writeInt(id.getBytes().length);
        byteBuf.writeBytes(id.getBytes());


        byteBuf.writeInt(desc.getBytes().length);
        byteBuf.writeBytes(desc.getBytes());
        String str = sampleMessage.getThreadName() + ":" + sampleMessage.getDesc() + ":" + sampleMessage.getId();

        System.out.println(str);
    }
}

3、编写解码器

public class SampleDecoder extends ByteToMessageDecoder {

    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {

        byteBuf.markReaderIndex();

        String threadName = read(byteBuf);
        if (threadName == null) {
            byteBuf.resetReaderIndex();
            return;
        }

        String id = read(byteBuf);
        if (id == null) {
            byteBuf.resetReaderIndex();
            return;
        }

        String desc = read(byteBuf);
        if (desc == null) {
            byteBuf.resetReaderIndex();
            return;
        }

        SampleMessage sampleMessage = new SampleMessage();
        sampleMessage.setId(id);
        sampleMessage.setThreadName(threadName);
        sampleMessage.setDesc(desc);
        list.add(sampleMessage);
    }

    private String read(ByteBuf byteBuf) {
        if (canReadInt(byteBuf)) {
            int readInt = byteBuf.readInt();
            if (canReadN(byteBuf, readInt)) {
                byte[] bytes = new byte[readInt];
                byteBuf.readBytes(bytes);
                return new String(bytes);
            } 
        }
        return null;
    }


    private boolean canReadInt(ByteBuf byteBuf) {
        return canReadN(byteBuf, 4);
    }

    private boolean canReadN(ByteBuf byteBuf, int n) {
        if (!byteBuf.isReadable()) {
            return false;
        }
        return byteBuf.readableBytes() >= n;
    }
}

4、编写消息处理器

public class PrintChannelHandlers extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof SampleMessage) {
            SampleMessage sampleMessage = (SampleMessage) msg;
            System.out.println(sampleMessage.getThreadName() + ":" + sampleMessage.getId() + ":" + sampleMessage.getDesc());
        }
    }
    
}

5、编写服务端

public class NettyServerMain {

    public static void main(String[] args) {
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(new NioEventLoopGroup(1), new NioEventLoopGroup(12))
                .channel(NioServerSocketChannel.class)
                .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline()
                               // .addLast("log",new LoggingHandler(LogLevel.INFO))
                                .addLast("decoder", new SampleDecoder())
                                .addLast("encoder", new SampleEncoder())
                                .addLast("handler", new PrintChannelHandlers());
                    }
                });

        ChannelFuture channelFuture = serverBootstrap.bind(8888);
        channelFuture.syncUninterruptibly();
        System.out.println("链接前");
        Channel channel = channelFuture.channel();
        System.out.println("链接后");
    }
}

6、编写客户端

public class NettyClientMain {
    public static void main(String[] args) {
        NettyClientMain nettyClientMain = new NettyClientMain();
        nettyClientMain.open();
    }

    public void open() {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap = new Bootstrap();
        bootstrap.group(new NioEventLoopGroup(10))
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .channel(NioSocketChannel.class);

        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {

                ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
                        .addLast("decoder", new SampleDecoder())
                        .addLast("encoder", new SampleEncoder());
                //.addLast("handler", new PrintChannelHandlers());

            }
        });

        SocketAddress socketAddress = new InetSocketAddress("127.0.0.1", 8888);

        ChannelFuture future = bootstrap.connect(socketAddress);
        boolean ret = future.awaitUninterruptibly(3000, MILLISECONDS);

        if (ret && future.isSuccess()) {
            Channel newChannel = future.channel();
            doProcess(newChannel);
        }
    }

    private void doProcess(Channel channel) {

        AtomicLong atomicLong = new AtomicLong();
        for (int i = 0; i < 15; i++) {
            final char ch = (char) (i + 65);
            final String id = "id" + i;
            Thread t = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        SampleMessage sampleMessage = new SampleMessage();
                        sampleMessage.setThreadName(Thread.currentThread().getName());
                        sampleMessage.setDesc(getdes(ch));
                        sampleMessage.setId("id" + sampleMessage.getDesc().length() + "-" + atomicLong.getAndIncrement());
                        channel.writeAndFlush(sampleMessage);
                    }
                }
            });
            t.start();
        }
    }


    private String getdes(char a) {
        Random random = new Random();
        StringBuffer buffer = new StringBuffer();
        for (int i = 0; i < random.nextInt(500) + 1; i++) {
            buffer.append(a);
        }
        return buffer.toString();
    }
}

7、测试结果

result.jpg
结果符合预期,dubbo 也是通过服务底层公用一条TCP链接,多线程进行调用该链路channel。