diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/ConsumerWithReplyBytes.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/ConsumerWithReplyBytes.java
new file mode 100644
index 0000000..850fbd3
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/ConsumerWithReplyBytes.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.samples.springboot.consumer;
+
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQReplyListener;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Service;
+
+/**
+ * The consumer that replying bytes
+ */
+@Service
+@RocketMQMessageListener(topic = "${demo.rocketmq.bytesRequestTopic}", consumerGroup = "${demo.rocketmq.bytesRequestConsumer}", selectorExpression = "${demo.rocketmq.tag}")
+public class ConsumerWithReplyBytes implements RocketMQReplyListener<MessageExt, byte[]> {
+
+    @Override
+    public byte[] onMessage(MessageExt message) {
+        System.out.printf("------- ConsumerWithReplyBytes received: %s \n", message);
+        return "reply message content".getBytes();
+    }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/ConsumerWithReplyGeneric.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/ConsumerWithReplyGeneric.java
new file mode 100644
index 0000000..e17e675
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/ConsumerWithReplyGeneric.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.samples.springboot.consumer;
+
+import org.apache.rocketmq.samples.springboot.domain.ProductWithPayload;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQReplyListener;
+import org.springframework.stereotype.Service;
+
+/**
+ * The consumer that replying generic type
+ */
+@Service
+@RocketMQMessageListener(topic = "${demo.rocketmq.genericRequestTopic}", consumerGroup = "${demo.rocketmq.genericRequestConsumer}", selectorExpression = "${demo.rocketmq.tag}")
+public class ConsumerWithReplyGeneric implements RocketMQReplyListener<String, ProductWithPayload<String>> {
+    @Override
+    public ProductWithPayload<String> onMessage(String message) {
+        System.out.printf("------- ConsumerWithReplyGeneric received: %s \n", message);
+        return new ProductWithPayload<String>("replyProductName", "product payload");
+    }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/ObjectConsumerWithReplyUser.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/ObjectConsumerWithReplyUser.java
new file mode 100644
index 0000000..f66b003
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/ObjectConsumerWithReplyUser.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.samples.springboot.consumer;
+
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.samples.springboot.domain.User;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQReplyListener;
+import org.springframework.stereotype.Service;
+
+/**
+ * The consumer that replying Object
+ */
+@Service
+@RocketMQMessageListener(topic = "${demo.rocketmq.objectRequestTopic}", consumerGroup = "${demo.rocketmq.objectRequestConsumer}", selectorExpression = "${demo.rocketmq.tag}")
+public class ObjectConsumerWithReplyUser implements RocketMQReplyListener<User, User> {
+
+    @Override
+    public User onMessage(User user) {
+        System.out.printf("------- ObjectConsumerWithReplyUser received: %s \n", user);
+        User replyUser = new User();
+        replyUser.setUserAge((byte) 10);
+        replyUser.setUserName("replyUserName");
+        return replyUser;
+    }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumer.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumer.java
index 3801d48..11ac489 100644
--- a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumer.java
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumer.java
@@ -22,7 +22,7 @@
 import org.springframework.stereotype.Service;
 
 /**
- * RocketMQMessageListener
+ * StringConsumer
  */
 @Service
 @RocketMQMessageListener(topic = "${demo.rocketmq.topic}", consumerGroup = "string_consumer", selectorExpression = "${demo.rocketmq.tag}")
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumerWithReplyString.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumerWithReplyString.java
new file mode 100644
index 0000000..b194bc6
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumerWithReplyString.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.samples.springboot.consumer;
+
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQReplyListener;
+import org.springframework.stereotype.Service;
+
+/**
+ * The consumer that replying String
+ */
+@Service
+@RocketMQMessageListener(topic = "${demo.rocketmq.stringRequestTopic}", consumerGroup = "${demo.rocketmq.stringRequestConsumer}", selectorExpression = "${demo.rocketmq.tag}")
+public class StringConsumerWithReplyString implements RocketMQReplyListener<String, String> {
+
+    @Override
+    public String onMessage(String message) {
+        System.out.printf("------- StringConsumerWithReplyString received: %s \n", message);
+        return "reply string";
+    }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/ProductWithPayload.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/ProductWithPayload.java
new file mode 100644
index 0000000..e241929
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/ProductWithPayload.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.samples.springboot.domain;
+
+public class ProductWithPayload<T> {
+    private String productName;
+    private T payload;
+
+    public ProductWithPayload() {
+    }
+
+    public ProductWithPayload(String productName, T payload) {
+        this.productName = productName;
+        this.payload = payload;
+    }
+
+    public String getProductName() {
+        return productName;
+    }
+
+    public void setProductName(String productName) {
+        this.productName = productName;
+    }
+
+    public T getPayload() {
+        return payload;
+    }
+
+    public void setPayload(T payload) {
+        this.payload = payload;
+    }
+
+    @Override public String toString() {
+        return "ProductWithPayload{" +
+            "productName='" + productName + '\'' +
+            ", payload=" + payload +
+            '}';
+    }
+}
\ No newline at end of file
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/resources/application.properties b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/resources/application.properties
index b2b2690..5be0358 100644
--- a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/resources/application.properties
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/resources/application.properties
@@ -1,9 +1,16 @@
 spring.application.name=rocketmq-consume-demo
 
 rocketmq.name-server=localhost:9876
-
 # properties used in application code
 demo.rocketmq.topic=string-topic
+demo.rocketmq.bytesRequestTopic=bytesRequestTopic
+demo.rocketmq.stringRequestTopic=stringRequestTopic
+demo.rocketmq.objectRequestTopic=objectRequestTopic
+demo.rocketmq.genericRequestTopic=genericRequestTopic
+demo.rocketmq.bytesRequestConsumer=bytesRequestConsumer
+demo.rocketmq.stringRequestConsumer=stringRequestConsumer
+demo.rocketmq.objectRequestConsumer=objectRequestConsumer
+demo.rocketmq.genericRequestConsumer=genericRequestConsumer
 demo.rocketmq.orderTopic=order-paid-topic
 demo.rocketmq.msgExtTopic=message-ext-topic
 demo.rocketmq.transTopic=spring-transaction-topic
diff --git a/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/ProducerApplication.java b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/ProducerApplication.java
index d914bb5..46ac5ab 100644
--- a/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/ProducerApplication.java
+++ b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/ProducerApplication.java
@@ -17,6 +17,7 @@
 
 package org.apache.rocketmq.samples.springboot;
 
+import com.alibaba.fastjson.TypeReference;
 import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.List;
@@ -26,8 +27,10 @@
 import org.apache.rocketmq.client.producer.SendCallback;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.samples.springboot.domain.OrderPaidEvent;
+import org.apache.rocketmq.samples.springboot.domain.ProductWithPayload;
 import org.apache.rocketmq.samples.springboot.domain.User;
 import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
+import org.apache.rocketmq.spring.core.RocketMQLocalRequestCallback;
 import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
 import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
@@ -60,6 +63,15 @@
     private String orderPaidTopic;
     @Value("${demo.rocketmq.msgExtTopic}")
     private String msgExtTopic;
+    @Value("${demo.rocketmq.stringRequestTopic}")
+    private String stringRequestTopic;
+    @Value("${demo.rocketmq.bytesRequestTopic}")
+    private String bytesRequestTopic;
+    @Value("${demo.rocketmq.objectRequestTopic}")
+    private String objectRequestTopic;
+    @Value("${demo.rocketmq.genericRequestTopic}")
+    private String genericRequestTopic;
+
     @Resource(name = "extRocketMQTemplate")
     private RocketMQTemplate extRocketMQTemplate;
 
@@ -116,6 +128,45 @@
 
         // Send transactional messages using extRocketMQTemplate
         testExtRocketMQTemplateTransaction();
+
+        // Send request in sync mode and receive a reply of String type.
+        String replyString = rocketMQTemplate.sendAndReceive(stringRequestTopic, "request string", String.class);
+        System.out.printf("send %s and receive %s %n", "request string", replyString);
+
+        // Send request in sync mode with timeout parameter and receive a reply of byte[] type.
+        byte[] replyBytes = rocketMQTemplate.sendAndReceive(bytesRequestTopic, MessageBuilder.withPayload("request byte[]").build(), byte[].class, 3000);
+        System.out.printf("send %s and receive %s %n", "request byte[]", new String(replyBytes));
+
+        // Send request in sync mode with hashKey parameter and receive a reply of User type.
+        User requestUser = new User().setUserAge((byte) 9).setUserName("requestUserName");
+        User replyUser = rocketMQTemplate.sendAndReceive(objectRequestTopic, requestUser, User.class, "order-id");
+        System.out.printf("send %s and receive %s %n", requestUser, replyUser);
+        // Send request in sync mode with timeout and delayLevel parameter parameter and receive a reply of generic type.
+        ProductWithPayload<String> replyGenericObject = rocketMQTemplate.sendAndReceive(genericRequestTopic, "request generic",
+            new TypeReference<ProductWithPayload<String>>() {
+            }.getType(), 30000, 2);
+        System.out.printf("send %s and receive %s %n", "request generic", replyGenericObject);
+
+        // Send request in async mode and receive a reply of String type.
+        rocketMQTemplate.sendAndReceive(stringRequestTopic, "request string", new RocketMQLocalRequestCallback<String>() {
+            @Override public void onSuccess(String message) {
+                System.out.printf("send %s and receive %s %n", "request string", message);
+            }
+
+            @Override public void onException(Throwable e) {
+                e.printStackTrace();
+            }
+        });
+        // Send request in async mode and receive a reply of User type.
+        rocketMQTemplate.sendAndReceive(objectRequestTopic, new User().setUserAge((byte) 9).setUserName("requestUserName"), new RocketMQLocalRequestCallback<User>() {
+            @Override public void onSuccess(User message) {
+                System.out.printf("send user object and receive %s %n", message.toString());
+            }
+
+            @Override public void onException(Throwable e) {
+                e.printStackTrace();
+            }
+        }, 5000);
     }
 
     private void testBatchMessages() {
@@ -237,5 +288,4 @@
             return RocketMQLocalTransactionState.COMMIT;
         }
     }
-
 }
diff --git a/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/ProductWithPayload.java b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/ProductWithPayload.java
new file mode 100644
index 0000000..e241929
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/ProductWithPayload.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.samples.springboot.domain;
+
+public class ProductWithPayload<T> {
+    private String productName;
+    private T payload;
+
+    public ProductWithPayload() {
+    }
+
+    public ProductWithPayload(String productName, T payload) {
+        this.productName = productName;
+        this.payload = payload;
+    }
+
+    public String getProductName() {
+        return productName;
+    }
+
+    public void setProductName(String productName) {
+        this.productName = productName;
+    }
+
+    public T getPayload() {
+        return payload;
+    }
+
+    public void setPayload(T payload) {
+        this.payload = payload;
+    }
+
+    @Override public String toString() {
+        return "ProductWithPayload{" +
+            "productName='" + productName + '\'' +
+            ", payload=" + payload +
+            '}';
+    }
+}
\ No newline at end of file
diff --git a/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/User.java b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/User.java
index 42acc11..4f2579f 100644
--- a/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/User.java
+++ b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/User.java
@@ -18,32 +18,32 @@
 package org.apache.rocketmq.samples.springboot.domain;
 
 public class User {
-        private String userName;
-        private Byte userAge;
+    private String userName;
+    private Byte userAge;
 
-        public String getUserName() {
-            return userName;
-        }
+    public String getUserName() {
+        return userName;
+    }
 
-        public User setUserName(String userName) {
-            this.userName = userName;
-            return this;
-        }
+    public User setUserName(String userName) {
+        this.userName = userName;
+        return this;
+    }
 
-        public Byte getUserAge() {
-            return userAge;
-        }
+    public Byte getUserAge() {
+        return userAge;
+    }
 
-        public User setUserAge(Byte userAge) {
-            this.userAge = userAge;
-            return this;
-        }
+    public User setUserAge(Byte userAge) {
+        this.userAge = userAge;
+        return this;
+    }
 
-        @Override
-        public String toString() {
-            return "User{" +
-                "userName='" + userName + '\'' +
-                ", userAge=" + userAge +
-                '}';
-        }
-    }
\ No newline at end of file
+    @Override
+    public String toString() {
+        return "User{" +
+            "userName='" + userName + '\'' +
+            ", userAge=" + userAge +
+            '}';
+    }
+}
\ No newline at end of file
diff --git a/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/resources/application.properties b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/resources/application.properties
index 27a3abc..c68ac3a 100644
--- a/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/resources/application.properties
+++ b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/resources/application.properties
@@ -9,4 +9,9 @@
 demo.rocketmq.transTopic=spring-transaction-topic
 demo.rocketmq.topic.user=user-topic
 
