Netty客户端 异步结果通知类开发

byit-plugin-core/myth-plugin-rpc-client/src/main/java/com/byit/future/PluginFutureResponse.java

+package com.byit.future;
+
+import com.byit.factory.PluginClientFactory;
+import com.byit.packet.request.PluginRpcRequestPacket;
+import com.byit.packet.response.PluginRpcResponsePacket;
+import com.byit.param.PluginCallback;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * 结果对象获取方法
+ * 这个对象封装了  在获取对象时加锁和 结果集被返回时解锁的方法
+ * @author huangfu
+ */
+public class PluginFutureResponse implements Future<PluginRpcResponsePacket> {
+    private PluginClientFactory pluginClientFactory;
+
+    private PluginRpcRequestPacket rpcRequestPacket;
+    private PluginRpcResponsePacket rpcResponsePacket;
+    private PluginCallback pluginCallback;
+
+    private boolean done = false;
+    /**
+     * 锁对象    在获取执行结果时加锁   在结果被返回时 解锁
+     */
+    private final Object lock = new Object();
+
+    public PluginFutureResponse(PluginClientFactory pluginClientFactory, PluginRpcRequestPacket rpcRequestPacket, PluginCallback pluginCallback) {
+        this.pluginClientFactory = pluginClientFactory;
+        this.rpcRequestPacket = rpcRequestPacket;
+        this.pluginCallback = pluginCallback;
+        //设置缓存
+        this.pluginClientFactory.setPluginFutureResponseMap(this.rpcRequestPacket.getRequestId(),this);
+    }
+
+    public void remove(String key){
+        this.pluginClientFactory.removePluginFutureResponseMap(key);
+    }
+
+    public PluginClientFactory getPluginClientFactory() {
+        return pluginClientFactory;
+    }
+
+    public void setPluginClientFactory(PluginClientFactory pluginClientFactory) {
+        this.pluginClientFactory = pluginClientFactory;
+    }
+
+    public PluginRpcRequestPacket getRpcRequestPacket() {
+        return rpcRequestPacket;
+    }
+
+    public void setRpcRequestPacket(PluginRpcRequestPacket rpcRequestPacket) {
+        this.rpcRequestPacket = rpcRequestPacket;
+    }
+
+    public PluginRpcResponsePacket getRpcResponsePacket() {
+        return rpcResponsePacket;
+    }
+
+    public void setRpcResponsePacket(PluginRpcResponsePacket rpcResponsePacket) {
+        this.rpcResponsePacket = rpcResponsePacket;
+        synchronized (lock){
+            done = true;
+            lock.notifyAll();
+        }
+    }
+
+    public PluginCallback getPluginCallback() {
+        return pluginCallback;
+    }
+
+    public void setPluginCallback(PluginCallback pluginCallback) {
+        this.pluginCallback = pluginCallback;
+    }
+
+    @Override
+    public boolean cancel(boolean mayInterruptIfRunning) {
+        return false;
+    }
+
+    @Override
+    public boolean isCancelled() {
+        return false;
+    }
+
+
+
+    @Override
+    public boolean isDone() {
+        return done;
+    }
+
+    @Override
+    public PluginRpcResponsePacket get() throws InterruptedException, ExecutionException {
+        try {
+            rpcResponsePacket = this.get(-1, TimeUnit.MILLISECONDS);
+
+        } catch (TimeoutException e) {
+            throw new RuntimeException(e);
+        }
+        return rpcResponsePacket;
+    }
+
+    @Override
+    public PluginRpcResponsePacket get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
+        if (!done) {
+            synchronized (lock){
+                try {
+                    if (timeout < 0) {
+                        lock.wait();
+                    } else {
+                        long timeoutMillis = (TimeUnit.MILLISECONDS==unit)?timeout:TimeUnit.MILLISECONDS.convert(timeout , unit);
+                        lock.wait(timeoutMillis);
+                    }
+                } catch (InterruptedException e) {
+                    throw e;
+                }
+            }
+        }
+        if (!done) {
+            throw new RuntimeException("plugin result timeout:"+rpcRequestPacket);
+        }
+
+        return rpcResponsePacket;
+    }
+}