RocketMQ高手之路系列之二:RocketMQ之消息通信

本文阅读 6 分钟
首页 代码,Java 正文

引言

本文主要从源码角度分析RocketMQ的底层通信机制以及RPC调用的过程。对于RocketMQ通信机制的深入理解,是我们分析和领会整个RocketMQ系统消息流转流程的基石。

  • 消息
  • 消息底层如何流转
  • 总结

一、消息

我们可以设想一下,如果自己是RocketMQ的设计者,我们该如何设计一个消息系统。大家都知道RocketMQ是消息中间件,那么首先要解决的问题就是消息本身该如何设计。因为整个系统中它是传递的对象,是数据的载体。那么对于消息本身来说,我们需要定义消息的格式,这样客户端与服务端之间可以遵循定义好的消息格式来进行通信。在RocketMQ体系中,通过RemotingCommand对象来进行交互,,它对数据进行了封装。

(1)消息协议设计以及编解码

我们看下RemotingCommand中的部分属性如下所示:

private int code;   //操作码
private LanguageCode language = LanguageCode.JAVA;   //实现语言
private int version = 0;   //程序版本
private int opaque = requestId.getAndIncrement();   //reqeustId,在同一个连接上的不同请求标识码,与响应消息中的相对应
private int flag = 0;    //区分是普通RPC还是onewayRPC的标志
private String remark;    //传输自定义文本信息
private HashMap<String, String> extFields;    //自定义扩展信息
private transient CommandCustomHeader customHeader;

我们再看下具体的消息的格式是怎样的,如下图所示: img 消息长度:总长度,四个字节存储,占用一个int类型;

序列化类型&消息头长度:同样占用一个int类型,第一个字节表示序列化类型,后面三个字节表示消息头长度;

消息头数据:经过序列化后的消息头数据;

消息主体数据:消息主体的二进制字节数据内容。

所以消息本身结构以及消息如何传递,都是我们需要了解的 img 图片来源于网络

我们看下remoting模块中的protocal文件夹下的相关定义类,如下所示: img 对于消息还需要进行编码以及解码操作来提高消息传递的效率,所以RemotingCommand 中还提供了编码、解码的方法。RemotingCommand 的encode()源码如下所示:

public ByteBuffer encode() { 
        // 1> header length size
        int length = 4; //存储头部长度

        // 2> header data length
        byte[] headerData = this.headerEncode(); //报文头部数据
        length += headerData.length;

        // 3> body data length
        if (this.body != null) { 
            length += body.length;
        }

        ByteBuffer result = ByteBuffer.allocate(4 + length);  //分配字节缓冲区

        // length
        result.putInt(length);

        // header length
        result.put(markProtocolType(headerData.length, serializeTypeCurrentRPC));

        // header data
        result.put(headerData);

        // body data;
        if (this.body != null) { 
            result.put(this.body);
        }

        result.flip();  //将缓冲区翻转,用于将ByteBuffer放到网络通道中进行传输

        return result;
    }

我们再看下消息的解码操作,源码中的实现如下所示:

//解码操作
public static RemotingCommand decode(final byte[] array) { 
        ByteBuffer byteBuffer = ByteBuffer.wrap(array);
        return decode(byteBuffer);
    }

//
public static RemotingCommand decode(final ByteBuffer byteBuffer) { 
        int length = byteBuffer.limit();
        int oriHeaderLen = byteBuffer.getInt();
        int headerLength = getHeaderLength(oriHeaderLen);

        byte[] headerData = new byte[headerLength];
        byteBuffer.get(headerData);

        RemotingCommand cmd = headerDecode(headerData, getProtocolType(oriHeaderLen));

        int bodyLength = length - 4 - headerLength;
        byte[] bodyData = null;
        if (bodyLength > 0) { 
            bodyData = new byte[bodyLength];
            byteBuffer.get(bodyData);
        }
        cmd.body = bodyData;

        return cmd;
    }

(2)消息传输方式 在RocketMQ系统中,消息的通信方式主要有三种: 1、可靠的同步传输; 2、可靠的异步传输; 3、单向传输;

二、消息如何流转

我们知道RocketMQ底层的通信是通过Netty实现。其中NettyRemotingClient与NettyRemotingServer是RemotingClient和RemotingServer这两个接口的实现类,也是比较重要的实现类。NettyRemotingClient与NettyRemotingServer也同时继承了NettyRemotingAbstract抽象类,该类抽象了invokeSync、invokeOneway等公用的方法实现。

(1)客户端请求发送 首先我们来看下 NettyRemotingClient中的重要源码分析,如下所示:

public class NettyRemotingClient extends NettyRemotingAbstract implements RemotingClient { 
...
}

