MetaQ源码分析——消息传输流程

朴灿烈づ我的快乐病毒、 2021-12-14 04:27 293阅读 0赞

Broker接收从Producer(Client端)发送的消息，也能够返回消息到Consumer(Client)，对于Broker来说，就是网络输入输出流的处理。

Broker使用淘宝内部的gecko框架作为网络传输框架，gecko是一个NIO框架，有很多优秀的特性。

由于网络模块与其他模块关联性极强，不像存储模块可以独立分析，所以此篇文章开始将从全局开始分析Broker。

首先看一下Broker的启动类MetamorphosisStartup：  
![在这里插入图片描述][20190707202846425.png]

public static void main(final String[] args) { 
            final String configFilePath = getConfigFilePath(args);
            final MetaConfig metaConfig = getMetaConfig(configFilePath);
            final MetaMorphosisBroker server = new MetaMorphosisBroker(metaConfig);
            server.start();
    
        }

*  先加载配置文件
 *  构造MetaMorphosisBroker实例
 *  调用该实例的start方法
 *  可以看到，真正的启动类是MetaMorphosisBroker

接下来看看MetaMorphosisBroker里面都干了什么事情：

public MetaMorphosisBroker(final MetaConfig metaConfig) { 
            //配置信息
    	this.metaConfig = metaConfig;
            //Broker对外提供的nio Server
    	this.remotingServer = newRemotingServer(metaConfig);
           //线程池管理器，主要是提供给nio Server在并发环境下可以使用多线程处理，提高性能
            this.executorsManager = new ExecutorsManager(metaConfig);
           //全局唯一的id生成器 
            this.idWorker = new IdWorker(metaConfig.getBrokerId());
           //存储模块管理器
            this.storeManager = new MessageStoreManager(metaConfig, this.newDeletePolicy(metaConfig));
           //统计模块管理器
    	this.statsManager = new StatsManager(this.metaConfig, this.storeManager, this.remotingServer);
          //zookeeper客户端，前面介绍过metaq使用zookeeper作为中间协调者，Broker会将自己注册到zookeeper上，也会从zookeeper查询相关数据
    	this.brokerZooKeeper = new BrokerZooKeeper(metaConfig);
          //网络输入输出流处理器 
            final BrokerCommandProcessor next = new BrokerCommandProcessor(this.storeManager, this.executorsManager,          this.statsManager, this.remotingServer, metaConfig, this.idWorker, this.brokerZooKeeper);
         //事务存储引擎
    	JournalTransactionStore transactionStore = null;
    	try { 
    		transactionStore = new JournalTransactionStore(metaConfig.getDataLogPath(), this.storeManager, metaConfig);
    	} catch (final Exception e) { 
    		throw new MetamorphosisServerStartupException("Initializing transaction store failed.", e);
    	}
          //带事务处理的网络输入输出流处理器，设计采用了责任链的设计模式，使用事务存储引擎存储中间结果
    	this.brokerProcessor = new TransactionalCommandProcessor(metaConfig, this.storeManager, this.idWorker, next, transactionStore, this.statsManager);
          //钩子，JVM退出钩子，钩子实现在JVM退出的时候尽力正确关闭 MetaMorphosisBroker 
            this.shutdownHook = new ShutdownHook();
          //注册钩子
    	Runtime.getRuntime().addShutdownHook(this.shutdownHook);
         //注册MBean，因为MetaMorphosisBroker实现MetaMorphosisBrokerMBean接口，可以将自己作为MBean注册到MBeanServer
    	MetaMBeanServer.registMBean(this, null);
    }

--------------------

## 请求 ##

Gecko采用请求/响应的方式组织传输。MetaQ依据定义了请求和响应的命令，由于命令Client和Broker均需要使用，所以放在了common工程的类MetaEncodeCommand，这个类定义了协议编码接口和常量：

public interface MetaEncodeCommand { 
        /** * 编码协议 * * @return 编码后的buffer */
        public IoBuffer encode();
    
        byte SPACE = (byte) ' ';
        byte[] CRLF = {  '\r', '\n' };
        public String GET_CMD = "get";
        public String RESULT_CMD = "result";
        public String OFFSET_CMD = "offset";
        public String VALUE_CMD = "value";
        public String PUT_CMD = "put";
        public String SYNC_CMD = "sync";
        public String QUIT_CMD = "quit";
        public String VERSION_CMD = "version";
        public String STATS_CMD = "stats";
        public String TRANS_CMD = "transaction";
    }

