在Java中实现协同程序

我想看看这个: http://www.chiark.greenend.org.uk/~sgtatham/coroutines.html 很有意思,应该是个好的起点。当然,我们使用Java,所以我们可以做得更好(或者更糟,因为没有宏:)

根据我对协同训练的理解,你通常有 生产者 和A 消费者 协程(或者至少这是最常见的模式)。但是从语义上讲,你不希望生产者给消费者打电话,或者反之亦然,因为这会导致不对称。但是考虑到基于堆栈的语言的工作方式,我们需要有人来进行调用。

下面是一个非常简单的类型层次结构:

public interface CoroutineProducer<T>
{
    public T Produce();
    public boolean isDone();
}

public interface CoroutineConsumer<T>
{
    public void Consume(T t);
}

public class CoroutineManager
{
    public static Execute<T>(CoroutineProducer<T> prod, CoroutineConsumer<T> con)
    {
        while(!prod.IsDone()) // really simple
        {
            T d = prod.Produce();
            con.Consume(d);
        }
    }
}

当然,最困难的部分是 实施 接口,尤其是很难将计算分解为单独的步骤。为此,你可能需要一整套其他的 持久控制结构 . 基本的想法是,我们要模拟非本地控制转移(最终,它有点像我们在模拟 goto )我们基本上不想使用堆栈和 pc (程序计数器)将当前操作的状态保留在堆中而不是堆栈上。因此,我们需要一些助手类。

例如:

假设在一个理想的世界中,您希望编写一个类似于此的消费者(psuedocode):

boolean is_done;
int other_state;
while(!is_done)
{
    //read input
    //parse input
    //yield input to coroutine
    //update is_done and other_state;
}

我们需要像 is_done 和 other_state 我们需要抽象while循环本身,因为 yield like操作不会使用堆栈。因此,让我们创建一个while循环抽象和相关类:

enum WhileState {BREAK, CONTINUE, YIELD}
abstract class WhileLoop<T>
{
    private boolean is_done;
    public boolean isDone() { return is_done;}
    private T rval;
    public T getReturnValue() {return rval;} 
    protected void setReturnValue(T val)
    {
        rval = val;
    }


    public T loop()
    {
        while(true)
        {
            WhileState state = execute();
            if(state == WhileState.YIELD)
                return getReturnValue();
            else if(state == WhileState.BREAK)
                    {
                       is_done = true;
                return null;
                    }
        }
    }
    protected abstract WhileState execute();
}

这里的基本技巧是移动 地方的 变量是 班 变量并将作用域块转换为类,从而使我们能够在生成返回值后“重新进入”我们的“循环”。

现在实施我们的生产商

public class SampleProducer : CoroutineProducer<Object>
{
    private WhileLoop<Object> loop;//our control structures become state!!
    public SampleProducer()
    {
        loop = new WhileLoop()
        {
            private int other_state;//our local variables become state of the control structure
            protected WhileState execute() 
            {
                //this implements a single iteration of the loop
                if(is_done) return WhileState.BREAK;
                //read input
                //parse input
                Object calcluated_value = ...;
                //update is_done, figure out if we want to continue
                setReturnValue(calculated_value);
                return WhileState.YIELD;
            }
        };
    }
    public Object Produce()
    {
        Object val = loop.loop();
        return val;
    }
    public boolean isDone()
    {
        //we are done when the loop has exited
        return loop.isDone();
    }
}

对于其他基本的控制流结构也可以使用类似的技巧。理想情况下,您将构建这些助手类的库,然后使用它们来实现这些简单的接口,这些接口最终将为您提供共同例程的语义。我相信我在这里写的所有东西都可以被概括和扩展。

我建议你看看 Kotlin coroutines on JVM . 不过,它属于不同的类别。不涉及字节码操作,它也适用于Android。但是,你必须用考特林写你的校外活动。好处是,Kotlin设计的是与Java的互操作性,因此您仍然可以继续使用所有Java库,并在同一个项目中自由组合Kotlin和Java代码,甚至将它们并排放置在同一目录和包中。

这个 Guide to kotlinx.coroutines 提供了更多的示例, the coroutines design 文档解释了所有的动机、用例和实现细节。

