栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 前沿技术 > 大数据 > 大数据系统

全站最硬核 百万字强肝RocketMq源码 火热更新中~(十一)

全站最硬核 百万字强肝RocketMq源码 火热更新中~(十一)

文章目录

org.apache.rocketmq.remoting.netty.ResponseFutureorg.apache.rocketmq.remoting.common.SemaphoreReleaseOnlyOnce
可以看到这个回调执行器其实指的就是一个异步线程池,所以通过看源码,上面的注释我们可以这样理解:

如果异步线程池存在,则提交异步任务执行内置回调方法,如果异步线程池不存在,则当前线程直接执行。接下来我们看下所谓执行的回调方法内部是干了什么:

这个回调方法是写在responseFuture中的,所以我们直接把这个类通读下:

org.apache.rocketmq.remoting.netty.ResponseFuture
private final int opaque;
private final Channel processChannel;
private final long timeoutMillis;
private final InvokeCallback invokeCallback;
private final long beginTimestamp = System.currentTimeMillis();
private final CountDownLatch countDownLatch = new CountDownLatch(1);

private final SemaphoreReleaseOnlyOnce once;

private final AtomicBoolean executeCallbackonlyOnce = new AtomicBoolean(false);
private volatile RemotingCommand responseCommand;
private volatile boolean sendRequestOK = true;
private volatile Throwable cause;

首先是一些声明的变量。值得注意的是 一个只能释放一次的信号once和一个原子布尔变量executeCallbackOnlyOnce,含义是该回调方法只能被调用一次。关于SemaphoreReleaseOnlyOnce,过一下这个类:

org.apache.rocketmq.remoting.common.SemaphoreReleaseonlyOnce

这个类的内容本身不是很复杂,主要是对Semaphore做了一层包装:

private final AtomicBoolean released = new AtomicBoolean(false);
private final Semaphore semaphore;

public SemaphoreReleaseOnlyOnce(Semaphore semaphore) {
    this.semaphore = semaphore;
}

public void release() {
    if (this.semaphore != null) {
        if (this.released.compareAndSet(false, true)) {
            this.semaphore.release();
        }
    }
}

public Semaphore getSemaphore() {
    return semaphore;
}

代码量不大,主要是一个标记为只能释放一次的原子布尔变量,和一个release方法,逻辑就是如果没有被release的情况下,才能release

Semaphore是jdk1.5之后新增加的一个类,翻译为信号,但其实可以理解为一个锁,这个不属于rocketMQ源码,这里就不深究了,简单说就是必须从Semaphore中获取到一个锁,才能执行对应方法,释放后,其他线程才能获取该锁(当然允许同时执行的线程数是自定义的)。

而这里可以看到SemaphoreReleaseOnlyOnce,回调方法是只允许执行一次的,

再回到ResponseFuture,我们就明白了,也就是说为了保证回调方法只执行一次,作者做了两道限制:

一个是executeCallbackOnlyOnce原子布尔变量,一个是SemaphoreReleaseOnlyOnce互斥锁

除了

executeInvokeCallback()

方法外,再往下看:

public boolean isTimeout() {
    long diff = System.currentTimeMillis() - this.beginTimestamp;
    return diff > this.timeoutMillis;
}
转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/711388.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号