tang-jie / nettyrpc Goto Github PK

看了一下源码，发现这个地方感觉直接这样写，是否有问题，假如客户端同时配置了调用多个不同远程地址服务的接口服务，这样获取messageSendHandler 对象是否可行？

您好，我学习Java半年多了，最近在学习这个项目的源代码，我觉得学到了很多东西。

不过我有点疑惑，这个Listener的作用是什么，感觉好像就是在ModuleProviderWrapper里面不断地循环调用这些listener的exported方法。能不能加一个Test示例 ><

最后感谢您对于源代码的分享！！

rpc客户端如果先启动，不会主动重连服务器。异常没有捕捉

刚开始研究这份代码,看了 Spring中Bean的生命中期与InitializingBean和DisposableBean接口,说 init-method 方法比实现InitializingBean接口来得好;

能不能在自定义标签里面写自定义初始化和销毁方法

类似 :

<bean init-method="init" destroy-method="destroy"/>

架构如下：
Jstorm实时拉源数据，数据交给bolt处理，bolt处理完的数据同步该框架把数据回写到远程，当处理20W数据的时候，会出现超时的情况，无论同步的RPC或者异步的RPC请求，超时时间30s，网络状态良好，我把博主的代码copy到项目中，报错信息如下：
cn.com.jit.pmi2.collection.netty.rpc.exception.InvokeTimeoutException: Timeout request,NettyRPC server request timeout!
at cn.com.jit.pmi2.collection.netty.rpc.core.MessageCallBack.await(MessageCallBack.java:86)
at cn.com.jit.pmi2.collection.netty.rpc.core.MessageCallBack.start(MessageCallBack.java:48)
at cn.com.jit.pmi2.collection.netty.rpc.netty.MessageSendProxy.handleInvocation(MessageSendProxy.java:43)
at com.google.common.reflect.AbstractInvocationHandler.invoke(AbstractInvocationHandler.java:84)
at com.sun.proxy.$Proxy98.addOrUpdateUser(Unknown Source)
at cn.com.jit.pmi2.collection.topology.bolt.user.AddUserBolt.execute(AddUserBolt.java:71)
at com.alibaba.jstorm.task.execute.BoltExecutors.processTupleEvent(BoltExecutors.java:248)
at com.alibaba.jstorm.task.execute.BoltExecutors.onEvent(BoltExecutors.java:184)
at backtype.storm.utils.DisruptorQueueImpl.consumeBatchToCursor(DisruptorQueueImpl.java:261)
at backtype.storm.utils.DisruptorQueueImpl.consumeBatchWhenAvailable(DisruptorQueueImpl.java:226)
at backtype.storm.utils.DisruptorQueueImpl.consumeBatchWhenAvailable(DisruptorQueueImpl.java:185)
at com.alibaba.jstorm.task.execute.BoltExecutors.run(BoltExecutors.java:148)
at com.alibaba.jstorm.callback.AsyncLoopRunnable.run(AsyncLoopRunnable.java:95)
at java.lang.Thread.run(Thread.java:748)

另外可以提供一个q群学习交流哈

public class MethodProxyAdvisor implements MethodInterceptor {
...
boolean existFilter = ServiceFilterBinder.class.isAssignableFrom(serviceBean.getClass());
...
}

无论是否配置过滤器，都会返回true.

在阅读作者源码的时候，发现了这样一个问题：

在单例RpcServerLoader中封装了messageSendHandler：

public class RpcServerLoader {
  ...
  private MessageSendHandler messageSendHandler = null;
  ...
}

而messageHandler一个的重要作用就是就是向RPC服务器发送信息：

public class MessageSendProxy<T> extends AbstractInvocationHandler {

    @Override
    public Object handleInvocation(Object proxy, Method method, Object[] args) throws Throwable {
		...
        MessageSendHandler handler = RpcServerLoader.getInstance().getMessageSendHandler();
        MessageCallBack callBack = handler.sendRequest(request);
        return callBack.start();
    }
}

这样就会发生一个bug，就是当我们的客户端连接到了多台RPC服务器时，就会产生多个messageHandler对象，所以当我们执行：

RpcServerLoader.getInstance().getMessageSendHandler();

的时候。就可能拿到的是别的PRC服务器的messageHandler，然后往错误的IP地址发送请求。

我们可以看下面的例子：

设置两个service调用不同地址的RPC服务：

在MessageHandler的channelRead方法中设置一个日志：