*  在分析Broker启动类MetaMorphosisBroker时，分析过registerProcessors()方法，针对于不同的请求，Broker注册了不同的处理器，详情见registerProcessors()方法。
 *  MetaQ传输采用文本协议设计，非常透明，MetaEncodeCommand定义是请求类型。
 *  Broker分为请求和响应命令，请求的类图如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70]  
注意：

*  所有的请求的命令均继承AbstractRequestCommand类并实现RequestCommand接口
 *  RequestCommand是Gecko框架定义的接口，所有的命令均在编码后能被Gecko框架组织传输，传输协议是透明的文本协议。
 *  AbstractRequestCommand定义了基本属性topic和opaque。

看一下AbstractRequestCommand代码：

public abstract class AbstractRequestCommand implements RequestCommand, MetaEncodeCommand { 
        private Integer opaque;
        private String topic;
        static final long serialVersionUID = -1L;
    
    
        public AbstractRequestCommand(final String topic, final Integer opaque) { 
            super();
            this.topic = topic;
            this.opaque = opaque;
        }

*  opaque主要用来标识请求，响应的时候该标识被带回，用于客户端区分哪个请求的响应

Meta是通讯是走TCP长连接，它的的协议是基于文本行的协议，类似memcached的文本协议。通用的协议格式如下

command params opaque\r\n
    [body]

*  command为协议命令
 *  params为参数列表
 *  opaque为协议的自增序列号，用于请求和应答的映射：客户端发送协议的时候需要自增此序列号，而服务端将拷贝来自客户端的序列号并作为应答的序列号返回，客户端可根据应答的序列号将应答和请求对应起来。
 *  body为协议体，可选，在协议头里需要必须有字段指明body长度。

## 详细请求介绍 ##

### 1. put：发送消息协议 ###

这个是transactionKey不为空的情况下

ByteUtils.setArguments(buffer, MetaEncodeCommand.PUT_CMD, this.getTopic(), 
    this.partition, dataLen, this.flag, this.checkSum, transactionKey, this.getOpaque());
    if (this.data != null) { 
                buffer.put(this.data);
            }

*  首先是PUT\_CMD也就是put命令
 *  然后是topic
 *  然后是partition
 *  然后是数据长度
 *  然后是flag为消息标志位
 *  checkSum为更改次数
 *  transactionKey可选
 *  最后是Opaque，也就是协议自增序号
 *  最后是数据

可以看到协议格式：

put topic partition value-length flag checksum * [transactionkey] opaque\r\n data

简单的例子：

put meta-test 0 5 0 1\r\nhello

### 2. get拉取消息协议 ###

get代码：

ByteUtils.setArguments(buffer, MetaEncodeCommand.GET_CMD, this.getTopic(), this.getGroup(), this.partition,
                this.offset, this.maxSize, this.getOpaque());

可以看到get的格式：

*  首先是协议命令get
 *  然后是topic
 *  然后是group也就是消费者分组名称
 *  partition为拉取的分区
 *  offset为拉取的起始偏移量
 *  maxSize为本次拉取的最大数据量大小  
    简单的例子：

get meta-test example 0 1024 512 1\r\n

--------------------

## 响应 ##

响应命令的类图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 1]

*  带有消息的响应集合DataCommand
 *  带有其他结果的响应BooleanCommand
 *  而opaque为协议的自增序列号，用于请求和应答的映射。

### 1. DataCommand:Get请求返回的应答 ###

value total-length opaque\r\n data

其中data的结构如下：

* <li>4个字节的消息数据长度（可能包括属性）</li>
     * <li>4个字节的check sum</li>
     * <li>8个字节的消息id</li>
     * <li>4个字节的flag</li>

1.  4个字节的消息数据长度
2.  4个字节的check sum
3.  8个字节的消息id
4.  4个字节的flag

消息数据，如果有属性(flag的最低位为1，即 (flag & 0x1) == 1 )，则为：

*  4个字节的属性长度+ 消息属性 + payload

否则为：

*  payload

示范：  
get请求，得到一个没有attribute，内容是"FFFF"的消息：

value 24 -214748 \r\n \x00\x00\x00\x04  \x52\xb0\x25\xa9  \xb4\x1a\xda\x94\xd0\x00\x00\x00  \x00\x00\x00\x00  FFFF

