rocketmq源碼3－通信－協(xié)議

一 協(xié)議體

  • RemotingCommand請求報文信息
public class RemotingCommand {
    private int code;//請求碼
    private LanguageCode language = LanguageCode.JAVA;//語言
    private int version = 0;//rocketmq版本
    private int opaque = requestId.getAndIncrement();//請求id
    private int flag = 0;//報文類型標記
    private String remark;//描述信息
    private HashMap<String, String> extFields;//customHeader的屬性信息
    private transient CommandCustomHeader customHeader;
//序列化類型
    private SerializeType serializeTypeCurrentRPC = serializeTypeConfigInThisServer;
//之前的部分為協(xié)議報文頭
//報文體
    private transient byte[] body;
}
  • 設(shè)置版本信息
private static void setCmdVersion(RemotingCommand cmd) {
    if (configVersion >= 0) {
        cmd.setVersion(configVersion);
    } else {
        String v = System.getProperty(REMOTING_VERSION_KEY);
        if (v != null) {
            int value = Integer.parseInt(v);
            cmd.setVersion(value);
            configVersion = value;
        }
    }
}
  • createResponseCommand()創(chuàng)建報文體

1.1 編碼

  • 報文序列化格式為：總長度|序列化類型＋頭長度|頭|消息體，計算總長度，序列化數(shù)據(jù)
  • 計算總長度，依次存儲數(shù)據(jù)。
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();

    return result;
}
  • 報文頭序列化
private byte[] headerEncode() {
    this.makeCustomHeaderToNet();
    if (SerializeType.ROCKETMQ == serializeTypeCurrentRPC) {
        return RocketMQSerializable.rocketMQProtocolEncode(this);
    } else {
        return RemotingSerializable.encode(this);
    }
}
  • makeCustomHeaderToNet()把customHeader屬性信息轉(zhuǎn)換成報文頭的擴展域數(shù)據(jù)
  • markProtocolType(headerData.length, serializeTypeCurrentRPC)，第一位存儲序列化類型，后三位存儲報文頭長度

1.2 解碼

  • 獲取總長度信息，分配空間存儲數(shù)據(jù)
  • 獲取序列化類型，反序列化報文頭信息
  • 計算報文體長度，獲取報文體信息
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;
}

二 序列化

  • 兩種序列化類型
public enum SerializeType {
    JSON((byte) 0),
    ROCKETMQ((byte) 1);

    private byte code;

    SerializeType(byte code) {
        this.code = code;
    }

    public static SerializeType valueOf(byte code) {
        for (SerializeType serializeType : SerializeType.values()) {
            if (serializeType.getCode() == code) {
                return serializeType;
            }
        }
        return null;
    }

    public byte getCode() {
        return code;
    }
}

2.1 JSON

  • 編碼就是轉(zhuǎn)成json串
public static byte[] encode(final Object obj) {
    final String json = toJson(obj, false);
    if (json != null) {
        return json.getBytes(CHARSET_UTF8);
    }
    return null;
}

public static String toJson(final Object obj, boolean prettyFormat) {
    return JSON.toJSONString(obj, prettyFormat);
}
  • 解碼就是json解析
public static <T> T decode(final byte[] data, Class<T> classOfT) {
    final String json = new String(data, CHARSET_UTF8);
    return fromJson(json, classOfT);
}

public static <T> T fromJson(String json, Class<T> classOfT) {
    return JSON.parseObject(json, classOfT);
}

2.2 ROCKETMQ

2.2.1 編碼

  • 獲取remark信息
// String remark
byte[] remarkBytes = null;
int remarkLen = 0;
if (cmd.getRemark() != null && cmd.getRemark().length() > 0) {
    remarkBytes = cmd.getRemark().getBytes(CHARSET_UTF8);
    remarkLen = remarkBytes.length;
}
  • 獲取擴展域信息
// HashMap<String, String> extFields
byte[] extFieldsBytes = null;
int extLen = 0;
if (cmd.getExtFields() != null && !cmd.getExtFields().isEmpty()) {
    extFieldsBytes = mapSerialize(cmd.getExtFields());
    extLen = extFieldsBytes.length;
}
  • 計算報文總長度
private static int calTotalLen(int remark, int ext) {
    // int code(~32767)
    int length = 2
        // LanguageCode language
        + 1
        // int version(~32767)
        + 2
        // int opaque
        + 4
        // int flag
        + 4
        // String remark
        + 4 + remark
        // HashMap<String, String> extFields
        + 4 + ext;

    return length;
}
  • 分配內(nèi)存ByteBuffer headerBuffer = ByteBuffer.allocate(totalLen);
  • 寫入報文數(shù)據(jù)
