diff --git a/CMakeLists.txt b/CMakeLists.txt
index bd9d13a0..cce7435f 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -128,6 +128,7 @@ set(NEURON_SOURCES
     src/core/node_manager.c
     src/core/template_manager.c
     src/core/storage.c
+    src/adapter/msg_q.c
     src/adapter/storage.c
     src/adapter/adapter.c
     src/adapter/driver/cache.c
diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c
index 789dab89..cb6219d1 100644
--- a/src/adapter/adapter.c
+++ b/src/adapter/adapter.c
@@ -39,11 +39,12 @@
 #include "plugin.h"
 #include "storage.h"
 
-static int adapter_trans_data(enum neu_event_io_type type, int fd,
-                              void *usr_data);
-static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
-static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
-                           void *data);
+static void *adapter_consumer(void *arg);
+static int   adapter_trans_data(enum neu_event_io_type type, int fd,
+                                void *usr_data);
+static int   adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
+static int   adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
+                             void *data);
 static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header,
                             void *data);
 static int adapter_responseto(neu_adapter_t *     adapter,
@@ -110,6 +111,27 @@ void neu_adapter_set_error(int error)
     create_adapter_error = error;
 }
 
+static void *adapter_consumer(void *arg)
+{
+    neu_adapter_t *adapter = (neu_adapter_t *) arg;
+
+    while (1) {
+        neu_msg_t *         msg    = NULL;
+        uint32_t            n      = adapter_msg_q_pop(adapter->msg_q, &msg);
+        neu_reqresp_head_t *header = neu_msg_get_header(msg);
+
+        nlog_debug("adapter(%s) recv msg from: %s %p, type: %s, %u",
+                   adapter->name, header->sender, header->ctx,
+                   neu_reqresp_type_string(header->type), n);
+        adapter->module->intf_funs->request(
+            adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+        neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
+        neu_msg_free(msg);
+    }
+
+    return NULL;
+}
+
 static inline zlog_category_t *get_log_category(const char *node)
 {
     char name[NEU_NODE_NAME_LEN] = { 0 };
@@ -181,6 +203,9 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
         neu_adapter_driver_init((neu_adapter_driver_t *) adapter);
         break;
     case NEU_NA_TYPE_APP: {
+        adapter->msg_q = adapter_msg_q_new(adapter->name, 1024);
+        pthread_create(&adapter->consumer_tid, NULL, adapter_consumer,
+                       (void *) adapter);
         while (true) {
             uint16_t           port  = neu_manager_get_port();
             struct sockaddr_in local = {
@@ -596,11 +621,18 @@ static int adapter_trans_data(enum neu_event_io_type type, int fd,
         return 0;
     }
 
+    if (header->type == NEU_REQRESP_TRANS_DATA) {
+        if (adapter_msg_q_push(adapter->msg_q, msg) < 0) {
+            nlog_warn("adapter: %s trans data msg q is full, drop msg",
+                      adapter->name);
+            neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
+            neu_msg_free(msg);
+        }
+        return 0;
+    }
+
     adapter->module->intf_funs->request(
         adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
-    if (header->type == NEU_REQRESP_TRANS_DATA) {
-        neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
-    }
     neu_msg_free(msg);
     return 0;
 }
@@ -1256,6 +1288,13 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
         neu_node_metrics_free(adapter->metrics);
     }
 
+    if (adapter->consumer_tid != 0) {
+        pthread_cancel(adapter->consumer_tid);
+    }
+    if (adapter->msg_q != NULL) {
+        adapter_msg_q_free(adapter->msg_q);
+    }
+
     char *setting = NULL;
     if (adapter_load_setting(adapter->name, &setting) != 0) {
         remove_logs(adapter->name);
diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h
index 0fd28be9..63e09cd8 100644
--- a/src/adapter/adapter_internal.h
+++ b/src/adapter/adapter_internal.h
@@ -26,6 +26,7 @@
 
 #include "adapter_info.h"
 #include "core/manager.h"
+#include "msg_q.h"
 
 struct neu_adapter {
     char *name;
@@ -45,6 +46,9 @@ struct neu_adapter {
     int control_fd;
     int trans_data_fd;
 
+    adapter_msg_q_t *msg_q;
+    pthread_t        consumer_tid;
+
     uint16_t trans_data_port;
 
     neu_events_t *events;
diff --git a/src/adapter/msg_q.c b/src/adapter/msg_q.c
new file mode 100644
index 00000000..6e2b1bde
--- /dev/null
+++ b/src/adapter/msg_q.c
@@ -0,0 +1,116 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+#include <pthread.h>
+
+#include "utils/log.h"
+#include "utils/utextend.h"
+
+#include "msg_q.h"
+
+struct item {
+    neu_msg_t *  msg;
+    struct item *prev, *next;
+};
+
+struct adapter_msg_q {
+    struct item *list;
+    uint32_t     max;
+    uint32_t     current;
+    char *       name;
+
+    pthread_mutex_t mtx;
+    pthread_cond_t  cond;
+};
+
+adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size)
+{
+    struct adapter_msg_q *q = calloc(1, sizeof(struct adapter_msg_q));
+
+    pthread_mutex_init(&q->mtx, NULL);
+    pthread_cond_init(&q->cond, NULL);
+    q->max     = size;
+    q->name    = strdup(name);
+    q->current = 0;
+
+    return q;
+}
+
+void adapter_msg_q_free(adapter_msg_q_t *q)
+{
+    struct item *tmp = NULL, *elt = NULL;
+    nlog_warn("app: %s, drop %u msg", q->name, q->current);
+    pthread_mutex_destroy(&q->mtx);
+    pthread_cond_destroy(&q->cond);
+
+    DL_FOREACH_SAFE(q->list, elt, tmp)
+    {
+        DL_DELETE(q->list, elt);
+        neu_reqresp_head_t *header = neu_msg_get_header(elt->msg);
+        neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
+        neu_msg_free(elt->msg);
+        free(elt);
+    }
+    free(q->name);
+    free(q);
+}
+
+int adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg)
+{
+    int ret = -1;
+    pthread_mutex_lock(&q->mtx);
+    if (q->current < q->max) {
+        q->current += 1;
+        struct item *elt = calloc(1, sizeof(struct item));
+        elt->msg         = msg;
+        DL_APPEND(q->list, elt);
+        ret = 0;
+    }
+    pthread_mutex_unlock(&q->mtx);
+
+    if (ret == -1) {
+        nlog_warn("app: %s, msg q is full, %u(%u)", q->name, q->current,
+                  q->max);
+    } else {
+        pthread_cond_signal(&q->cond);
+    }
+
+    return ret;
+}
+
+uint32_t adapter_msg_q_pop(adapter_msg_q_t *q, neu_msg_t **p_data)
+{
+    uint32_t ret = 0;
+
+    pthread_mutex_lock(&q->mtx);
+    while (q->current == 0) {
+        pthread_cond_wait(&q->cond, &q->mtx);
+    }
+    struct item *elt = DL_LAST(q->list);
+
+    if (elt != NULL) {
+        DL_DELETE(q->list, elt);
+        *p_data = elt->msg;
+        free(elt);
+        q->current -= 1;
+        ret = q->current;
+    }
+
+    pthread_mutex_unlock(&q->mtx);
+    return ret;
+}
diff --git a/src/adapter/msg_q.h b/src/adapter/msg_q.h
new file mode 100644
index 00000000..215c93ff
--- /dev/null
+++ b/src/adapter/msg_q.h
@@ -0,0 +1,36 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#ifndef ADAPTER_MSG_Q_H
+#define ADAPTER_MSG_Q_H
+
+#include <stdint.h>
+
+#include "base/msg_internal.h"
+#include "msg.h"
+
+typedef struct adapter_msg_q adapter_msg_q_t;
+
+adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size);
+void             adapter_msg_q_free(adapter_msg_q_t *q);
+
+int      adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg);
+uint32_t adapter_msg_q_pop(adapter_msg_q_t *q, neu_msg_t **p_data);
+
+#endif
\ No newline at end of file