Index: src/components/judger/mqtt_msg_proc/mqtt_msg_proc.c
===================================================================
--- src/components/judger/mqtt_msg_proc/mqtt_msg_proc.c	(revision 3331)
+++ src/components/judger/mqtt_msg_proc/mqtt_msg_proc.c	(working copy)
@@ -10,8 +10,8 @@
 #include "judger.h"
 #include "id_transform.h"
 #include "bmser_mqtt_upgrade.h"
+#include "bmser_ems_db_control_db.h"
 
-
 #define LOCAL_MQTT_IP               "localhost"
 
 #define DB_SET_TOPIC_FMT  \
@@ -20,6 +20,9 @@
 #define BIN_SET_TOPIC_FMT  \
     "ipc/%s/bms/+/upgrade/Set/binaryFlow"
 
+#define DB_GET_TOPIC_FMT  \
+    "ipc/%s/bms/+/upgrade/Get/binaryCmd"
+
 #define VERSION_RESP_FMT    \
     "{\"frames\":[{\"regs\":[{\"key\":\"bmserTrans_version\",\"value\":\"%s\"}]}]}"
 
@@ -32,6 +35,7 @@
     char gwSn[64];
     char devSn[64];
     char dbSetTopic[MQTT_TOPIC_MAX_LEN];
+    char dbGetTopic[MQTT_TOPIC_MAX_LEN];
     char binSetTopic[MQTT_TOPIC_MAX_LEN];
     char versionGetTopic[MQTT_TOPIC_MAX_LEN];
     struct mosquitto* mosq;
@@ -133,6 +137,59 @@
     return 0;
 }
 
+static int PublishGetRespBinaryCmd(struct DB_MQTT_MSG_S *pDbMsg, char *data, uint16_t bankID, uint16_t rackID)
+{
+    int ret = 0;
+    int msgSize = 0;
+    uint16_t devId = 0;
+    char topic[MQTT_TOPIC_MAX_LEN] = {0};
+    char mqttBuff[PAYLOAD_BUFFER_SIZE] = {0};
+    char devSN[64] = {0};
+
+    snprintf(devSN, 64, "000%02d%03d", (int)bankID, (int)rackID);
+    ret = bmser_dev_calcActualDeviceAddress(bankID, rackID, &devId);
+    if (ret)
+    {
+        BMSER_MyLogError("bmser_dev_calcActualDeviceAddress err!");
+        return -1;
+    }
+    pDbMsg->devId = devId;
+    msgSize = htons(pDbMsg->regCnt) * 2 + sizeof(struct DB_MQTT_MSG_S);
+
+    memcpy(mqttBuff, pDbMsg, \
+            sizeof(struct DB_MQTT_MSG_S));
+    memcpy(mqttBuff + sizeof(struct DB_MQTT_MSG_S), \
+            data, \
+            htons(pDbMsg->regCnt) * 2);
+
+    bmser_calculate_crc((uint8_t *)mqttBuff,    \
+                            msgSize);
+
+    snprintf(topic,    \
+            sizeof(topic), \
+            DB_GET_RESP_TOPIC_FMT, \
+            g_mqtt_mgr.gwSn, \
+            devSN);
+    
+    ret = mosquitto_publish(g_mqtt_mgr.mosq,  \
+                            NULL,   \
+                            topic,  \
+                            msgSize + CRC16_SIZE, \
+                            mqttBuff,   \
+                            1, \
+                            false);
+    if (ret != MOSQ_ERR_SUCCESS)
+    {
+        BMSER_MyLogError("mosquitto_publish failed, ret = %d", ret);
+
+        return -1;
+    }
+
+    BMSER_MyLogDebug("publish msg to %s", topic);
+
+    return 0;
+}
+
 int InitTopic()
 {
     int ret;
@@ -149,6 +206,11 @@
             MQTT_TOPIC_MAX_LEN, \
             DB_SET_TOPIC_FMT, \
             g_mqtt_mgr.gwSn);
+    
+    snprintf(g_mqtt_mgr.dbGetTopic,   \
+            MQTT_TOPIC_MAX_LEN, \
+            DB_GET_TOPIC_FMT, \
+            g_mqtt_mgr.gwSn);
 
     snprintf(g_mqtt_mgr.binSetTopic,   \
             MQTT_TOPIC_MAX_LEN, \
@@ -161,6 +223,7 @@
             g_mqtt_mgr.gwSn);
 
     BMSER_MyLogInfo("dbSetTopic = %s", g_mqtt_mgr.dbSetTopic);
+    BMSER_MyLogInfo("dbGetTopic = %s", g_mqtt_mgr.dbGetTopic);
     BMSER_MyLogInfo("binSetTopic = %s", g_mqtt_mgr.binSetTopic);
     BMSER_MyLogInfo("versionGetTopic = %s", g_mqtt_mgr.versionGetTopic);
 
@@ -352,6 +415,10 @@
                             MQTT_QOS_LEVEL);
     ret |= mosquitto_subscribe(mosq, \
                             NULL,   \
+                            g_mqtt_mgr.dbGetTopic,  \
+                            MQTT_QOS_LEVEL);
+    ret |= mosquitto_subscribe(mosq, \
+                            NULL,   \
                             g_mqtt_mgr.binSetTopic,  \
                             MQTT_QOS_LEVEL);
     ret |= mosquitto_subscribe(mosq, \
@@ -366,6 +433,7 @@
     }
 
     BMSER_MyLogDebug("subsribe success:%s", g_mqtt_mgr.dbSetTopic);