我刚刚遇到这个问题,只是想提一下,我认为可以用C的类似方式实现协程或生成器。也就是说,我实际上并不使用Java,但是CIL与JVM有着非常相似的局限性。

这个 yield statement 在C中是纯语言特性,不是CIL字节码的一部分。C编译器只是为每个生成器函数创建一个隐藏的私有类。如果在函数中使用yield语句,它必须返回IEnumerator或IEnumerable。编译器将您的代码“打包”到一个类似状态机的类中。

C编译器可能会在生成的代码中使用一些“goto”来简化到状态机的转换。我不知道Java字节码的能力,如果有一种简单的无条件跳转,但在“汇编级”,通常是可能的。

如前所述,此功能必须在编译器中实现。因为我对Java和编译器的知识很少,所以我无法判断是否有可能用一个“预处理器”或其他东西来改变/扩展编译器。

我个人喜欢协同训练。作为一个Unity游戏开发者,我经常使用它们。因为我和ComputerCraft一起玩很多Minecraft,我很好奇为什么Lua(Luaj)中的协程是用线程实现的。

Kotlin对共同程序使用以下方法
(从 https://kotlinlang.org/docs/reference/coroutines.html ):

协程是通过编译技术完全实现的(不需要来自VM或OS端的支持),而挂起通过代码转换工作。基本上,每个挂起函数(可能会应用优化,但在这里我们不讨论)都会转换为状态机,其中状态对应于挂起的调用。在暂停之前,下一个状态存储在编译器生成的类以及相关局部变量等的字段中。恢复协同程序后,局部变量将被恢复,状态机将在暂停之后立即从状态中恢复。

一个悬挂的协同程序可以作为一个保持其悬挂状态和局部状态的对象来存储和传递。这些对象的类型是continue,这里描述的整个代码转换对应于经典的continue传递样式。因此,挂起函数在引擎盖下采用了一个额外的类型延续参数。

查看设计文档 https://github.com/Kotlin/kotlin-coroutines/blob/master/kotlin-coroutines-informal.md

我有一个在Java中使用的协同程序类。它是基于线程的,使用线程具有允许并行操作的优点,在多核机器上这是一种优势。因此,您可能需要考虑一种基于线程的方法。

Java6还有另一个选择+

一个pythonic coroutine实现:

import java.lang.ref.WeakReference;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

class CorRunRAII {
    private final List<WeakReference<? extends CorRun>> resources = new ArrayList<>();

    public CorRunRAII add(CorRun resource) {
        if (resource == null) {
            return this;
        }
        resources.add(new WeakReference<>(resource));

        return this;
    }

    public CorRunRAII addAll(List<? extends CorRun> arrayList) {
        if (arrayList == null) {
            return this;
        }
        for (CorRun corRun : arrayList) {
            add(corRun);
        }

        return this;
    }

    @Override
    protected void finalize() throws Throwable {
        super.finalize();

        for (WeakReference<? extends CorRun> corRunWeakReference : resources) {
            CorRun corRun = corRunWeakReference.get();
            if (corRun != null) {
                corRun.stop();
            }
        }
    }
}

class CorRunYieldReturn<ReceiveType, YieldReturnType> {
    public final AtomicReference<ReceiveType> receiveValue;
    public final LinkedBlockingDeque<AtomicReference<YieldReturnType>> yieldReturnValue;

    CorRunYieldReturn(AtomicReference<ReceiveType> receiveValue, LinkedBlockingDeque<AtomicReference<YieldReturnType>> yieldReturnValue) {
        this.receiveValue = receiveValue;
        this.yieldReturnValue = yieldReturnValue;
    }
}

interface CorRun<ReceiveType, YieldReturnType> extends Runnable, Callable<YieldReturnType> {
    boolean start();
    void stop();
    void stop(final Throwable throwable);
    boolean isStarted();
    boolean isEnded();
    Throwable getError();

    ReceiveType getReceiveValue();
    void setResultForOuter(YieldReturnType resultForOuter);
    YieldReturnType getResultForOuter();

    YieldReturnType receive(ReceiveType value);
    ReceiveType yield();
    ReceiveType yield(YieldReturnType value);
    <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(final CorRun<TargetReceiveType, TargetYieldReturnType> another);
    <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(final CorRun<TargetReceiveType, TargetYieldReturnType> another, final TargetReceiveType value);
}

abstract class CorRunSync<ReceiveType, YieldReturnType> implements CorRun<ReceiveType, YieldReturnType> {

    private ReceiveType receiveValue;
    public final List<WeakReference<CorRun>> potentialChildrenCoroutineList = new ArrayList<>();

    // Outside

    private AtomicBoolean isStarted = new AtomicBoolean(false);
    private AtomicBoolean isEnded = new AtomicBoolean(false);
    private Throwable error;

    private YieldReturnType resultForOuter;

    @Override
    public boolean start() {

        boolean isStarted = this.isStarted.getAndSet(true);
        if ((! isStarted)
                && (! isEnded())) {
            receive(null);
        }

        return isStarted;
    }

    @Override
    public void stop() {
        stop(null);
    }

    @Override
    public void stop(Throwable throwable) {
        isEnded.set(true);
        if (throwable != null) {
            error = throwable;
        }

        for (WeakReference<CorRun> weakReference : potentialChildrenCoroutineList) {
            CorRun child = weakReference.get();
            if (child != null) {
                child.stop();
            }
        }
    }

    @Override
    public boolean isStarted() {
        return isStarted.get();
    }

    @Override
    public boolean isEnded() {
        return isEnded.get();
    }

    @Override
    public Throwable getError() {
        return error;
    }

    @Override
    public ReceiveType getReceiveValue() {
        return receiveValue;
    }

    @Override
    public void setResultForOuter(YieldReturnType resultForOuter) {
        this.resultForOuter = resultForOuter;
    }

    @Override
    public YieldReturnType getResultForOuter() {
        return resultForOuter;
    }

    @Override
    public synchronized YieldReturnType receive(ReceiveType value) {
        receiveValue = value;

        run();

        return getResultForOuter();
    }

    @Override
    public ReceiveType yield() {
        return yield(null);
    }

    @Override
    public ReceiveType yield(YieldReturnType value) {
        resultForOuter = value;
        return receiveValue;
    }

    @Override
    public <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(CorRun<TargetReceiveType, TargetYieldReturnType> another) {
        return yieldFrom(another, null);
    }

    @Override
    public <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(CorRun<TargetReceiveType, TargetYieldReturnType> another, TargetReceiveType value) {
        if (another == null || another.isEnded()) {
            throw new RuntimeException("Call null or isEnded coroutine");
        }

        potentialChildrenCoroutineList.add(new WeakReference<CorRun>(another));

        synchronized (another) {
            boolean isStarted = another.start();
            boolean isJustStarting = ! isStarted;
            if (isJustStarting && another instanceof CorRunSync) {
                return another.getResultForOuter();
            }

            return another.receive(value);
        }
    }

    @Override
    public void run() {
        try {
            this.call();
        }
        catch (Exception e) {
            e.printStackTrace();

            stop(e);
            return;
        }
    }
}

abstract class CorRunThread<ReceiveType, YieldReturnType> implements CorRun<ReceiveType, YieldReturnType> {

    private final ExecutorService childExecutorService = newExecutorService();
    private ExecutorService executingOnExecutorService;

    private static final CorRunYieldReturn DUMMY_COR_RUN_YIELD_RETURN = new CorRunYieldReturn(new AtomicReference<>(null), new LinkedBlockingDeque<AtomicReference>());

    private final CorRun<ReceiveType, YieldReturnType> self;
    public final List<WeakReference<CorRun>> potentialChildrenCoroutineList;
    private CorRunYieldReturn<ReceiveType, YieldReturnType> lastCorRunYieldReturn;

    private final LinkedBlockingDeque<CorRunYieldReturn<ReceiveType, YieldReturnType>> receiveQueue;

    // Outside

    private AtomicBoolean isStarted = new AtomicBoolean(false);
    private AtomicBoolean isEnded = new AtomicBoolean(false);
    private Future<YieldReturnType> future;
    private Throwable error;

    private final AtomicReference<YieldReturnType> resultForOuter = new AtomicReference<>();

    CorRunThread() {
        executingOnExecutorService = childExecutorService;

        receiveQueue = new LinkedBlockingDeque<>();
        potentialChildrenCoroutineList = new ArrayList<>();

        self = this;
    }

    @Override
    public void run() {
        try {
            self.call();
        }
        catch (Exception e) {
            stop(e);
            return;
        }

        stop();
    }

    @Override
    public abstract YieldReturnType call();

    @Override
    public boolean start() {
        return start(childExecutorService);
    }

    protected boolean start(ExecutorService executorService) {
        boolean isStarted = this.isStarted.getAndSet(true);
        if (!isStarted) {
            executingOnExecutorService = executorService;
            future = (Future<YieldReturnType>) executingOnExecutorService.submit((Runnable) self);
        }
        return isStarted;
    }

    @Override
    public void stop() {
        stop(null);
    }

    @Override
    public void stop(final Throwable throwable) {
        if (throwable != null) {
            error = throwable;
        }
        isEnded.set(true);

        returnYieldValue(null);
        // Do this for making sure the coroutine has checked isEnd() after getting a dummy value
        receiveQueue.offer(DUMMY_COR_RUN_YIELD_RETURN);

        for (WeakReference<CorRun> weakReference : potentialChildrenCoroutineList) {
            CorRun child = weakReference.get();
            if (child != null) {
                if (child instanceof CorRunThread) {
                    ((CorRunThread)child).tryStop(childExecutorService);
                }
            }
        }

        childExecutorService.shutdownNow();
    }

    protected void tryStop(ExecutorService executorService) {
        if (this.executingOnExecutorService == executorService) {
            stop();
        }
    }

    @Override
    public boolean isEnded() {
        return isEnded.get() || (
                future != null && (future.isCancelled() || future.isDone())
                );
    }

    @Override
    public boolean isStarted() {
        return isStarted.get();
    }

    public Future<YieldReturnType> getFuture() {
        return future;
    }

    @Override
    public Throwable getError() {
        return error;
    }

    @Override
    public void setResultForOuter(YieldReturnType resultForOuter) {
        this.resultForOuter.set(resultForOuter);
    }

    @Override
    public YieldReturnType getResultForOuter() {
        return this.resultForOuter.get();
    }

    @Override
    public YieldReturnType receive(ReceiveType value) {

        LinkedBlockingDeque<AtomicReference<YieldReturnType>> yieldReturnValue = new LinkedBlockingDeque<>();

        offerReceiveValue(value, yieldReturnValue);

        try {
            AtomicReference<YieldReturnType> takeValue = yieldReturnValue.take();
            return takeValue == null ? null : takeValue.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return null;
    }

    @Override
    public ReceiveType yield() {
        return yield(null);
    }

    @Override
    public ReceiveType yield(final YieldReturnType value) {
        returnYieldValue(value);

        return getReceiveValue();
    }

    @Override
    public <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(final CorRun<TargetReceiveType, TargetYieldReturnType> another) {
        return yieldFrom(another, null);
    }

    @Override
    public <TargetReceiveType, TargetYieldReturnType> TargetYieldReturnType yieldFrom(final CorRun<TargetReceiveType, TargetYieldReturnType> another, final TargetReceiveType value) {
        if (another == null || another.isEnded()) {
            throw new RuntimeException("Call null or isEnded coroutine");
        }

        boolean isStarted = false;
        potentialChildrenCoroutineList.add(new WeakReference<CorRun>(another));

        synchronized (another) {
            if (another instanceof CorRunThread) {
                isStarted = ((CorRunThread)another).start(childExecutorService);
            }
            else {
                isStarted = another.start();
            }

            boolean isJustStarting = ! isStarted;
            if (isJustStarting && another instanceof CorRunSync) {
                return another.getResultForOuter();
            }

            TargetYieldReturnType send = another.receive(value);
            return send;
        }
    }

    @Override
    public ReceiveType getReceiveValue() {

        setLastCorRunYieldReturn(takeLastCorRunYieldReturn());

        return lastCorRunYieldReturn.receiveValue.get();
    }

    protected void returnYieldValue(final YieldReturnType value) {
        CorRunYieldReturn<ReceiveType, YieldReturnType> corRunYieldReturn = lastCorRunYieldReturn;
        if (corRunYieldReturn != null) {
            corRunYieldReturn.yieldReturnValue.offer(new AtomicReference<>(value));
        }
    }

    protected void offerReceiveValue(final ReceiveType value, LinkedBlockingDeque<AtomicReference<YieldReturnType>> yieldReturnValue) {
        receiveQueue.offer(new CorRunYieldReturn(new AtomicReference<>(value), yieldReturnValue));
    }

    protected CorRunYieldReturn<ReceiveType, YieldReturnType> takeLastCorRunYieldReturn() {
        try {
            return receiveQueue.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return null;
    }

    protected void setLastCorRunYieldReturn(CorRunYieldReturn<ReceiveType,YieldReturnType> lastCorRunYieldReturn) {
        this.lastCorRunYieldReturn = lastCorRunYieldReturn;
    }

    protected ExecutorService newExecutorService() {
        return Executors.newCachedThreadPool(getThreadFactory());
    }

    protected ThreadFactory getThreadFactory() {
        return new ThreadFactory() {
            @Override
            public Thread newThread(final Runnable runnable) {
                Thread thread = new Thread(runnable);
                thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                    @Override
                    public void uncaughtException(Thread thread, Throwable throwable) {
                        throwable.printStackTrace();
                        if (runnable instanceof CorRun) {
                            CorRun self = (CorRun) runnable;
                            self.stop(throwable);
                            thread.interrupt();
                        }
                    }
                });
                return thread;
            }
        };
    }
}

现在你可以用这种方式使用蟒蛇式的冠状体。 (例如斐波那契数)

线程版本:

class Fib extends CorRunThread<Integer, Integer> {

    @Override
    public Integer call() {
        Integer times = getReceiveValue();
        do {
            int a = 1, b = 1;
            for (int i = 0; times != null && i < times; i++) {
                int temp = a + b;
                a = b;
                b = temp;
            }
            // A pythonic "yield", i.e., it returns `a` to the caller and waits `times` value from the next caller
            times = yield(a);
        } while (! isEnded());

        setResultForOuter(Integer.MAX_VALUE);
        return getResultForOuter();
    }
}

class MainRun extends CorRunThread<String, String> {

    @Override
    public String call() {

        // The fib coroutine would be recycled by its parent
        // (no requirement to call its start() and stop() manually)
        // Otherwise, if you want to share its instance and start/stop it manually,
        // please start it before being called by yieldFrom() and stop it in the end.
        Fib fib = new Fib();
        String result = "";
        Integer current;
        int times = 10;
        for (int i = 0; i < times; i++) {

            // A pythonic "yield from", i.e., it calls fib with `i` parameter and waits for returned value as `current`
            current = yieldFrom(fib, i);

            if (fib.getError() != null) {
                throw new RuntimeException(fib.getError());
            }

            if (current == null) {
                continue;
            }

            if (i > 0) {
                result += ",";
            }
            result += current;

        }

        setResultForOuter(result);

        return result;
    }
}

同步(非线程)版本:

class Fib extends CorRunSync<Integer, Integer> {

    @Override
    public Integer call() {
        Integer times = getReceiveValue();

        int a = 1, b = 1;
        for (int i = 0; times != null && i < times; i++) {
            int temp = a + b;
            a = b;
            b = temp;
        }
        yield(a);

        return getResultForOuter();
    }
}

class MainRun extends CorRunSync<String, String> {

    @Override
    public String call() {

        CorRun<Integer, Integer> fib = null;
        try {
            fib = new Fib();
        } catch (Exception e) {
            e.printStackTrace();
        }

        String result = "";
        Integer current;
        int times = 10;
        for (int i = 0; i < times; i++) {

            current = yieldFrom(fib, i);

            if (fib.getError() != null) {
                throw new RuntimeException(fib.getError());
            }

            if (current == null) {
                continue;
            }

            if (i > 0) {
                result += ",";
            }
            result += current;
        }

        stop();
        setResultForOuter(result);

        if (Utils.isEmpty(result)) {
            throw new RuntimeException("Error");
        }

        return result;
    }
}

执行(两个版本都有效):

// Run the entry coroutine
MainRun mainRun = new MainRun();
mainRun.start();

// Wait for mainRun ending for 5 seconds
long startTimestamp = System.currentTimeMillis();
while(!mainRun.isEnded()) {
    if (System.currentTimeMillis() - startTimestamp > TimeUnit.SECONDS.toMillis(5)) {
        throw new RuntimeException("Wait too much time");
    }
}
// The result should be "1,1,2,3,5,8,13,21,34,55"
System.out.println(mainRun.getResultForOuter());