其他代码都不变。

在两台电脑都启动RPC服务器，调用RpcParallelTest测试用例（parallel =1）

可以发现，原本要调用本地RPC服务的加法运行，调用了另一台机器上RPC服务的。

究其原因，我觉得应该是RpcServerLoader设计成了单例模式，而MessageSendHandler却不是单例而导致的。

dubbo framework also use netty implementing the rpc communication,but it is too large

MessageCallBack类

public Object start() throws InterruptedException {
        try {
            lock.lock();
            finish.await(10 * 1000, TimeUnit.MILLISECONDS);
            if (this.response != null) {
                return this.response.getResult();
            } else {
                return null;
            }
        } finally {
            lock.unlock();
        }
 }

不明白这句代码何意 finish.await(10 * 1000, TimeUnit.MILLISECONDS); 一定要嘛？

是否应该先设置返回值，再唤醒等待好些？

不能在同一个线程中调用sync方法,会触发NioEventLoop的死锁检查。 此处sync()方法实际会执行失败。
可以在operationComplete方法结尾尝试System.out.println标志来验证。

如题，望回复。
另外，我想在你的2.0的基础上，改造成IM即时通讯服务器，客户端为手机端。

看代码是没有对netty的长连接有重连机制，后续是否会考虑？

你好，请问就是服务一段时间未访问，等再次访问之后finish.await会出现超时，有什么办法可以避免

像业务相关的包不应该出现在同一个maven模块中，如果第三方使用的话就引入了无关的代码；
建议放到另外一个demo maven模块中，该模块依赖rpc核心模块

请问，这个可以作手游服务端网络层框架嘛，或者需要做哪些变化才能实现

An exception was thrown by com.newlandframework.rpc.netty.MessageRecvExecutor$2.operationComplete()
io.netty.util.concurrent.BlockingOperationException: AbstractChannel$CloseFuture@12f3c0d1(incomplete)
at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:391) ~[netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.channel.DefaultChannelPromise.checkDeadLock(DefaultChannelPromise.java:157) ~[netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:252) ~[netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.channel.DefaultChannelPromise.await(DefaultChannelPromise.java:129) ~[netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.channel.DefaultChannelPromise.await(DefaultChannelPromise.java:28) ~[netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.util.concurrent.DefaultPromise.sync(DefaultPromise.java:219) ~[netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.channel.DefaultChannelPromise.sync(DefaultChannelPromise.java:117) ~[netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.channel.DefaultChannelPromise.sync(DefaultChannelPromise.java:28) ~[netty-all-4.0.36.Final.jar:4.0.36.Final]
at com.newlandframework.rpc.netty.MessageRecvExecutor$2.operationComplete(MessageRecvExecutor.java:144) ~[classes/:?]
at com.newlandframework.rpc.netty.MessageRecvExecutor$2.operationComplete(MessageRecvExecutor.java:135) ~[classes/:?]
at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:683) [netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.util.concurrent.DefaultPromise$LateListeners.run(DefaultPromise.java:854) [netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.util.concurrent.DefaultPromise$LateListenerNotifier.run(DefaultPromise.java:882) [netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:358) [netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:374) [netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112) [netty-all-4.0.36.Final.jar:4.0.36.Final]
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137) [netty-all-4.0.36.Final.jar:4.0.36.Final]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
[id: 0x891ad835] REGISTERED
[id: 0x891ad835] BIND(0.0.0.0/0.0.0.0:18886)
You can open your web browser see NettyRPC server api interface: http://127.0.0.1:18886/NettyRPC.html
[id: 0x891ad835, L:/0:0:0:0:0:0:0:0:18886] ACTIVE
Disconnected from the target VM, address: '127.0.0.1:51430', transport: 'socket'

建议编写-快速开始手册

当想整个远程服务使用spring原生事务的时候

@Service
public class PersonManageImpl implements PersonManage {

    @Transactional
    public int save(Person p) {
        //your business logic code here!
        System.out.println("person data[" + p + "] has save!");
        return 0;
    }
}

就会出现server假死的问题，在下面这个方法假死住

 private Object reflect(MessageRequest request) throws Throwable {
        String className = request.getClassName();
        Object serviceBean = handlerMap.get(className);
        String methodName = request.getMethodName();
        Object[] parameters = request.getParameters();
        return MethodUtils.invokeMethod(serviceBean, methodName, parameters);
    }

哪里添加header 或者 metadata呀， 需要添加token