我们可以看出客户端继承了NettyRemotingAbstract抽象类同时实现了RemotingClient接口。其中比较重要的方法包括同步调用、异步调用以及单向调用。我么通过异步调用的方式来举例,NettyRemotingClient 根据地址信息获取或者创建channel,接着会进行invokeAsyncImpl方法的调用,此时客户端会将数据流转给公共的抽象类NettyRemotingAbstract 进行统一处理,在此类中完成真正的请求发送动作。发送消息的源码如下所示:

public void invokeAsyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis,
        final InvokeCallback invokeCallback)
        throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException { 
        long beginStartTime = System.currentTimeMillis();
        //请求的ID
        final int opaque = request.getOpaque();
        boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) { 
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);
            long costTime = System.currentTimeMillis() - beginStartTime;
            if (timeoutMillis < costTime) { 
                once.release();
                throw new RemotingTimeoutException("invokeAsyncImpl call timeout");
            }
            //构建ResponseFuture 
            final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis - costTime, invokeCallback, once);
            //将ResponseFuture放入responseTable
            this.responseTable.put(opaque, responseFuture);
            try { 
                //使用Netty的channel进行请求数据的发送
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() { 
                    //消息发送后执行动作
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception { 
                        if (f.isSuccess()) { 
                            //如果消息发送成功,则直接将属性置为true,同时返回
                            responseFuture.setSendRequestOK(true);
                            return;
                        }
                        //如果失败
                        requestFail(opaque);
                        log.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
                    }
                });
            } catch (Exception e) { 
                //异常处理过程
                responseFuture.release();
                log.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
                throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
            }
        } else { 
            if (timeoutMillis <= 0) { 
                throw new RemotingTooMuchRequestException("invokeAsyncImpl invoke too fast");
            } else { 
                String info =
                    String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                        timeoutMillis,
                        this.semaphoreAsync.getQueueLength(),
                        this.semaphoreAsync.availablePermits()
                    );
                log.warn(info);
                throw new RemotingTimeoutException(info);
            }
        }
    }

此处有需要留意的部分,如下:

protected final ConcurrentMap<Integer /* opaque */, ResponseFuture> responseTable =
        new ConcurrentHashMap<Integer, ResponseFuture>(256);

opaque 表示请求码,用来标识不同的请求。这个请求码会与ResponseFuture映射在一起。

(2)服务端响应流程 服务端代码如下,继承了NettyRemotingAbstract同时实现了RemotingServer。

public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer { 
...
}

在该类中的start方法如下所示,它的主要作用是启动Netty服务器

@Override
    public void start() { 
        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
            nettyServerConfig.getServerWorkerThreads(),
            new ThreadFactory() { 

                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) { 
                    return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
                }
            });

        ServerBootstrap childHandler =
            this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
                .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
                .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
                .childHandler(new ChannelInitializer<SocketChannel>() { 
                    //向管道中添加数据处理逻辑
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception { 
                        ch.pipeline()
                            .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME,
                                new HandshakeHandler(TlsSystemConfig.tlsMode))
                            .addLast(defaultEventExecutorGroup,
                                new NettyEncoder(),
                                new NettyDecoder(),
                                new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                                new NettyConnectManageHandler(),
                                new NettyServerHandler()
                            );
                    }
                });

        if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) { 
            childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        }

        try { 
            ChannelFuture sync = this.serverBootstrap.bind().sync();
            InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
            this.port = addr.getPort();
        } catch (InterruptedException e1) { 
            throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
        }

        if (this.channelEventListener != null) { 
            this.nettyEventExecutor.start();
        }

        this.timer.scheduleAtFixedRate(new TimerTask() { 

            @Override
            public void run() { 
                try { 
                    NettyRemotingServer.this.scanResponseTable();
                } catch (Throwable e) { 
                    log.error("scanResponseTable exception", e);
                }
            }
        }, 1000 * 3, 1000);
    }

本文主要介绍了RocketMQ在消息传递过程中底层通信的过程,同时阐述了消息的格式以及相应的编解码的过程。下篇文章会和大家来一起继续拆解RocketMQ上层的流程。

本文为互联网自动采集或经作者授权后发布,本文观点不代表立场,若侵权下架请联系我们删帖处理!文章出自:https://blog.csdn.net/Diamond_Tao/article/details/88924103
-- 展开阅读全文 --
大白话讲解JDK源码系列:从头到尾再讲一遍ThreadLocal
« 上一篇 01-30
KillDefender 的 Beacon 对象文件 PoC 实现
下一篇 » 02-09

发表评论

成为第一个评论的人

热门文章

标签TAG

最近回复