如何使用Netty与Avro异步回调
问题描述:
我试图通过使用NettyServer实现实现异步Avro调用。在挖掘源代码之后,我发现了一个关于如何使用TestNettyServerWithCallbacks.java中的NettyServer的例子。如何使用Netty与Avro异步回调
当运行一些测试时,我意识到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"
}
}
}
答
的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);
嘿Carbotex,我有同样的问题。你有没有想过为什么会发生? – Wanderer 2012-07-18 13:21:09