java – Instrument Netty的EventLoop,用于确定性执行计划任务
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了java – Instrument Netty的EventLoop,用于确定性执行计划任务，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含15907字，纯文字阅读大概需要23分钟。
内容图文
我想调整Netty的EventLoop以便：
>以确定的顺序运行任务.
>抓住计划任务的截止日期.
>快进虚拟时钟,触发执行截止日期的任务.
我知道EmbeddedChannel,我在一些测试中使用它.但我想要的是单元测试和集成测试之间的问题,这些测试对于某些极端情况仍然是盲目的.断开连接 – 重新连接和ping方案会大量使用计划任务.我可以添加具有巨大延迟的确定性,但我不希望我的测试等待几秒或更长时间.因此,使用Netty的EventLoop看起来就像是解决方案.
我已经编写了至少对我有用的代码.
>我修改ScheduledFutureTask#nanoTime以返回我的值.
>我派生出NioEventLoopGroup,所以我抓住任务截止日期.
>我改变了ScheduledFutureTask#nanoTime返回的值.
有趣的是,Netty代码只依赖于ScheduledFutureTask#nanoTime(伟大的设计！)返回的值,所以这是一个非常有限的变化.我使用ByteBuddy来避免复制粘贴Netty代码,但这并不重要.
由于AbstractScheduledEventExecutor#pollScheduledTask(long)具有空队列,因此仅调度1个任务时,像InstrumentedNioEventLoopGroupTest这样的非常简单的测试会失败.
我发现每个NioEventLoop都有自己的任务队列,并且可能不会发生队列轮询,因为NioEventLoopGroup等待Selector发出信号,这是有道理的.所以我将NioEventLoopGroup的线程数增加到2.我还尝试将ioRatio设置为1并安排更多任务,没有更好的结果.使用调试器,似乎我的任务总是“落在”未轮询的任务队列中.
有什么想法使这项工作？我正在使用Netty 4.1.24.Final.
ScheduledFutureTaskHack.java
package com.otcdlink.chiron.integration.harness;
import com.otcdlink.chiron.toolbox.ToStringTools;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.agent.ByteBuddyAgent;
import net.bytebuddy.dynamic.loading.ClassReloadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.concurrent.Future;
import java.util.function.LongSupplier;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static net.bytebuddy.matcher.ElementMatchers.isPackagePrivate;
import static net.bytebuddy.matcher.ElementMatchers.isStatic;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.returns;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
/**
*
* Got the delegation working with the help of
* https://www.infoq.com/articles/Easily-Create-Java-Agents-with-ByteBuddy
*/
final class ScheduledFutureTaskHack {
private static final Logger LOGGER = LoggerFactory.getLogger( ScheduledFutureTaskHack.class ) ;
private static final Class< ? > SCHEDULEDFUTURETASK_CLASS ;
private static final Method SCHEDULEDFUTURETASK_NANOTIME_METHOD ;
private static final Method SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD ;
private static final Field SCHEDULEDFUTURETASK_DEADLINENANOS_FIELD ;
private static final Field SCHEDULEDFUTURETASK_STARTTIME_FIELD ;
static {
try {
SCHEDULEDFUTURETASK_CLASS = Class.forName( "io.netty.util.concurrent.ScheduledFutureTask" ) ;
SCHEDULEDFUTURETASK_NANOTIME_METHOD =
SCHEDULEDFUTURETASK_CLASS.getDeclaredMethod( "nanoTime" ) ;
SCHEDULEDFUTURETASK_NANOTIME_METHOD.setAccessible( true ) ;
SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD =
SCHEDULEDFUTURETASK_CLASS.getDeclaredMethod( "deadlineNanos") ;
SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD.setAccessible( true ) ;
SCHEDULEDFUTURETASK_DEADLINENANOS_FIELD =
SCHEDULEDFUTURETASK_CLASS.getDeclaredField( "deadlineNanos" ) ;
SCHEDULEDFUTURETASK_DEADLINENANOS_FIELD.setAccessible( true ) ;
SCHEDULEDFUTURETASK_STARTTIME_FIELD =
SCHEDULEDFUTURETASK_CLASS.getDeclaredField( "START_TIME" ) ;
SCHEDULEDFUTURETASK_STARTTIME_FIELD.setAccessible( true ) ;
} catch( ClassNotFoundException | NoSuchMethodException | NoSuchFieldException e ) {
throw new Error( e ) ;
}
}
/**
* Everything is this class must be visible from the redefined class.
*/
@SuppressWarnings( "unused" )
public static final class StaticMethodDelegate {
/**
* Calls to {@link io.netty.util.concurrent.ScheduledFutureTask#nanoTime()} are redirected
* to this method.
* Sadly we can't use parameter annotated with {@link @This} or something giving a hint
* about the call context. It looks like a consequence of JVMTI reload not supporting method
* addition (adding a parameter would imply creating a new method).
*/
public static long nanoTime() {
final long supplied = longSupplier.getAsLong() ;
LOGGER.debug( "Called " + StaticMethodDelegate.class.getSimpleName() + "#nanoTime(), " +
"returns " + supplied + "." ) ;
return supplied ;
}
}
private static LongSupplier longSupplier = null ;
static void install( final LongSupplier longSupplier ) {
install( longSupplier, true ) ;
}
/**
*
* @param longSupplier
* @param suppliedNanosRelativeToClassloadingTime if {@code true}, supplied nanoseconds are
* relative to {@link io.netty.util.concurrent.ScheduledFutureTask#START_TIME}.
* Original behavior of the hacked method is to substract
* {@link io.netty.util.concurrent.ScheduledFutureTask#START_TIME} from value returned
* by {@link System#nanoTime()} (probably to make number more readable and reduce the risk
* of an overflow). During tests we prefer to not care about start time so there is this
* option to add it automatically.
*/
static void install(
final LongSupplier longSupplier,
final boolean suppliedNanosRelativeToClassloadingTime
) {
checkState( ScheduledFutureTaskHack.longSupplier == null ) ;
if( suppliedNanosRelativeToClassloadingTime ) {
final long startTime = START_TIME ;
LOGGER.debug(
"Installing with value of " +
SCHEDULEDFUTURETASK_STARTTIME_FIELD.toGenericString() +
" = " + startTime + " automatically added to the values supplied."
) ;
class AdjustedLongSupplier implements LongSupplier {
@Override
public long getAsLong() {
return longSupplier.getAsLong() + startTime ;
}
@Override
public String toString() {
return ToStringTools.getNiceClassName( this ) + "{startTime=" + startTime + "}" ;
}
}
ScheduledFutureTaskHack.longSupplier = new AdjustedLongSupplier() ;
} else {
ScheduledFutureTaskHack.longSupplier = checkNotNull( longSupplier ) ;
}
ByteBuddyAgent.install() ;
LOGGER.info( "Successfully installed ByteBuddy Agent." ) ;
redefineClass() ;
LOGGER.info( "Successfully redefined static method implementation." ) ;
}
private static void redefineClass() {
new ByteBuddy()
.redefine( SCHEDULEDFUTURETASK_CLASS )
.method( named( "nanoTime" )
.and( isStatic() )
.and( isPackagePrivate() )
.and( takesArguments( 0 ) )
.and( returns( long.class ) )
)
.intercept( MethodDelegation.to( StaticMethodDelegate.class ) )
.make()
.load( ScheduledFutureTaskHack.class.getClassLoader(), ClassReloadingStrategy.fromInstalledAgent() )
;
}
/**
* Invokes method replacing {@link io.netty.util.concurrent.ScheduledFutureTask#nanoTime()}.
*/
public static long invokeNanoTime() {
try {
return ( long ) SCHEDULEDFUTURETASK_NANOTIME_METHOD.invoke( null ) ;
} catch( IllegalAccessException | InvocationTargetException e ) {
throw new Error( e ) ;
}
}
/**
* The {@link io.netty.util.concurrent.ScheduledFutureTask#deadlineNanos()} method returns
* the value made from {@link System#nanoTime()},
* minus {@link io.netty.util.concurrent.ScheduledFutureTask#START_TIME},
* plus the delay before executing the task.
*/
public static Long invokeDeadlineNanos( final Future future ) {
try {
if( SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD.getDeclaringClass()
.isAssignableFrom( future.getClass() )
) {
return ( long ) SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD.invoke( future ) ;
} else {
return null ;
}
} catch( IllegalAccessException | InvocationTargetException e ) {
throw new Error(
"Could not access method " + SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD + " in " + future,
e
) ;
}
}
private static long readStartTime() {
try {
return ( long ) SCHEDULEDFUTURETASK_STARTTIME_FIELD.get( null ) ;
} catch( IllegalAccessException e ) {
throw new Error(
"Could not access static field " + SCHEDULEDFUTURETASK_STARTTIME_FIELD,
e
) ;
}
}
public static final long START_TIME = readStartTime() ;
}
ScheduledFutureTaskHackTest.java
package com.otcdlink.chiron.integration.harness;
import com.otcdlink.chiron.toolbox.ToStringTools;
import com.otcdlink.chiron.toolbox.netty.NettyTools;
import io.netty.channel.nio.NioEventLoopGroup;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static org.assertj.core.api.Assertions.assertThat;
public class ScheduledFutureTaskHackTest {
@Test
public void fastForward() throws InterruptedException {
final AtomicLong nanotimeHolder = new AtomicLong( 0 ) ;
ScheduledFutureTaskHack.install( nanotimeHolder::get ) ;
final long startTime = hackedNanoTime() ;
final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup() ;
final Semaphore scheduledTaskCompleted = new Semaphore( 0 ) ;
nioEventLoopGroup.schedule(
() -> {
scheduledTaskCompleted.release() ;
LOGGER.info( "Scheduled task completed." ) ;
},
1,
TimeUnit.HOURS
) ;
LOGGER.info( "Scheduled task for in 1 hour, now fast-forwarding Netty's clock ..." ) ;
// Test fails when disabling fast-forward below.
nanotimeHolder.set( startTime + TimeUnit.HOURS.toNanos( 1 ) + 1 ) ;
Thread.sleep( 1000 ) ;
hackedNanoTime() ;
// Amazingly Netty detected clock change and ran the task!
assertThat( scheduledTaskCompleted.tryAcquire( 1, TimeUnit.SECONDS ) )
.describedAs( "Scheduled task should have completed within 1 second" )
.isTrue()
;
}
// =======
// Fixture
// =======
private static final Logger LOGGER = LoggerFactory.getLogger(
ScheduledFutureTaskHackTest.class ) ;
static {
NettyTools.forceNettyClassesToLoad() ;
}
private static long hackedNanoTime() {
final long nanoTime = ScheduledFutureTaskHack.invokeNanoTime() ;
LOGGER.info(
ToStringTools.getNiceName( ScheduledFutureTaskHack.StaticMethodDelegate.class ) +
"#nanoTime(): " + nanoTime + "."
) ;
return nanoTime ;
}
}
InstrumentedNioEventLoopGroup.java
package com.otcdlink.chiron.integration.harness;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.ScheduledFuture;
import javax.annotation.Nonnull;
import java.time.Instant;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import static com.google.common.base.Preconditions.checkNotNull;
class InstrumentedNioEventLoopGroup extends NioEventLoopGroup {
/**
* Consume the value obtained from
* {@link io.netty.util.concurrent.ScheduledFutureTask#deadlineNanos()}.
* This is hardly mappable to an exact {@link Instant} (even if the Java flavor retains
* nanoseconds) but this is enough to compare with {@link System#nanoTime()}.
*/
private final Consumer< Long > scheduledTaskMomentConsumer ;
public InstrumentedNioEventLoopGroup(
final ThreadFactory threadFactory,
final Consumer< Long > scheduledTaskMomentConsumer
) {
// Need 2 threads because one will block on Socket Selector if there is no IO,
// so we add one to poll Tasks.
super( 2, threadFactory ) ;
this.scheduledTaskMomentConsumer = checkNotNull( scheduledTaskMomentConsumer ) ;
}
private < FUTURE extends Future > FUTURE recordDeadlineNanos( final FUTURE future ) {
final Long deadlineNanos = ScheduledFutureTaskHack.invokeDeadlineNanos( future ) ;
if( deadlineNanos != null ) {
scheduledTaskMomentConsumer.accept( deadlineNanos ) ;
}
return future ;
}
@Nonnull
@Override
public Future< ? > submit( final Runnable task ) {
return recordDeadlineNanos( super.submit( task ) ) ;
}
@Nonnull
@Override
public < T > Future< T > submit(
final Runnable task,
final T result
) {
return recordDeadlineNanos( super.submit( task, result ) ) ;
}
@Nonnull
@Override
public < T > Future< T > submit( final Callable< T > task ) {
return recordDeadlineNanos( super.submit( task ) ) ;
}
@Nonnull
@Override
public ScheduledFuture< ? > schedule(
final Runnable command,
final long delay,
final TimeUnit unit
) {
return recordDeadlineNanos( super.schedule( command, delay, unit ) ) ;
}
@Nonnull
@Override
public < V > ScheduledFuture< V > schedule(
final Callable< V > callable,
final long delay,
final TimeUnit unit
) {
return recordDeadlineNanos( super.schedule( callable, delay, unit ) ) ;
}
@Nonnull
@Override
public ScheduledFuture< ? > scheduleAtFixedRate(
final Runnable command,
final long initialDelay,
final long period,
final TimeUnit unit
) {
return recordDeadlineNanos(
super.scheduleAtFixedRate( command, initialDelay, period, unit ) ) ;
}
@Nonnull
@Override
public ScheduledFuture< ? > scheduleWithFixedDelay(
final Runnable command,
final long initialDelay,
final long delay,
final TimeUnit unit
) {
return recordDeadlineNanos(
super.scheduleWithFixedDelay( command, initialDelay, delay, unit ) ) ;
}
}
InstrumentedNioEventLoopGroupTest.java
package com.otcdlink.chiron.integration.harness;
import com.otcdlink.chiron.toolbox.concurrent.ExecutorTools;
import com.otcdlink.chiron.toolbox.netty.NettyTools;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.ScheduledFuture;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static org.assertj.core.api.Assertions.assertThat;
public class InstrumentedNioEventLoopGroupTest {
@Test
public void recordAndAdjust() throws InterruptedException {
final int delay = 10 ;
final TimeUnit timeUnit = TimeUnit.SECONDS ;
final AtomicLong nanoInstantSupplier = new AtomicLong() ;
ScheduledFutureTaskHack.install( nanoInstantSupplier::get ) ;
final List< Long > taskDeadlineRecorder = Collections.synchronizedList( new ArrayList<>() ) ;
final InstrumentedNioEventLoopGroup executor = new InstrumentedNioEventLoopGroup(
ExecutorTools.newThreadFactory( "executor" ), taskDeadlineRecorder::add ) ;
executor.setIoRatio( 1 ) ; // Silly but worth trying to see what can get wrong.
final Semaphore doneSemaphore = new Semaphore( 0 ) ;
final ScheduledFuture< ? > scheduledFuture1 =
executor.schedule( ( Runnable ) doneSemaphore::release, delay, timeUnit ) ;
LOGGER.info( "Scheduled " + scheduledFuture1 + "." ) ;
assertThat( taskDeadlineRecorder ).hasSize( 1 ) ;
final Long nanoTime = taskDeadlineRecorder.get( 0 ) - ScheduledFutureTaskHack.START_TIME ;
LOGGER.info( "Recorded " + nanoTime + " as nanoTime deadline for next task." ) ;
assertThat( nanoTime ).isEqualTo( timeUnit.toNanos( delay ) ) ;
final long pastDeadline = nanoTime + 1 ;
nanoInstantSupplier.set( pastDeadline ) ;
LOGGER.info(
"Did set nanoTime to " + pastDeadline + ", past to Task's deadline. " +
"Invocation of hacked nanoTime() returns " +
ScheduledFutureTaskHack.invokeNanoTime() + "."
) ;
LOGGER.info( "Now waiting for task completion ..." ) ;
assertThat( doneSemaphore.tryAcquire( 3, TimeUnit.SECONDS ) ).isTrue() ;
}
/**
* Fails when ran after {@link #recordAndAdjust()} because JUnit doesn't reload classes for
* each method inside a test class.
*/
@Test
public void noInstrumentation() throws InterruptedException {
final NioEventLoopGroup executor =
new NioEventLoopGroup( 1, ExecutorTools.newThreadFactory( "executor" ) ) ;
final Semaphore doneSemaphore = new Semaphore( 0 ) ;
executor.submit( () -> LOGGER.info( "Plain submission works!" ) ) ;
final ScheduledFuture< ? > scheduledFuture =
executor.schedule( ( Runnable ) doneSemaphore::release, 1, TimeUnit.SECONDS ) ;
LOGGER.info( "Scheduled " + scheduledFuture + "." ) ;
assertThat( doneSemaphore.tryAcquire( 3, TimeUnit.SECONDS ) ).isTrue() ;
}
// =======
// Fixture
// =======
private static final Logger LOGGER =
LoggerFactory.getLogger( InstrumentedNioEventLoopGroupTest.class ) ;
static {
NettyTools.forceNettyClassesToLoad() ;
}
}
我是Chiron Framework的作者
,一个基于WebSocket的网络框架,具有纯Java客户端和非阻塞双因素身份验证.它使用Netty很多.遗憾的是,有许多基于JMockit的测试无法可靠运行,因为执行顺序可能是非确定性的(这是调度任务的每一段代码的问题).
解决方法:
哦,伙计们很明显：我覆盖每个schedule *方法,将Runnable / Callable和其他参数堆积在一个数据对象中,并将其添加到某个队列中.然后我从测试中明确地触发任务执行.
由于我创建任务的代码是我的,我使用标记接口装饰每个任务(恰好是方法引用).然后测试可以检查它运行预期的任务.
内容总结
以上是互联网集市为您收集整理的java – Instrument Netty的EventLoop,用于确定性执行计划任务全部内容，希望文章能够帮你解决java – Instrument Netty的EventLoop,用于确定性执行计划任务所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】