zoukankan      html  css  js  c++  java
  • RocketMQ的异步调用

    这个异步调用方法中传入一个final 回调对象。

        public void invokeAsyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis,
                                    final InvokeCallback invokeCallback)
                throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
            System.out.println("NettyRemotingAbstract.invokeAsyncImpl()****************");
            final int opaque = request.getOpaque();
            //超时信号量锁
            boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
            if (acquired) {
                final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);
    
                //这个ResonseFuture的封装的思路。
                final ResponseFuture responseFuture = new ResponseFuture(opaque, timeoutMillis, invokeCallback, once);
                this.responseTable.put(opaque, responseFuture);
                try {
                    //----》执行io请求,添加channelfuture监听器
                    channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture f) throws Exception {
                            //这里只是发送成功!!!!!!!!!!!!!!!!!!!!!!!!!
                            if (f.isSuccess()) {
                                responseFuture.setSendRequestOK(true);
                                return;
                            } else {
                                responseFuture.setSendRequestOK(false);
                            }
                            //这里只是发送成功,返回设为nul。putResponse -> NULL
                            responseFuture.putResponse(null);
                            responseTable.remove(opaque);
                            try {
                                //发送成功后还没有返回消息时的,调用回调方法。参见:MQClientAPIImpl.sendMessageAsync(..)
                                System.out.println("**************NettyRemotingAbstract.invokeAsyncImpl().callback()");
                                responseFuture.executeInvokeCallback();
                            } catch (Throwable e) {
                                plog.warn("excute callback in writeAndFlush addListener, and callback throw", e);
                            } finally {
                                responseFuture.release();
                            }
    
                            plog.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
                        }
                    });
                } catch (Exception e) {
                    responseFuture.release();
                    plog.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
                    throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
                }
            } else {
                String info =
                        String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d", //
                                timeoutMillis, //
                                this.semaphoreAsync.getQueueLength(), //
                                this.semaphoreAsync.availablePermits()//
                        );
                plog.warn(info);
                throw new RemotingTooMuchRequestException(info);
            }
        }

    我们往上面看看这个回调对象的回调方法:

     

  • 相关阅读:
    安卓开发环境的演变
    对软件工程实践课程的预定目标
    Angular4.0踩坑之路:探索子路由和懒加载
    Angular4.0踩坑之路:如何成功读取本地json文件
    Angular踩坑之路:在Angular中监听键盘事件
    Angular踩坑之路:初探Angular过程中的一些总结与记录
    Angular踩坑之路:初探webpack
    Angular踩坑之路:设置开发环境
    (Nodejs)安装教程、切换全局模块安装路径、切换npm下载源
    (ES、ik分词器)ES的分词器插件elasticsearch-analysis-ik
  • 原文地址:https://www.cnblogs.com/guazi/p/6675760.html
Copyright © 2011-2022 走看看