get请求，得到一个attribute为"AAA"，内容是"FFF"的消息：

value 24 -214748 \r\n \x00\x00\x00\x0b  \x55\x17\x9a\x47  \xb4\x1a\xdc\x98\x86\xc0\x00\x00  \x00\x00\x00\x01  \x00\x00\x00\x03  AAAFFF

### BooleanCommand：Result通用应答协议 ###

result message messageLen\r\n

例如：

result 200 1\r\n

分析BooleanCommand源码：

package com.taobao.metamorphosis.network;
    
    import com.taobao.gecko.core.buffer.IoBuffer;
    import com.taobao.gecko.core.command.ResponseStatus;
    import com.taobao.gecko.core.command.kernel.BooleanAckCommand;
    
    
    /** * 应答命令,协议格式如下：</br> result code length opaque\r\n message * * @author boyan * @Date 2011-4-19 * */
    public class BooleanCommand extends AbstractResponseCommand implements BooleanAckCommand { 
    
        private String message;
    
        /** * status code in http protocol */
        private final int code;
        static final long serialVersionUID = -1L;
    
    
        public BooleanCommand(final int code, final String message, final Integer opaque) { 
            super(opaque);
            this.code = code;
            switch (this.code) { 
            case HttpStatus.Success:
                this.setResponseStatus(ResponseStatus.NO_ERROR);
                break;
            default:
                this.setResponseStatus(ResponseStatus.ERROR);
                break;
            }
            this.message = message;
        }
    
    
        @Override
        public String getErrorMsg() { 
            return this.message;
        }
    
    
        public int getCode() { 
            return this.code;
        }
    
    
        @Override
        public void setErrorMsg(final String errorMsg) { 
            this.message = errorMsg;
    
        }
    
    
        @Override
        public IoBuffer encode() { 
        //对消息进行编码，方便在网络上面传输
            final byte[] bytes = ByteUtils.getBytes(this.message);
            final int messageLen = bytes == null ? 0 : bytes.length;
            final IoBuffer buffer =
                    IoBuffer.allocate(11 + ByteUtils.stringSize(this.code) + ByteUtils.stringSize(this.getOpaque())
                            + ByteUtils.stringSize(messageLen) + messageLen);
            ByteUtils.setArguments(buffer, MetaEncodeCommand.RESULT_CMD, this.code, messageLen, this.getOpaque());
            if (bytes != null) { 
                buffer.put(bytes);
            }
            buffer.flip();
            return buffer;
        }
    
    
        @Override
        public int hashCode() { 
            final int prime = 31;
            int result = super.hashCode();
            result = prime * result + this.code;
            result = prime * result + (this.message == null ? 0 : this.message.hashCode());
            return result;
        }
    
    
        @Override
        public boolean equals(final Object obj) { 
            if (this == obj) { 
                return true;
            }
            if (!super.equals(obj)) { 
                return false;
            }
            if (this.getClass() != obj.getClass()) { 
                return false;
            }
            final BooleanCommand other = (BooleanCommand) obj;
            if (this.code != other.code) { 
                return false;
            }
            if (this.message == null) { 
                if (other.message != null) { 
                    return false;
                }
            }
            else if (!this.message.equals(other.message)) { 
                return false;
            }
            return true;
        }
    
    
        @Override
        public boolean isBoolean() { 
            return true;
        }
    
    }

首先，响应状态码遵循http协议：

public class HttpStatus { 
        public static final int BadRequest = 400;
        public static final int NotFound = 404;
        public static final int Forbidden = 403;
        public static final int Unauthorized = 401;
    
        public static final int InternalServerError = 500;
        public static final int ServiceUnavilable = 503;
        public static final int GatewayTimeout = 504;
    
        public static final int Success = 200;
    
        public static final int Moved = 301;
    }

### OffsetCommand：查询最近有效的offset格式 ###

offset topic group partition offset\r\n

查询离某个offset的最近有效的offset,topic为查询的消息主题，group为消费者分组名称，partition为查询的分区,offset为查询的offset，示范：

offset meta-test example 0 1024 1\r\n

[20190707202846425.png]: /images/20211213/e027c8c86b874766bd6d89a9d8bdf7b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70]: /images/20211213/3b00007d41bc4a498930348d6b7665fd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 1]: /images/20211213/3004be3321ba4aaab76a81c901880fc4.png