// int code(~32767)，請求編碼
headerBuffer.putShort((short) cmd.getCode());
// LanguageCode language
headerBuffer.put(cmd.getLanguage().getCode());
// int version(~32767)，mcq版本
headerBuffer.putShort((short) cmd.getVersion());
// int opaque，請求id
headerBuffer.putInt(cmd.getOpaque());
// int flag，報文類型
headerBuffer.putInt(cmd.getFlag());
// String remark，描述信息。序列化時先寫入長度，后寫入數(shù)據(jù)
if (remarkBytes != null) {
    headerBuffer.putInt(remarkBytes.length);
    headerBuffer.put(remarkBytes);
} else {
    headerBuffer.putInt(0);
}
// HashMap<String, String> extFields; 擴展信息。序列化時先寫入長度，后寫入數(shù)據(jù)
if (extFieldsBytes != null) {
    headerBuffer.putInt(extFieldsBytes.length);
    headerBuffer.put(extFieldsBytes);
} else {
    headerBuffer.putInt(0);
}

return headerBuffer.array();
  • flag標識報文類型，
    第一位為1表示響應(yīng)報文，0表示請求報文
    第二位為1表示單向報文，0表示雙向報文
RPC_TYPE = 0;
RPC_ONEWAY = 1;
public void markResponseType() {
    int bits = 1 << RPC_TYPE;
    this.flag |= bits;
}
public void markOnewayRPC() {
    int bits = 1 << RPC_ONEWAY;
    this.flag |= bits;
}

2.2.1.1 mapSerialize

public static byte[] mapSerialize(HashMap<String, String> map) {
    // keySize+key+valSize+val
    if (null == map || map.isEmpty())
        return null;

    int totalLength = 0;
    int kvLength;
    Iterator<Map.Entry<String, String>> it = map.entrySet().iterator();
//計算序列化所需的總長度
    while (it.hasNext()) {
        Map.Entry<String, String> entry = it.next();
        if (entry.getKey() != null && entry.getValue() != null) {
            kvLength =
                // keySize + Key
                2 + entry.getKey().getBytes(CHARSET_UTF8).length
                    // valSize + val
                    + 4 + entry.getValue().getBytes(CHARSET_UTF8).length;
            totalLength += kvLength;
        }
    }

    ByteBuffer content = ByteBuffer.allocate(totalLength);
    byte[] key;
    byte[] val;
    it = map.entrySet().iterator();
//序列化存儲，key長度＋key值＋val長度＋val值
    while (it.hasNext()) {
        Map.Entry<String, String> entry = it.next();
        if (entry.getKey() != null && entry.getValue() != null) {
            key = entry.getKey().getBytes(CHARSET_UTF8);
            val = entry.getValue().getBytes(CHARSET_UTF8);

            content.putShort((short) key.length);
            content.put(key);

            content.putInt(val.length);
            content.put(val);
        }
    }

    return content.array();
}

2.2.2 解碼

  • 根據(jù)編碼規(guī)則反向解析數(shù)據(jù)為結(jié)構(gòu)化的對象。
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • 概述 RocketMQ 底層通訊是使用Netty來實現(xiàn)的。下面我們通過源碼分析下RocketMQ是怎么利用Nett...
    jijs閱讀 2,657評論 0贊 7
  • Swift1> Swift和OC的區(qū)別1.1> Swift沒有地址/指針的概念1.2> 泛型1.3> 類型嚴謹 對...
    cosWriter閱讀 11,671評論 1贊 32
  • 第一章 概述 一、協(xié)議定義 為網(wǎng)絡(luò)中互相通信的對等實體間進行數(shù)據(jù)交換二建立的規(guī)則、標準或約定，保證實體在計算機網(wǎng)絡(luò)...
    丿曰閱讀 14,822評論 0贊 6
  • 不知不覺天又黑下來了，我的簡書第五天日更還未完成，刷著簡書，刷著微信讀書，《寫出我心》讀了一半，初讀時激動...
    寒小雪閱讀 323評論 0贊 0
  • 一、寫在前面 二、梯度下降原理 三、梯度下降圖形與代碼表述 * coding:utf-8 * import num...
    氵幻世丨逐月丿閱讀 1,019評論 0贊 1

友情鏈接更多精彩內(nèi)容