如何使用Netty与Avro异步回调

Question

我试图通过使用NettyServer实现实现异步Avro调用。在挖掘源代码之后，我发现了一个关于如何使用TestNettyServerWithCallbacks.java中的NettyServer的例子。

当运行一些测试时，我意识到NettyServer从不调用hello（Callback）方法，而是一直调用同步hello（）方法。客户端程序打印出“Hello”，但我期待“Hello-ASYNC”作为结果。我真的不知道发生了什么事。

我希望有人能照亮我，也许指出这个错误。以下是我用来执行简单的异步avro测试的代码。

AvroClient.java - 客户端代码。

public class AvroClient { 
    public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException { 
     try { 
      NettyTransceiver transceiver = new NettyTransceiver(new InetSocketAddress(6666)); 
      Chat.Callback client = SpecificRequestor.getClient(Chat.Callback.class, transceiver); 

      final CallFuture<CharSequence> future1 = new CallFuture<CharSequence>(); 
      client.hello(future1); 

      System.out.println(future1.get()); 

      transceiver.close(); 

     } catch (IOException ex) { 
      System.err.println(ex); 
     } 
    } 
} 

AvroNetty.java - 服务器代码

public class AvroNetty { 
    public static void main(String[] args) { 
     Index indexImpl = new AsyncIndexImpl(); 
     Chat chatImpl = new ChatImpl(); 

     Server server = new NettyServer(new SpecificResponder(Chat.class, chatImpl), new InetSocketAddress(6666)); 
     server.start(); 
     System.out.println("Server is listening at port " + server.getPort()); 
    } 
} 

ChatImpl.java

public class ChatImpl implements Chat.Callback { 
    @Override 
    public void hello(org.apache.avro.ipc.Callback<CharSequence> callback) throws IOException { 
     callback.handleResult("Hello-ASYNC");  
    } 

    @Override 
    public CharSequence hello() throws AvroRemoteException { 
     return new Utf8("Hello");  
    } 
} 

该接口是自动生成的Avro的工具 Chat.java

@SuppressWarnings("all") 
public interface Chat { 
    public static final org.apache.avro.Protocol PROTOCOL = org.apache.avro.Protocol.parse("{\"protocol\":\"Chat\",\"namespace\":\"avro.test\",\"types\":[],\"messages\":{\"hello\":{\"request\":[],\"response\":\"string\"}}}"); 
    java.lang.CharSequence hello() throws org.apache.avro.AvroRemoteException; 

    @SuppressWarnings("all") 
    public interface Callback extends Chat { 
     public static final org.apache.avro.Protocol PROTOCOL = avro.test.Chat.PROTOCOL; 
     void hello(org.apache.avro.ipc.Callback<java.lang.CharSequence> callback) throws java.io.IOException; 
    } 
} 

这里是t他的Avro架构

{ 
    "namespace": "avro.test", 
    "protocol": "Chat", 

    "types" : [], 

    "messages": { 
     "hello": { 
        "request": [], 
        "response": "string" 
     } 
    } 
} 

Answer 1

的NettyServer实现实际上没有实现异步风格可言。这是图书馆的一个缺陷。相反，您需要指定一个异步执行处理程序，而不是通过回调尝试和链接服务。这里是我用来设置我的NettyServer允许这个：

ExecutorService es = Executors.newCachedThreadPool(); 
OrderedMemoryAwareThreadPoolExecutor executor = new OrderedMemoryAwareThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), 0, 0); 
ExecutionHandler executionHandler = new ExecutionHandler(executor); 
final NettyServer server = new NettyServer(responder, addr, new NioServerSocketChannelFactory(es, es), executionHandler);