From c34a6d8f49c91b884a399c733d9e521408dd8eab Mon Sep 17 00:00:00 2001
From: Redkale <8730487+redkale@users.noreply.github.com>
Date: Fri, 29 May 2020 22:34:53 +0800
Subject: [PATCH]

---
 src/org/redkale/mq/MessageAgent.java    | 4 ++--
 src/org/redkale/mq/MessageConsumer.java | 7 ++++---
 2 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/src/org/redkale/mq/MessageAgent.java b/src/org/redkale/mq/MessageAgent.java
index 93cf41f26..addc80d53 100644
--- a/src/org/redkale/mq/MessageAgent.java
+++ b/src/org/redkale/mq/MessageAgent.java
@@ -6,7 +6,7 @@
 package org.redkale.mq;
 
 import java.util.*;
-import java.util.function.Function;
+import java.util.function.*;
 import java.util.logging.Logger;
 import org.redkale.boot.*;
 import org.redkale.net.http.Rest;
@@ -68,7 +68,7 @@ public abstract class MessageAgent {
     public abstract List<String> queryTopic();
 
     //创建指定topic的消费处理器
-    public abstract MessageConsumer createConsumer(String topic, MessageProcessor processor);
+    public abstract MessageConsumer createConsumer(String topic, Consumer<MessageRecord> processor);
 
     //创建指定topic的生产处理器
     public abstract MessageProducer createProducer();
diff --git a/src/org/redkale/mq/MessageConsumer.java b/src/org/redkale/mq/MessageConsumer.java
index 74261993b..565bd3cac 100644
--- a/src/org/redkale/mq/MessageConsumer.java
+++ b/src/org/redkale/mq/MessageConsumer.java
@@ -6,6 +6,7 @@
 package org.redkale.mq;
 
 import java.util.Objects;
+import java.util.function.Consumer;
 import java.util.logging.Logger;
 
 /**
@@ -19,20 +20,20 @@ public abstract class MessageConsumer extends Thread {
 
     protected final String topic;
 
-    protected final MessageProcessor processor;
+    protected final Consumer<MessageRecord> processor;
 
     protected final Logger logger = Logger.getLogger(this.getClass().getSimpleName());
 
     protected volatile boolean closed;
 
-    protected MessageConsumer(String topic, MessageProcessor processor) {
+    protected MessageConsumer(String topic, Consumer<MessageRecord> processor) {
         Objects.requireNonNull(topic);
         Objects.requireNonNull(processor);
         this.topic = topic;
         this.processor = processor;
     }
 
-    public MessageProcessor getProcessor() {
+    public Consumer<MessageRecord> getProcessor() {
         return processor;
     }