import org.jetlang.fibers.PoolFiberFactory; //导入依赖的package包/类
public AsyncProcessor() {
    ExecutorService exec = Executors.newCachedThreadPool();
    PoolFiberFactory factory = new PoolFiberFactory(exec);

    this.fiberThread = factory.create();

    //回调方法
    Callback<List<NMessage>> callback = new AsyncCallback(this);

    //像消息通道注册消息处理观察者
    BatchSubscriber<NMessage> batchSubscriber = new BatchSubscriber<NMessage>(this.fiberThread, callback, 30, TimeUnit.MILLISECONDS);
    this.messageChannel.subscribe(this.fiberThread, batchSubscriber);
}
 

import org.jetlang.fibers.PoolFiberFactory; //导入依赖的package包/类
public InRamSim(final int peerSize) {
    this.peerSize = peerSize;
    this.fiberPool = new PoolFiberFactory(Executors.newCachedThreadPool());
    Random r = new Random();

    for (int i = 0; i < peerSize; i++) {
        peerUUIDs.add((long)r.nextInt());
    }

    long plusMillis = 0;
    for( long peerId : peerUUIDs) {
        // make me a ....
        FleaseLease fl = new FleaseLease(fiberPool.create(), new Info(plusMillis), ""+peerId, "lease", peerId, peerUUIDs, rpcChannel);
        fleaseRunners.put(peerId, fl);
        plusMillis += 500;
    }

    rpcFiber = fiberPool.create();


    // subscribe to the rpcChannel:
    rpcChannel.subscribe(rpcFiber, new Callback<Request<OutgoingRpcRequest, IncomingRpcReply>>() {
        @Override
        public void onMessage(Request<OutgoingRpcRequest, IncomingRpcReply> message) {
            messageForwarder(message);
        }
    });

    rpcFiber.start();
}
 

import org.jetlang.fibers.PoolFiberFactory; //导入依赖的package包/类
/** Setup all "services" for all test methods.
 * @throws Exception */
@Before
public void setup() throws Exception {
    if (testDirect || testJetLang || Benchmark10MTest.testExecutorService) {
        executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
    }
    if (Benchmark10MTest.testAkkaBlocking
            || Benchmark10MTest.testAkkaNonBlocking) {
        final ConfigValue num = ConfigValueFactory
                .fromAnyRef(THREAD_POOL_SIZE);
        final Config config = ConfigFactory
                .load()
                .withValue(
                        "akka.actor.default-dispatcher.fork-join-executor.parallelism-max",
                        num)
                .withValue(
                        "akka.actor.default-dispatcher.thread-pool-executor.core-pool-size-max",
                        num)
                .withValue(
                        "akka.actor.default-dispatcher.thread-pool-executor.max-pool-size-max",
                        num);
        system = ActorSystem.create("AkkaTest", config);
    }
    if (testJActorIterator || testJActorStackOverflow
            || Benchmark10MTest.testJActorBlocking) {
        jaMailboxFactory = JAMailboxFactory
                .newMailboxFactory(THREAD_POOL_SIZE);
    }
    if (testJetLang) {
        fiberPool = new PoolFiberFactory(executorService);
    }
    if (testJActor2Local || testJActor2SharedNonBlocking
            || Benchmark10MTest.testJActor2NonBlocking
            || Benchmark10MTest.testJActor2Isolation) {
        new IsolationFriendlyPlantMtImpl(THREAD_POOL_SIZE);
    }
}
 

import org.jetlang.fibers.PoolFiberFactory; //导入依赖的package包/类
@Override
protected void configure() {
  install(new StateModule(timeout));

  PoolFiberFactory fiberFactory = new PoolFiberFactory(executor);

  Fiber raftFiber = fiberFactory.create(new BatchExecutor());
  raftFiber.start();
  bind(Fiber.class).annotatedWith(RaftExecutor.class).toInstance(raftFiber);

  Fiber stateMachineFiber = fiberFactory.create(new BatchExecutor());
  stateMachineFiber.start();

  install(new LogModule(logDir, stateMachine, stateMachineFiber));

  bind(ClusterConfig.class).toInstance(config);

  bind(Client.class).asEagerSingleton();

  expose(Raft.class);
  expose(ClusterConfig.class);
}