+    BMSER_MyLogDebug("subsribe success:%s", g_mqtt_mgr.dbGetTopic);
     BMSER_MyLogDebug("subsribe success:%s", g_mqtt_mgr.binSetTopic);
     BMSER_MyLogDebug("subsribe success:%s", g_mqtt_mgr.versionGetTopic);
 
@@ -406,6 +474,7 @@
     int ret = 0;
     int bcuid;
     int isBroadcast = 0;
+    uint16_t bankId = 0, rackId = 0, devId = 0;
     struct DB_MQTT_MSG_S *pDbMsg;
     struct BIN_MQTT_MSG_S *pBinMsg;
     BMSER_TRANS_COMM_TUNNEL_S tunnel = NULL;
@@ -441,11 +510,74 @@
             decode db msg
         */
         MqttMsgProc_DecodeMsg_Db(message, pMsg);
+        pDbMsg = (struct DB_MQTT_MSG_S *)message->payload;
 
-        pDbMsg = (struct DB_MQTT_MSG_S *)message->payload;
-        bcuid = pDbMsg->devId;
+        ret = bmser_dev_get_bankid_from_topic(message->topic, &bankId, &rackId);
+        if (ret)
+        {
+            BMSER_MyLogError("bmser_dev_get_bankid_from_topic err!");
+            return;
+        }
+
+        ret = bmser_dev_calcActualDeviceAddress(bankId, rackId, &devId);
+        if (ret)
+        {
+            BMSER_MyLogError("bmser_dev_calcActualDeviceAddress err!");
+            return;
+        }
+        /* 以topic devSN 计算出的簇设备ID为准 */
+        bcuid = (int)devId;
+        pMsg->info.dbInfo.devId = bcuid;
         tunnel = GetMasterTunnel(bcuid);
     }
+    else if (strstr(message->topic, DB_GET_TOPIC_KEY_NAME))
+    {
+        pDbMsg = (struct DB_MQTT_MSG_S *)message->payload;
+        EMS_DEV_REG_DATA_S regData = {0};
+
+        ret = bmser_dev_get_bankid_from_topic(message->topic, &bankId, &rackId);
+        if (ret)
+        {
+            BMSER_MyLogError("bmser_dev_get_bankid_from_topic err!");
+            return;
+        }
+
+        /* TODO 支持数据流动后，需要把有效位检验打开 */
+        // char valid = 0;
+        uint16_t dataBuf[TRANS_MSG_BUFFER_SIZE] = {0};
+        regData.bankId = bankId;
+        regData.rackId = rackId;
+        regData.startAddr = ntohs(pDbMsg->startAddr);
+        regData.regCount = ntohs(pDbMsg->regCnt);
+        regData.len = regData.regCount*2;
+        // regData.dbValid = &valid;
+
+        if(regData.regCount > TRANS_MSG_BUFFER_SIZE)
+        {
+            /* 单次查询数量限制为TRANS_MSG_BUFFER_SIZE */
+            BMSER_MyLogError("func READ: regCount over the limit(%d)!", \
+                TRANS_MSG_BUFFER_SIZE);
+            return;
+        }
+
+        ret = BmserProtocolExt_ReadRegisterDb(&regData, (uint8_t *)dataBuf, regData.len);
+        if(ret)
+        {
+            BMSER_MyLogError("read reg db fail addr = 0x%x, ret = %d.", \
+                regData.startAddr, ret);
+            /* TODO 读取失败再向下位机请求一次，需要走数据流动 */
+            return;
+        }
+
+        pDbMsg->func = FUNC_READ_DB_RESP;
+        ret = PublishGetRespBinaryCmd(pDbMsg, (char *)dataBuf, bankId, rackId);
+        if(ret)
+        {
+            BMSER_MyLogError("PublishGetRespBinaryCmd failed, ret = %d", ret);
+            return;
+        }
+        return;
+    }
     else if (strstr(message->topic, VERSION_GET_TOPIC_KEY_NAME))
     {
         ret = ResponseVersion();
Index: src/components/judger/mqtt_msg_proc/mqtt_msg_proc.h
===================================================================
--- src/components/judger/mqtt_msg_proc/mqtt_msg_proc.h	(revision 3331)
+++ src/components/judger/mqtt_msg_proc/mqtt_msg_proc.h	(working copy)
@@ -3,6 +3,7 @@
 
 #define BIN_SET_TOPIC_KEY_NAME  "/Set/binaryFlow"
 #define DB_SET_TOPIC_KEY_NAME   "/Set/binaryCmd"
+#define DB_GET_TOPIC_KEY_NAME   "/Get/binaryCmd"
 #define VERSION_GET_TOPIC_KEY_NAME   "/Get/VersionInfoList"
 #define VERSION_FUCK_CMD    "VersionInfoList"
 
@@ -24,6 +25,12 @@
 #define VERSION_RESP_TOPIC_FMT  \
     "ipc/%s/bms/00000000/data/GetResp/VersionInfoList"
 
+#define DB_GET_TOPIC_FMT    \
+    "ipc/%s/bms/+/upgrade/Get/binaryCmd"
+
+#define DB_GET_RESP_TOPIC_FMT   \
+    "ipc/%s/bms/%s/upgrade/GetResp/binaryCmd"
+
 int MqttMsgProc_Init();
 int MqttMsgProc_Start();
 // void MqttMsgProc_DecodeMsg_Bin();