+demo.rocketmq.bytesRequestTopic=bytesRequestTopic:tagA
+demo.rocketmq.stringRequestTopic=stringRequestTopic:tagA
+demo.rocketmq.objectRequestTopic=objectRequestTopic:tagA
+demo.rocketmq.genericRequestTopic=genericRequestTopic:tagA
+
 demo.rocketmq.extNameServer=127.0.0.1:9876
\ No newline at end of file
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQTransactionListener.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQTransactionListener.java
index dac280f..0faa57a 100644
--- a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQTransactionListener.java
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQTransactionListener.java
@@ -26,7 +26,8 @@
 
 /**
  * This annotation is used over a class which implements interface
- * org.apache.rocketmq.client.producer.TransactionListener. The class implements
+ * org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener, which will be converted to
+ * org.apache.rocketmq.client.producer.TransactionListener later. The class implements
  * two methods for process callback events after the txProducer sends a transactional message.
  * <p>Note: The annotation is used only on RocketMQ client producer side, it can not be used
  * on consumer side.
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ListenerContainerConfiguration.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ListenerContainerConfiguration.java
index 7bda36c..699474d 100644
--- a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ListenerContainerConfiguration.java
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ListenerContainerConfiguration.java
@@ -26,6 +26,7 @@
 import org.apache.rocketmq.spring.annotation.MessageModel;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.core.RocketMQReplyListener;
 import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
 import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
 import org.slf4j.Logger;
@@ -65,7 +66,7 @@
 
     @Override
     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-        this.applicationContext = (ConfigurableApplicationContext)applicationContext;
+        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
     }
 
     @Override
@@ -80,8 +81,12 @@
     private void registerContainer(String beanName, Object bean) {
         Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
 
-        if (!RocketMQListener.class.isAssignableFrom(bean.getClass())) {
-            throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName());
+        if (RocketMQListener.class.isAssignableFrom(bean.getClass()) && RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
+            throw new IllegalStateException(clazz + " cannot be both instance of " + RocketMQListener.class.getName() + " and " + RocketMQReplyListener.class.getName());
+        }
+
+        if (!RocketMQListener.class.isAssignableFrom(bean.getClass()) && !RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
+            throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName() + " or " + RocketMQReplyListener.class.getName());
         }
 
         RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);
@@ -90,7 +95,7 @@
         String topic = this.environment.resolvePlaceholders(annotation.topic());
 
         boolean listenerEnabled =
-            (boolean)rocketMQProperties.getConsumer().getListeners().getOrDefault(consumerGroup, Collections.EMPTY_MAP)
+            (boolean) rocketMQProperties.getConsumer().getListeners().getOrDefault(consumerGroup, Collections.EMPTY_MAP)
                 .getOrDefault(topic, true);
 
         if (!listenerEnabled) {
@@ -103,7 +108,7 @@
 
         String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
             counter.incrementAndGet());
-        GenericApplicationContext genericApplicationContext = (GenericApplicationContext)applicationContext;
+        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
 
         genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
             () -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
@@ -124,9 +129,9 @@
     private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String name, Object bean,
         RocketMQMessageListener annotation) {
         DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();
-        
+
         container.setRocketMQMessageListener(annotation);
-        
+
         String nameServer = environment.resolvePlaceholders(annotation.nameServer());
         nameServer = StringUtils.isEmpty(nameServer) ? rocketMQProperties.getNameServer() : nameServer;
         String accessChannel = environment.resolvePlaceholders(annotation.accessChannel());
@@ -140,7 +145,11 @@
             container.setSelectorExpression(tags);
         }
         container.setConsumerGroup(environment.resolvePlaceholders(annotation.consumerGroup()));
-        container.setRocketMQListener((RocketMQListener)bean);
+        if (RocketMQListener.class.isAssignableFrom(bean.getClass())) {
+            container.setRocketMQListener((RocketMQListener) bean);
+        } else if (RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
+            container.setRocketMQReplyListener((RocketMQReplyListener) bean);
+        }
         container.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
         container.setName(name);  // REVIEW ME, use the same clientId or multiple?
 
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQLocalRequestCallback.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQLocalRequestCallback.java
new file mode 100644
index 0000000..56b15ff
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQLocalRequestCallback.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.spring.core;
+
+/**
+ * Classes implementing this interface are used for processing callback events after receiving
+ * reply messages from consumers.
+ *
+ * @param <T> the type of message that wanted to receive from consumer
+ */
+public interface RocketMQLocalRequestCallback<T> {
+    void onSuccess(final T message);
+
+    void onException(final Throwable e);
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQReplyListener.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQReplyListener.java
new file mode 100644
index 0000000..916368d
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQReplyListener.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.spring.core;
+
+/**
+ * The consumer supporting request-reply should implement this interface.
+ *
+ * @param <T> the type of data received by the listener
+ * @param <R> the type of data replying to producer
+ */
+public interface RocketMQReplyListener<T, R> {
+    /**
+     * @param message data received by the listener
+     * @return data replying to producer
+     */
+    R onMessage(T message);
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQTemplate.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQTemplate.java
index c655696..089016a 100644
--- a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQTemplate.java
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQTemplate.java
@@ -17,6 +17,9 @@
 
 package org.apache.rocketmq.spring.core;
 
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Map;
@@ -24,19 +27,24 @@
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.MessageQueueSelector;
+import org.apache.rocketmq.client.producer.RequestCallback;
 import org.apache.rocketmq.client.producer.SendCallback;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.client.producer.TransactionMQProducer;
 import org.apache.rocketmq.client.producer.TransactionSendResult;
 import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
 import org.apache.rocketmq.spring.support.RocketMQUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.aop.framework.AopProxyUtils;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.core.AbstractMessageSendingTemplate;
 import org.springframework.messaging.core.MessagePostProcessor;
 import org.springframework.messaging.support.MessageBuilder;
@@ -52,6 +60,8 @@
 
     private MessageQueueSelector messageQueueSelector = new SelectMessageQueueByHash();
 
+    private RocketMQMessageConverter rocketMQMessageConverter = new RocketMQMessageConverter();
+
     public DefaultMQProducer getProducer() {
         return producer;
     }
@@ -77,6 +87,356 @@
     }
 
     /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param type The type of T
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Message<?> message, Type type) {
+        return sendAndReceive(destination, message, type, null, producer.getSendMsgTimeout(), 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param type The type of T
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Object payload, Type type) {
+        return sendAndReceive(destination, payload, type, null, producer.getSendMsgTimeout(), 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param type The type of T
+     * @param timeout send timeout in millis
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Message<?> message, Type type, long timeout) {
+        return sendAndReceive(destination, message, type, null, timeout, 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param type The type of T
+     * @param timeout send timeout in millis
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Object payload, Type type, long timeout) {
+        return sendAndReceive(destination, payload, type, null, timeout, 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param type The type of T
+     * @param timeout send timeout in millis
+     * @param delayLevel message delay level(0 means no delay)
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Message<?> message, Type type, long timeout, int delayLevel) {
+        return sendAndReceive(destination, message, type, null, timeout, delayLevel);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param type The type of T
+     * @param timeout send timeout in millis
+     * @param delayLevel message delay level(0 means no delay)
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Object payload, Type type, long timeout, int delayLevel) {
+        return sendAndReceive(destination, payload, type, null, timeout, delayLevel);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param type The type of T
+     * @param hashKey needed when sending message orderly
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Message<?> message, Type type, String hashKey) {
+        return sendAndReceive(destination, message, type, hashKey, producer.getSendMsgTimeout(), 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param type The type of T
+     * @param hashKey needed when sending message orderly
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Object payload, Type type, String hashKey) {
+        return sendAndReceive(destination, payload, type, hashKey, producer.getSendMsgTimeout(), 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param type The type of T
+     * @param hashKey needed when sending message orderly
+     * @param timeout send timeout in millis
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Message<?> message, Type type, String hashKey, long timeout) {
+        return sendAndReceive(destination, message, type, hashKey, timeout, 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param type The type of T
+     * @param hashKey
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Object payload, Type type, String hashKey, long timeout) {
+        return sendAndReceive(destination, payload, type, hashKey, timeout, 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param type The type that receive
+     * @param hashKey needed when sending message orderly
+     * @param timeout send timeout in millis
+     * @param delayLevel message delay level(0 means no delay)
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Message<?> message, Type type, String hashKey,
+        long timeout, int delayLevel) {
+        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
+            log.error("send request message failed. destination:{}, message is null ", destination);
+            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
+        }
+
+        try {
+            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
+            if (delayLevel > 0) {
+                rocketMsg.setDelayTimeLevel(delayLevel);
+            }
+            MessageExt replyMessage;
+
+            if (Objects.isNull(hashKey) || hashKey.isEmpty()) {
+                replyMessage = (MessageExt) producer.request(rocketMsg, timeout);
+            } else {
+                replyMessage = (MessageExt) producer.request(rocketMsg, messageQueueSelector, hashKey, timeout);
+            }
+            return replyMessage != null ? (T) doConvertMessage(replyMessage, type) : null;
+        } catch (Exception e) {
+            log.error("send request message failed. destination:{}, message:{} ", destination, message);
+            throw new MessagingException(e.getMessage(), e);
+        }
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param type The type that receive
+     * @param hashKey needed when sending message orderly
+     * @param timeout send timeout in millis
+     * @param delayLevel message delay level(0 means no delay)
+     * @return
+     */
+    public <T> T sendAndReceive(String destination, Object payload, Type type, String hashKey,
+        long timeout, int delayLevel) {
+        Message<?> message = MessageBuilder.withPayload(payload).build();
+        return sendAndReceive(destination, message, type, hashKey, timeout, delayLevel);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @return
+     */
+    public void sendAndReceive(String destination, Message<?> message,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback) {
+        sendAndReceive(destination, message, rocketMQLocalRequestCallback, null, producer.getSendMsgTimeout(), 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @return
+     */
+    public void sendAndReceive(String destination, Object payload,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback) {
+        sendAndReceive(destination, payload, rocketMQLocalRequestCallback, null, producer.getSendMsgTimeout(), 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @param timeout send timeout in millis
+     * @return
+     */
+    public void sendAndReceive(String destination, Message<?> message,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout) {
+        sendAndReceive(destination, message, rocketMQLocalRequestCallback, null, timeout, 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @param timeout send timeout in millis
+     * @return
+     */
+    public void sendAndReceive(String destination, Object payload,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout) {
+        sendAndReceive(destination, payload, rocketMQLocalRequestCallback, null, timeout, 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @param timeout send timeout in millis
+     * @param delayLevel message delay level(0 means no delay)
+     * @return
+     */
+    public void sendAndReceive(String destination, Message<?> message,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout, int delayLevel) {
+        sendAndReceive(destination, message, rocketMQLocalRequestCallback, null, timeout, delayLevel);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @param hashKey needed when sending message orderly
+     * @return
+     */
+    public void sendAndReceive(String destination, Object payload,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey) {
+        sendAndReceive(destination, payload, rocketMQLocalRequestCallback, hashKey, producer.getSendMsgTimeout(), 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @param hashKey needed when sending message orderly
+     * @param timeout send timeout in millis
+     * @return
+     */
+    public void sendAndReceive(String destination, Message<?> message,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout) {
+        sendAndReceive(destination, message, rocketMQLocalRequestCallback, hashKey, timeout, 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @param hashKey needed when sending message orderly
+     * @param timeout send timeout in millis
+     * @return
+     */
+    public void sendAndReceive(String destination, Object payload,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout) {
+        sendAndReceive(destination, payload, rocketMQLocalRequestCallback, hashKey, timeout, 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @param hashKey needed when sending message orderly
+     * @return
+     */
+    public void sendAndReceive(String destination, Message<?> message,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey) {
+        sendAndReceive(destination, message, rocketMQLocalRequestCallback, hashKey, producer.getSendMsgTimeout(), 0);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @param timeout send timeout in millis
+     * @param delayLevel message delay level(0 means no delay)
+     * @return
+     */
+    public void sendAndReceive(String destination, Object payload,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout, int delayLevel) {
+        sendAndReceive(destination, payload, rocketMQLocalRequestCallback, null, timeout, delayLevel);
+    }
+
+    /**
+     * @param destination formats: `topicName:tags`
+     * @param payload the payload to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @param hashKey needed when sending message orderly
+     * @param timeout send timeout in millis
+     * @param delayLevel message delay level(0 means no delay)
+     * @return
+     */
+    public void sendAndReceive(String destination, Object payload,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout, int delayLevel) {
+        Message<?> message = MessageBuilder.withPayload(payload).build();
+        sendAndReceive(destination, message, rocketMQLocalRequestCallback, hashKey, timeout, delayLevel);
+    }
+
+    /**
+     * Send request message in asynchronous mode. </p> This method returns immediately. On receiving reply message,
+     * <code>rocketMQLocalRequestCallback</code> will be executed. </p>
+     *
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message} the message to be sent.
+     * @param rocketMQLocalRequestCallback callback that will invoked when reply message received.
+     * @param hashKey needed when sending message orderly
+     * @param timeout send timeout in millis
+     * @param delayLevel message delay level(0 means no delay)
+     * @return
+     */
+    public void sendAndReceive(String destination, Message<?> message,
+        RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout, int delayLevel) {
+        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
+            log.error("send request message failed. destination:{}, message is null ", destination);
+            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
+        }
+
+        try {
+            org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
+            if (delayLevel > 0) {
+                rocketMsg.setDelayTimeLevel(delayLevel);
+            }
+            if (timeout <= 0) {
+                timeout = producer.getSendMsgTimeout();
+            }
+            RequestCallback requestCallback = null;
+            if (rocketMQLocalRequestCallback != null) {
+                requestCallback = new RequestCallback() {
+                    @Override public void onSuccess(org.apache.rocketmq.common.message.Message message) {
+                        rocketMQLocalRequestCallback.onSuccess(doConvertMessage((MessageExt) message, getMessageType(rocketMQLocalRequestCallback)));
+                    }
+
+                    @Override public void onException(Throwable e) {
+                        rocketMQLocalRequestCallback.onException(e);
+                    }
+                };
+            }
+            if (Objects.isNull(hashKey) || hashKey.isEmpty()) {
+                producer.request(rocketMsg, requestCallback, timeout);
+            } else {
+                producer.request(rocketMsg, messageQueueSelector, hashKey, requestCallback, timeout);
+            }
+        } catch (
+            Exception e) {
+            log.error("send request message failed. destination:{}, message:{} ", destination, message);
+            throw new MessagingException(e.getMessage(), e);
+        }
+
+    }
+
+    /**
      * <p> Send message in synchronous mode. This method returns only when the sending procedure totally completes.
      * Reliable synchronous transmission is used in extensive scenes, such as important notification messages, SMS
      * notification, SMS marketing system, etc.. </p>
@@ -87,7 +447,7 @@
      * duplication issue.
      *
      * @param destination formats: `topicName:tags`
-     * @param message     {@link org.springframework.messaging.Message}
+     * @param message {@link org.springframework.messaging.Message}
      * @return {@link SendResult}
      */
     public SendResult syncSend(String destination, Message<?> message) {
@@ -98,8 +458,8 @@
      * Same to {@link #syncSend(String, Message)} with send timeout specified in addition.
      *
      * @param destination formats: `topicName:tags`
-     * @param message     {@link org.springframework.messaging.Message}
-     * @param timeout     send timeout with millis
+     * @param message {@link org.springframework.messaging.Message}
+     * @param timeout send timeout with millis
      * @return {@link SendResult}
      */
     public SendResult syncSend(String destination, Message<?> message, long timeout) {
@@ -110,8 +470,8 @@
      * syncSend batch messages in a given timeout.
      *
      * @param destination formats: `topicName:tags`
-     * @param messages    Collection of {@link org.springframework.messaging.Message}
-     * @param timeout     send timeout with millis
+     * @param messages Collection of {@link org.springframework.messaging.Message}
+     * @param timeout send timeout with millis
      * @return {@link SendResult}
      */
     public <T extends Message> SendResult syncSend(String destination, Collection<T> messages, long timeout) {
@@ -147,9 +507,9 @@
      * Same to {@link #syncSend(String, Message)} with send timeout specified in addition.
      *
      * @param destination formats: `topicName:tags`
-     * @param message     {@link org.springframework.messaging.Message}
-     * @param timeout     send timeout with millis
-     * @param delayLevel  level for the delay message
+     * @param message {@link org.springframework.messaging.Message}
+     * @param timeout send timeout with millis
+     * @param delayLevel level for the delay message
      * @return {@link SendResult}
      */
     public SendResult syncSend(String destination, Message<?> message, long timeout, int delayLevel) {
@@ -179,7 +539,7 @@
      * Same to {@link #syncSend(String, Message)}.
      *
      * @param destination formats: `topicName:tags`
-     * @param payload     the Object to use as payload
+     * @param payload the Object to use as payload
      * @return {@link SendResult}
      */
     public SendResult syncSend(String destination, Object payload) {
@@ -190,8 +550,8 @@
      * Same to {@link #syncSend(String, Object)} with send timeout specified in addition.
      *
      * @param destination formats: `topicName:tags`
-     * @param payload     the Object to use as payload
-     * @param timeout     send timeout with millis
+     * @param payload the Object to use as payload
+     * @param timeout send timeout with millis
      * @return {@link SendResult}
      */
     public SendResult syncSend(String destination, Object payload, long timeout) {
@@ -203,8 +563,8 @@
      * Same to {@link #syncSend(String, Message)} with send orderly with hashKey by specified.
      *
      * @param destination formats: `topicName:tags`
-     * @param message     {@link org.springframework.messaging.Message}
-     * @param hashKey     use this key to select queue. for example: orderId, productId ...
+     * @param message {@link org.springframework.messaging.Message}
+     * @param hashKey use this key to select queue. for example: orderId, productId ...
      * @return {@link SendResult}
      */
     public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey) {
@@ -215,9 +575,9 @@
      * Same to {@link #syncSendOrderly(String, Message, String)} with send timeout specified in addition.
      *
      * @param destination formats: `topicName:tags`
-     * @param message     {@link org.springframework.messaging.Message}
-     * @param hashKey     use this key to select queue. for example: orderId, productId ...
-     * @param timeout     send timeout with millis
+     * @param message {@link org.springframework.messaging.Message}
+     * @param hashKey use this key to select queue. for example: orderId, productId ...
+     * @param timeout send timeout with millis
      * @return {@link SendResult}
      */
     public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey, long timeout) {
@@ -244,8 +604,8 @@
      * Same to {@link #syncSend(String, Object)} with send orderly with hashKey by specified.
      *
      * @param destination formats: `topicName:tags`
-     * @param payload     the Object to use as payload
-     * @param hashKey     use this key to select queue. for example: orderId, productId ...
+     * @param payload the Object to use as payload
+     * @param hashKey use this key to select queue. for example: orderId, productId ...
      * @return {@link SendResult}
      */
     public SendResult syncSendOrderly(String destination, Object payload, String hashKey) {
@@ -256,9 +616,9 @@
      * Same to {@link #syncSendOrderly(String, Object, String)} with send timeout specified in addition.
      *
      * @param destination formats: `topicName:tags`
-     * @param payload     the Object to use as payload
-     * @param hashKey     use this key to select queue. for example: orderId, productId ...
-     * @param timeout     send timeout with millis
+     * @param payload the Object to use as payload
+     * @param hashKey use this key to select queue. for example: orderId, productId ...
+     * @param timeout send timeout with millis
      * @return {@link SendResult}
      */
     public SendResult syncSendOrderly(String destination, Object payload, String hashKey, long timeout) {
@@ -270,11 +630,11 @@
      * Same to {@link #asyncSend(String, Message, SendCallback)} with send timeout and delay level specified in
      * addition.
      *
-     * @param destination  formats: `topicName:tags`
-     * @param message      {@link org.springframework.messaging.Message}
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message}
      * @param sendCallback {@link SendCallback}
-     * @param timeout      send timeout with millis
-     * @param delayLevel   level for the delay message
+     * @param timeout send timeout with millis
+     * @param delayLevel level for the delay message
      */
     public void asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout,
         int delayLevel) {
@@ -297,10 +657,10 @@
     /**
      * Same to {@link #asyncSend(String, Message, SendCallback)} with send timeout specified in addition.
      *
-     * @param destination  formats: `topicName:tags`
-     * @param message      {@link org.springframework.messaging.Message}
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message}
      * @param sendCallback {@link SendCallback}
-     * @param timeout      send timeout with millis
+     * @param timeout send timeout with millis
      */
     public void asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout) {
         asyncSend(destination, message, sendCallback, timeout, 0);
@@ -316,8 +676,8 @@
      * DefaultMQProducer#getRetryTimesWhenSendAsyncFailed} times before claiming sending failure, which may yield
      * message duplication and application developers are the one to resolve this potential issue.
      *
-     * @param destination  formats: `topicName:tags`
-     * @param message      {@link org.springframework.messaging.Message}
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message}
      * @param sendCallback {@link SendCallback}
      */
     public void asyncSend(String destination, Message<?> message, SendCallback sendCallback) {
@@ -327,10 +687,10 @@
     /**
      * Same to {@link #asyncSend(String, Object, SendCallback)} with send timeout specified in addition.
      *
-     * @param destination  formats: `topicName:tags`
-     * @param payload      the Object to use as payload
+     * @param destination formats: `topicName:tags`
+     * @param payload the Object to use as payload
      * @param sendCallback {@link SendCallback}
-     * @param timeout      send timeout with millis
+     * @param timeout send timeout with millis
      */
     public void asyncSend(String destination, Object payload, SendCallback sendCallback, long timeout) {
         Message<?> message = MessageBuilder.withPayload(payload).build();
@@ -340,8 +700,8 @@
     /**
      * Same to {@link #asyncSend(String, Message, SendCallback)}.
      *
-     * @param destination  formats: `topicName:tags`
-     * @param payload      the Object to use as payload
+     * @param destination formats: `topicName:tags`
+     * @param payload the Object to use as payload
      * @param sendCallback {@link SendCallback}
      */
     public void asyncSend(String destination, Object payload, SendCallback sendCallback) {
@@ -352,11 +712,11 @@
      * Same to {@link #asyncSendOrderly(String, Message, String, SendCallback)} with send timeout specified in
      * addition.
      *
-     * @param destination  formats: `topicName:tags`
-     * @param message      {@link org.springframework.messaging.Message}
-     * @param hashKey      use this key to select queue. for example: orderId, productId ...
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message}
+     * @param hashKey use this key to select queue. for example: orderId, productId ...
      * @param sendCallback {@link SendCallback}
-     * @param timeout      send timeout with millis
+     * @param timeout send timeout with millis
      */
     public void asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback,
         long timeout) {
@@ -376,9 +736,9 @@
     /**
      * Same to {@link #asyncSend(String, Message, SendCallback)} with send orderly with hashKey by specified.
      *
-     * @param destination  formats: `topicName:tags`
-     * @param message      {@link org.springframework.messaging.Message}
-     * @param hashKey      use this key to select queue. for example: orderId, productId ...
+     * @param destination formats: `topicName:tags`
+     * @param message {@link org.springframework.messaging.Message}
+     * @param hashKey use this key to select queue. for example: orderId, productId ...
      * @param sendCallback {@link SendCallback}
      */
     public void asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback) {
@@ -388,9 +748,9 @@
     /**
      * Same to {@link #asyncSendOrderly(String, Message, String, SendCallback)}.
      *
-     * @param destination  formats: `topicName:tags`
-     * @param payload      the Object to use as payload
-     * @param hashKey      use this key to select queue. for example: orderId, productId ...
+     * @param destination formats: `topicName:tags`
+     * @param payload the Object to use as payload
+     * @param hashKey use this key to select queue. for example: orderId, productId ...
      * @param sendCallback {@link SendCallback}
      */
     public void asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback) {
@@ -400,11 +760,11 @@
     /**
      * Same to {@link #asyncSendOrderly(String, Object, String, SendCallback)} with send timeout specified in addition.
      *
-     * @param destination  formats: `topicName:tags`
-     * @param payload      the Object to use as payload
-     * @param hashKey      use this key to select queue. for example: orderId, productId ...
+     * @param destination formats: `topicName:tags`
+     * @param payload the Object to use as payload
+     * @param hashKey use this key to select queue. for example: orderId, productId ...
      * @param sendCallback {@link SendCallback}
-     * @param timeout      send timeout with millis
+     * @param timeout send timeout with millis
      */
     public void asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback,
         long timeout) {
@@ -419,7 +779,7 @@
      * One-way transmission is used for cases requiring moderate reliability, such as log collection.
      *
      * @param destination formats: `topicName:tags`
-     * @param message     {@link org.springframework.messaging.Message}
+     * @param message {@link org.springframework.messaging.Message}
      */
     public void sendOneWay(String destination, Message<?> message) {
         if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
@@ -439,7 +799,7 @@
      * Same to {@link #sendOneWay(String, Message)}
      *
      * @param destination formats: `topicName:tags`
-     * @param payload     the Object to use as payload
+     * @param payload the Object to use as payload
      */
     public void sendOneWay(String destination, Object payload) {
         Message<?> message = MessageBuilder.withPayload(payload).build();
@@ -450,8 +810,8 @@
      * Same to {@link #sendOneWay(String, Message)} with send orderly with hashKey by specified.
      *
      * @param destination formats: `topicName:tags`
-     * @param message     {@link org.springframework.messaging.Message}
-     * @param hashKey     use this key to select queue. for example: orderId, productId ...
+     * @param message {@link org.springframework.messaging.Message}
+     * @param hashKey use this key to select queue. for example: orderId, productId ...
      */
     public void sendOneWayOrderly(String destination, Message<?> message, String hashKey) {
         if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
@@ -471,7 +831,7 @@
      * Same to {@link #sendOneWayOrderly(String, Message, String)}
      *
      * @param destination formats: `topicName:tags`
-     * @param payload     the Object to use as payload
+     * @param payload the Object to use as payload
      */
     public void sendOneWayOrderly(String destination, Object payload, String hashKey) {
         Message<?> message = MessageBuilder.withPayload(payload).build();
@@ -511,9 +871,9 @@
     /**
      * Send Spring Message in Transaction
      *
-     * @param destination     destination formats: `topicName:tags`
-     * @param message         message {@link org.springframework.messaging.Message}
-     * @param arg             ext arg
+     * @param destination destination formats: `topicName:tags`
+     * @param message message {@link org.springframework.messaging.Message}
+     * @param arg ext arg
      * @return TransactionSendResult
      * @throws MessagingException
      */
@@ -537,4 +897,57 @@
             destination, msg);
     }
 
+    private Object doConvertMessage(MessageExt messageExt, Type type) {
+        if (Objects.equals(type, MessageExt.class)) {
+            return messageExt;
+        } else if (Objects.equals(type, byte[].class)) {
+            return messageExt.getBody();
+        } else {
+            String str = new String(messageExt.getBody(), Charset.forName(charset));
+            if (Objects.equals(type, String.class)) {
+                return str;
+            } else {
+                // If msgType not string, use objectMapper change it.
+                try {
+                    if (type instanceof Class) {
+                        //if the messageType has not Generic Parameter
+                        return this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) type);
+                    } else {
+                        //if the messageType has Generic Parameter, then use SmartMessageConverter#fromMessage with third parameter "conversionHint".
+                        //we have validate the MessageConverter is SmartMessageConverter in this#getMethodParameter.
+                        return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) type).getRawType(), null);
+                    }
+                } catch (Exception e) {
+                    log.error("convert failed. str:{}, msgType:{}", str, type);
+                    throw new RuntimeException("cannot convert message to " + type, e);
+                }
+            }
+        }
+    }
+
+    private Type getMessageType(RocketMQLocalRequestCallback rocketMQLocalRequestCallback) {
+        Class<?> targetClass = AopProxyUtils.ultimateTargetClass(rocketMQLocalRequestCallback);
+        Type matchedGenericInterface = null;
+        while (Objects.nonNull(targetClass)) {
+            Type[] interfaces = targetClass.getGenericInterfaces();
+            if (Objects.nonNull(interfaces)) {
+                for (Type type : interfaces) {
+                    if (type instanceof ParameterizedType && (Objects.equals(((ParameterizedType) type).getRawType(), RocketMQLocalRequestCallback.class))) {
+                        matchedGenericInterface = type;
+                        break;
+                    }
+                }
+            }
+            targetClass = targetClass.getSuperclass();
+        }
+        if (Objects.isNull(matchedGenericInterface)) {
+            return Object.class;
+        }
+
+        Type[] actualTypeArguments = ((ParameterizedType) matchedGenericInterface).getActualTypeArguments();
+        if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
+            return actualTypeArguments[0];
+        }
+        return Object.class;
+    }
 }
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
index 25ec320..2642d07 100644
--- a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
@@ -34,14 +34,20 @@
 import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
 import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
 import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.client.utils.MessageUtil;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.apache.rocketmq.spring.annotation.ConsumeMode;
 import org.apache.rocketmq.spring.annotation.MessageModel;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.annotation.SelectorType;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
+import org.apache.rocketmq.spring.core.RocketMQReplyListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.aop.framework.AopProxyUtils;
@@ -51,10 +57,14 @@
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.core.MethodParameter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
+import org.springframework.util.MimeTypeUtils;
 
 @SuppressWarnings("WeakerAccess")
 public class DefaultRocketMQListenerContainer implements InitializingBean,
@@ -92,6 +102,8 @@
 
     private RocketMQListener rocketMQListener;
 
+    private RocketMQReplyListener rocketMQReplyListener;
+
     private RocketMQMessageListener rocketMQMessageListener;
 
     private DefaultMQPushConsumer consumer;
@@ -186,6 +198,14 @@
         this.rocketMQListener = rocketMQListener;
     }
 
+    public RocketMQReplyListener getRocketMQReplyListener() {
+        return rocketMQReplyListener;
+    }
+
+    public void setRocketMQReplyListener(RocketMQReplyListener rocketMQReplyListener) {
+        this.rocketMQReplyListener = rocketMQReplyListener;
+    }
+
     public RocketMQMessageListener getRocketMQMessageListener() {
         return rocketMQMessageListener;
     }
@@ -209,14 +229,14 @@
         return selectorType;
     }
 
-    public String getSelectorExpression() {
-        return selectorExpression;
-    }
-
     public void setSelectorExpression(String selectorExpression) {
         this.selectorExpression = selectorExpression;
     }
 
+    public String getSelectorExpression() {
+        return selectorExpression;
+    }
+
     public MessageModel getMessageModel() {
         return messageModel;
     }
@@ -230,11 +250,6 @@
     }
 
     @Override
-    public void setupMessageListener(RocketMQListener rocketMQListener) {
-        this.rocketMQListener = rocketMQListener;
-    }
-
-    @Override
     public void destroy() {
         this.setRunning(false);
         if (Objects.nonNull(consumer)) {
@@ -327,6 +342,119 @@
         this.name = name;
     }
 
+    public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {
+
+        @SuppressWarnings("unchecked")
+        @Override
+        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
+            for (MessageExt messageExt : msgs) {
+                log.debug("received msg: {}", messageExt);
+                try {
+                    long now = System.currentTimeMillis();
+                    handleMessage(messageExt);
+                    long costTime = System.currentTimeMillis() - now;
+                    log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
+                } catch (Exception e) {
+                    log.warn("consume message failed. messageExt:{}, error:{}", messageExt, e);
+                    context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
+                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
+                }
+            }
+
+            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+        }
+    }
+
+    public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
+
+        @SuppressWarnings("unchecked")
+        @Override
+        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
+            for (MessageExt messageExt : msgs) {
+                log.debug("received msg: {}", messageExt);
+                try {
+                    long now = System.currentTimeMillis();
+                    handleMessage(messageExt);
+                    long costTime = System.currentTimeMillis() - now;
+                    log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
+                } catch (Exception e) {
+                    log.warn("consume message failed. messageExt:{}", messageExt, e);
+                    context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis);
+                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
+                }
+            }
+
+            return ConsumeOrderlyStatus.SUCCESS;
+        }
+    }
+
+    private void handleMessage(
+        MessageExt messageExt) throws MQClientException, RemotingException, InterruptedException {
+        if (rocketMQListener != null) {
+            rocketMQListener.onMessage(doConvertMessage(messageExt));
+        } else if (rocketMQReplyListener != null) {
+            Object replyContent = rocketMQReplyListener.onMessage(doConvertMessage(messageExt));
+            Message<?> message = MessageBuilder.withPayload(replyContent).build();
+
+            org.apache.rocketmq.common.message.Message replyMessage = MessageUtil.createReplyMessage(messageExt, convertToBytes(message));
+            consumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer().send(replyMessage, new SendCallback() {
+                @Override public void onSuccess(SendResult sendResult) {
+                    if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
+                        log.error("Consumer replies message failed. SendStatus: {}", sendResult.getSendStatus());
+                    } else {
+                        log.info("Consumer replies message success.");
+                    }
+                }
+
+                @Override public void onException(Throwable e) {
+                    log.error("Consumer replies message failed. error: {}", e.getLocalizedMessage());
+                }
+            });
+        }
+    }
+
+    private byte[] convertToBytes(Message<?> message) {
+        Message<?> messageWithSerializedPayload = doConvert(message.getPayload(), message.getHeaders());
+        Object payloadObj = messageWithSerializedPayload.getPayload();
+        byte[] payloads;
+        try {
+            if (null == payloadObj) {
+                throw new RuntimeException("the message cannot be empty");
+            }
+            if (payloadObj instanceof String) {
+                payloads = ((String) payloadObj).getBytes(Charset.forName(charset));
+            } else if (payloadObj instanceof byte[]) {
+                payloads = (byte[]) messageWithSerializedPayload.getPayload();
+            } else {
+                String jsonObj = (String) this.messageConverter.fromMessage(messageWithSerializedPayload, payloadObj.getClass());
+                if (null == jsonObj) {
+                    throw new RuntimeException(String.format(
+                        "empty after conversion [messageConverter:%s,payloadClass:%s,payloadObj:%s]",
+                        this.messageConverter.getClass(), payloadObj.getClass(), payloadObj));
+                }
+                payloads = jsonObj.getBytes(Charset.forName(charset));
+            }
+        } catch (Exception e) {
+            throw new RuntimeException("convert to bytes failed.", e);
+        }
+        return payloads;
+    }
+
+    private Message<?> doConvert(Object payload, MessageHeaders headers) {
+        Message<?> message = this.messageConverter instanceof SmartMessageConverter ?
+            ((SmartMessageConverter) this.messageConverter).toMessage(payload, headers, null) :
+            this.messageConverter.toMessage(payload, headers);
+        if (message == null) {
+            String payloadType = payload.getClass().getName();
+            Object contentType = headers != null ? headers.get(MessageHeaders.CONTENT_TYPE) : null;
+            throw new MessageConversionException("Unable to convert payload with type='" + payloadType +
+                "', contentType='" + contentType + "', converter=[" + this.messageConverter + "]");
+        }
+        MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
+        builder.setHeaderIfAbsent(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN);
+        return builder.build();
+    }
+
     @SuppressWarnings("unchecked")
     private Object doConvertMessage(MessageExt messageExt) {
         if (Objects.equals(messageType, MessageExt.class)) {
@@ -355,7 +483,12 @@
     }
 
     private MethodParameter getMethodParameter() {
-        Class<?> targetClass = AopProxyUtils.ultimateTargetClass(rocketMQListener);
+        Class<?> targetClass;
+        if (rocketMQListener != null) {
+            targetClass = AopProxyUtils.ultimateTargetClass(rocketMQListener);
+        } else {
+            targetClass = AopProxyUtils.ultimateTargetClass(rocketMQReplyListener);
+        }
         Type messageType = this.getMessageType();
         Class clazz = null;
         if (messageType instanceof ParameterizedType && messageConverter instanceof SmartMessageConverter) {
@@ -375,14 +508,19 @@
     }
 
     private Type getMessageType() {
-        Class<?> targetClass = AopProxyUtils.ultimateTargetClass(rocketMQListener);
+        Class<?> targetClass;
+        if (rocketMQListener != null) {
+            targetClass = AopProxyUtils.ultimateTargetClass(rocketMQListener);
+        } else {
+            targetClass = AopProxyUtils.ultimateTargetClass(rocketMQReplyListener);
+        }
         Type matchedGenericInterface = null;
         while (Objects.nonNull(targetClass)) {
             Type[] interfaces = targetClass.getGenericInterfaces();
             if (Objects.nonNull(interfaces)) {
                 for (Type type : interfaces) {
-                    if (type instanceof ParameterizedType
-                            && Objects.equals(((ParameterizedType) type).getRawType(), RocketMQListener.class)) {
+                    if (type instanceof ParameterizedType &&
+                        (Objects.equals(((ParameterizedType) type).getRawType(), RocketMQListener.class) || Objects.equals(((ParameterizedType) type).getRawType(), RocketMQReplyListener.class))) {
                         matchedGenericInterface = type;
                         break;
                     }
@@ -401,10 +539,10 @@
         return Object.class;
     }
 
-
-
     private void initRocketMQPushConsumer() throws MQClientException {
-        Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required");
+        if (rocketMQListener == null && rocketMQReplyListener == null) {
+            throw new IllegalArgumentException("Property 'rocketMQListener' or 'rocketMQReplyListener' is required");
+        }
         Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
         Assert.notNull(nameServer, "Property 'nameServer' is required");
         Assert.notNull(topic, "Property 'topic' is required");
@@ -475,54 +613,10 @@
 
         if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) {
             ((RocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(consumer);
+        } else if (rocketMQReplyListener instanceof RocketMQPushConsumerLifecycleListener) {
+            ((RocketMQPushConsumerLifecycleListener) rocketMQReplyListener).prepareStart(consumer);
         }
 
     }
 
-    public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {
-
-        @SuppressWarnings("unchecked")
-        @Override
-        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
-            for (MessageExt messageExt : msgs) {
-                log.debug("received msg: {}", messageExt);
-                try {
-                    long now = System.currentTimeMillis();
-                    rocketMQListener.onMessage(doConvertMessage(messageExt));
-                    long costTime = System.currentTimeMillis() - now;
-                    log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
-                } catch (Exception e) {
-                    log.warn("consume message failed. messageExt:{}", messageExt, e);
-                    context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
-                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
-                }
-            }
-
-            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-        }
-    }
-
-    public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
-
-        @SuppressWarnings("unchecked")
-        @Override
-        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
-            for (MessageExt messageExt : msgs) {
-                log.debug("received msg: {}", messageExt);
-                try {
-                    long now = System.currentTimeMillis();
-                    rocketMQListener.onMessage(doConvertMessage(messageExt));
-                    long costTime = System.currentTimeMillis() - now;
-                    log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
-                } catch (Exception e) {
-                    log.warn("consume message failed. messageExt:{}", messageExt, e);
-                    context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis);
-                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
-                }
-            }
-
-            return ConsumeOrderlyStatus.SUCCESS;
-        }
-    }
-
 }
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQListenerContainer.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQListenerContainer.java
index ee52de8..d9693bc 100644
--- a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQListenerContainer.java
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQListenerContainer.java
@@ -17,14 +17,8 @@
 
 package org.apache.rocketmq.spring.support;
 
-import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.beans.factory.DisposableBean;
 
 public interface RocketMQListenerContainer extends DisposableBean {
 
-    /**
-     * Setup the message listener to use. Throws an {@link IllegalArgumentException} if that message listener type is
-     * not supported.
-     */
-    void setupMessageListener(RocketMQListener<?> messageListener);
 }
diff --git a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/autoconfigure/RocketMQAutoConfigurationTest.java b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/autoconfigure/RocketMQAutoConfigurationTest.java
index 553183d..18a32f8 100644
--- a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/autoconfigure/RocketMQAutoConfigurationTest.java
+++ b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/autoconfigure/RocketMQAutoConfigurationTest.java
@@ -27,6 +27,7 @@
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
 import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
+import org.apache.rocketmq.spring.core.RocketMQReplyListener;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
 import org.junit.Assert;
@@ -183,6 +184,24 @@
             });
     }
 
+    @Test
+    public void testRocketMQListenerContainer_RocketMQReplyListener() {
+        runner.withPropertyValues("rocketmq.name-server=127.0.0.1:9876").
+            withUserConfiguration(TestConfigWithRocketMQReplyListener.class).
+            run((context) -> {
+                assertThat(context).getFailure().hasMessageContaining("connect to [127.0.0.1:9876] failed");
+            });
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void testRocketMQListenerContainer_WrongRocketMQListenerType() {
+        runner.withPropertyValues("rocketmq.name-server=127.0.0.1:9876").
+            withUserConfiguration(TestConfigWithWrongRocketMQListener.class).
+            run((context) -> {
+                context.getBean(RocketMQMessageConverter.class);
+            });
+    }
+
     @Configuration
     static class TestConfig {
 
@@ -199,6 +218,30 @@
     }
 
     @Configuration
+    static class TestConfigWithRocketMQReplyListener {
+
+        @Bean
+        public Object consumeListener() {
+            return new TestDefaultNameServerRocketMQReplyListener();
+        }
+
+        @Bean
+        public Object consumeListener1() {
+            return new TestCustomNameServerRocketMQReplyListener();
+        }
+
+    }
+
+    @Configuration
+    static class TestConfigWithWrongRocketMQListener {
+
+        @Bean
+        public Object consumeListener() {
+            return new WrongRocketMQListener();
+        }
+    }
+
+    @Configuration
     static class CustomObjectMapperConfig {
 
         @Bean
@@ -236,6 +279,32 @@
         }
     }
 
+    @RocketMQMessageListener(consumerGroup = "abcd", topic = "test")
+    static class TestDefaultNameServerRocketMQReplyListener implements RocketMQReplyListener<String, String> {
+
+        @Override
+        public String onMessage(String message) {
+            return "test";
+        }
+    }
+
+    @RocketMQMessageListener(consumerGroup = "abcde", topic = "test")
+    static class WrongRocketMQListener {
+
+        public String onMessage(String message) {
+            return "test";
+        }
+    }
+
+    @RocketMQMessageListener(nameServer = "127.0.1.1:9876", consumerGroup = "abcd1", topic = "test")
+    static class TestCustomNameServerRocketMQReplyListener implements RocketMQReplyListener<String, String> {
+
+        @Override
+        public String onMessage(String message) {
+            return "test";
+        }
+    }
+
     @Configuration
     static class TestTransactionListenerConfig {
         @Bean
diff --git a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/core/RocketMQTemplateTest.java b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/core/RocketMQTemplateTest.java
index a6cc91d..da6d777 100644
--- a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/core/RocketMQTemplateTest.java
+++ b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/core/RocketMQTemplateTest.java
@@ -21,6 +21,7 @@
 import org.apache.rocketmq.client.producer.SendCallback;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.client.producer.TransactionMQProducer;
+import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
 import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
 import org.junit.Test;
@@ -28,10 +29,13 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
 
 @RunWith(SpringJUnit4ClassRunner.class)
 @SpringBootTest(properties = {
@@ -48,6 +52,12 @@
     @Value("${test.rocketmq.topic}")
     String topic;
 
+    @Value("stringRequestTopic:tagA")
+    String stringRequestTopic;
+
+    @Value("objectRequestTopic:tagA")
+    String objectRequestTopic;
+
     @Test
     public void testSendMessage() {
         try {
@@ -78,6 +88,104 @@
     }
 
     @Test
+    public void testSendAndReceive_NullMessage() {
+        try {
+            String response = rocketMQTemplate.sendAndReceive(stringRequestTopic, new Message<String>() {
+                @Override public String getPayload() {
+                    return null;
+                }
+
+                @Override public MessageHeaders getHeaders() {
+                    return null;
+                }
+            }, String.class);
+        } catch (IllegalArgumentException e) {
+            assertThat(e).hasMessageContaining("`message` and `message.payload` cannot be null");
+        }
+
+        try {
+            String response = rocketMQTemplate.sendAndReceive(stringRequestTopic, (Object) null, String.class);
+        } catch (IllegalArgumentException e) {
+            assertThat(e).hasMessageContaining("Payload must not be null");
+        }
+    }
+
+    @Test
+    public void testSendAndReceive_Sync() throws InterruptedException {
+        try {
+            String responseMessage = rocketMQTemplate.sendAndReceive(stringRequestTopic, MessageBuilder.withPayload("requestTopicSync").build(), String.class);
+            assertThat(responseMessage).isNotNull();
+        } catch (MessagingException e) {
+            assertThat(e).hasMessageContaining("org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [127.0.0.1:9876] failed");
+        }
+
+        try {
+            String responseMessage = rocketMQTemplate.sendAndReceive(stringRequestTopic, "requestTopicSync", String.class, "orderId");
+            assertThat(responseMessage).isNotNull();
+        } catch (MessagingException e) {
+            assertThat(e).hasMessageContaining("org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [127.0.0.1:9876] failed");
+        }
+    }
+
+    @Test
+    public void testSendAndReceive_Async() {
+        try {
+            rocketMQTemplate.sendAndReceive(stringRequestTopic, MessageBuilder.withPayload("requestTopicASync").build(), new RocketMQLocalRequestCallback<String>() {
+                @Override public void onSuccess(String message) {
+                    System.out.printf("receive string: %s %n", message);
+                }
+
+                @Override public void onException(Throwable e) {
+                    e.printStackTrace();
+                }
+            });
+        } catch (MessagingException e) {
+            assertThat(e).hasMessageContaining("org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [127.0.0.1:9876] failed");
+        }
+
+        try {
+            rocketMQTemplate.sendAndReceive(stringRequestTopic, "requestTopicAsyncWithHasKey", new RocketMQLocalRequestCallback<String>() {
+                @Override public void onSuccess(String message) {
+                    System.out.printf("receive string: %s %n", message);
+                }
+
+                @Override public void onException(Throwable e) {
+                    e.printStackTrace();
+                }
+            }, "order-id");
+        } catch (MessagingException e) {
+            assertThat(e).hasMessageContaining("org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [127.0.0.1:9876] failed");
+        }
+
+        try {
+            rocketMQTemplate.sendAndReceive(stringRequestTopic, "requestTopicAsyncWithTimeout", new RocketMQLocalRequestCallback<String>() {
+                @Override public void onSuccess(String message) {
+                    System.out.printf("receive string: %s %n", message);
+                }
+
+                @Override public void onException(Throwable e) {
+                    e.printStackTrace();
+                }
+            }, "order-id", 5000);
+        } catch (MessagingException e) {
+            assertThat(e).hasMessageContaining("org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [127.0.0.1:9876] failed");
+        }
+        try {
+            rocketMQTemplate.sendAndReceive(objectRequestTopic, "requestTopicAsyncWithTimeout", new RocketMQLocalRequestCallback<MessageExt>() {
+                @Override public void onSuccess(MessageExt message) {
+                    System.out.printf("receive messageExt: %s %n", message.toString());
+                }
+
+                @Override public void onException(Throwable e) {
+                    e.printStackTrace();
+                }
+            }, 5000);
+        } catch (MessagingException e) {
+            assertThat(e).hasMessageContaining("org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [127.0.0.1:9876] failed");
+        }
+    }
+
+    @Test
     public void testProperties() {
         assertThat(rocketMQTemplate.getProducer().getNamesrvAddr()).isEqualTo("127.0.0.1:9876");
         assertThat(rocketMQTemplate.getProducer().getProducerGroup()).isEqualTo("rocketMQTemplate-test-producer_group");
diff --git a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainerTest.java b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainerTest.java
index 60014c9..7133b98 100644
--- a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainerTest.java
+++ b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainerTest.java
@@ -16,12 +16,16 @@
  */
 package org.apache.rocketmq.spring.support;
 
+import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.core.RocketMQReplyListener;
 import org.junit.Test;
 import org.springframework.core.MethodParameter;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.converter.MappingJackson2MessageConverter;
 import org.springframework.messaging.converter.StringMessageConverter;
@@ -29,6 +33,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Date;
+import org.springframework.messaging.support.MessageBuilder;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -54,6 +59,64 @@
         });
         result = (Class) getMessageType.invoke(listenerContainer);
         assertThat(result.getName().equals(MessageExt.class.getName()));
+
+        listenerContainer.setRocketMQReplyListener(new RocketMQReplyListener<MessageExt, String>() {
+            @Override
+            public String onMessage(MessageExt message) {
+                return "test";
+            }
+        });
+        result = (Class) getMessageType.invoke(listenerContainer);
+        assertThat(result.getName().equals(MessageExt.class.getName()));
+
+        listenerContainer.setRocketMQReplyListener(new RocketMQReplyListener<String, String>() {
+            @Override
+            public String onMessage(String message) {
+                return "test";
+            }
+        });
+        result = (Class) getMessageType.invoke(listenerContainer);
+        assertThat(result.getName().equals(String.class.getName()));
+    }
+
+    @Test
+    public void testDoConvertMessage() throws Exception {
+        DefaultRocketMQListenerContainer listenerContainer = new DefaultRocketMQListenerContainer();
+        Method doConvertMessage = DefaultRocketMQListenerContainer.class.getDeclaredMethod("doConvertMessage", MessageExt.class);
+        doConvertMessage.setAccessible(true);
+
+        listenerContainer.setRocketMQListener(new RocketMQListener<String>() {
+            @Override
+            public void onMessage(String message) {
+            }
+        });
+
+        Field messageType = DefaultRocketMQListenerContainer.class.getDeclaredField("messageType");
+        messageType.setAccessible(true);
+        messageType.set(listenerContainer, String.class);
+        MessageExt messageExt = new MessageExt(0, System.currentTimeMillis(), null, System.currentTimeMillis(), null, null);
+        messageExt.setBody("hello".getBytes());
+        String result = (String) doConvertMessage.invoke(listenerContainer, messageExt);
+        assertThat(result).isEqualTo("hello");
+
+        listenerContainer.setRocketMQListener(new RocketMQListener<MessageExt>() {
+            @Override
+            public void onMessage(MessageExt message) {
+            }
+        });
+        Field messageType2 = DefaultRocketMQListenerContainer.class.getDeclaredField("messageType");
+        messageType2.setAccessible(true);
+        messageType2.set(listenerContainer, MessageExt.class);
+        messageExt = new MessageExt(0, System.currentTimeMillis(), null, System.currentTimeMillis(), null, null);
+        messageExt.setBody("hello".getBytes());
+        MessageExt result2 = (MessageExt) doConvertMessage.invoke(listenerContainer, messageExt);
+        assertThat(result2).isEqualTo(messageExt);
+
+        listenerContainer.setRocketMQListener(new RocketMQListener<User>() {
+            @Override
+            public void onMessage(User message) {
+            }
+        });
     }
 
     @Test
@@ -76,6 +139,41 @@
         assertThat(type.getRawType() == ArrayList.class);
         MethodParameter methodParameter = ((MethodParameter) getMethodParameter.invoke(listenerContainer));
         assertThat(methodParameter.getParameterType() == ArrayList.class);
+
+        listenerContainer.setRocketMQReplyListener(new RocketMQReplyListener<ArrayList<Date>, String>() {
+            @Override
+            public String onMessage(ArrayList<Date> message) {
+                return "test";
+            }
+        });
+
+        type = (ParameterizedType) getMessageType.invoke(listenerContainer);
+        assertThat(type.getRawType() == ArrayList.class);
+        methodParameter = ((MethodParameter) getMethodParameter.invoke(listenerContainer));
+        assertThat(methodParameter.getParameterType() == ArrayList.class);
+    }
+
+    class User {
+        private String userName;
+        private int userAge;
+
+        public String getUserName() {
+            return userName;
+        }
+
+        public User setUserName(String userName) {
+            this.userName = userName;
+            return this;
+        }
+
+        public int getUserAge() {
+            return userAge;
+        }
+
+        public User setUserAge(int userAge) {
+            this.userAge = userAge;
+            return this;
+        }
     }
 }