简化 Java 中自定义迭代器的编写

Question

在 Java 中为自定义集合编写迭代器非常复杂，因为您实际上必须编写一个状态机，而不是编写提供一个又一个元素的直接代码：

public class CustomCollection<T> implements Iterable<T>
{
    private T[] data;
    private int size;

    @Override
    public Iterator<T> iterator()
    {
        return new Iterator<T>()
        {
            private int cursor = 0;

            @Override
            public boolean hasNext()
            {
                return cursor < size;
            }

            @Override
            public T next()
            {
                return data[cursor++];
            }

            @Override
            public void remove()
            {
                throw new UnsupportedOperationException();
            }
        };
    }
    // ...
}

对于比数组列表或链表更复杂的集合，获取正确地处理这些状态机是一项艰巨的任务。事实上，C# 设计团队认为编写自定义迭代器足够复杂，需要引入特殊语言支持 (yield return) 来让编译器构建状态机。

Java 的下一版本中是否会出现类似 yield return 的内容？或者是否有任何库解决方案可以让我在用 Java 编写自己的迭代器时变得更轻松？

Answer 1

不，Java 没有像 yield 这样的东西。就库而言，Guava 有许多有用的类，可以使某些类型的迭代器易于编写：

• < code>AbstractIterator 只需要您实现一个 TcomputeNext() 方法。
• AbstractLinkedIterator 要求您实现 TcomputeNext(T previous)。

AbstractIterator 可以用于此目的，如下所示：

return new AbstractIterator<T>() {
  private int index = 0;

  protected T computeNext() {
    return index == size ? endOfData() : data[index++];
  }
};

您也可以按照 Amir 的建议使用 Arrays.asList，甚至可以执行如下操作：

private final List<T> listView = new AbstractList<T>() {
  public int size() {
    return data.length;
  }

  public T get(int index) {
    return data[index];
  }
};

public Iterator<T> iterator() {
  return listView.iterator();
}

Answer 2

也许我只是不明白你的问题。你能不能return Arrays.asList(data).iterator()

Answer 3

如果您不介意启动一个新线程，可以使用 同步队列

public class InternalIterator<T> implements Iterator<T>{

    private SynchronousQueue<T> queue = new SynchronousQueue<T>();
    private volatile boolean empty = false;
    private T current =null;
    private Object lock = new Object();

    private Runner implements Runnable{//run in deamon

        public void run(){
            //iterate and call 
            synchronized()(lock){
                try{
                    queue.offer(t);
                    lock.wait();
                }catch(InteruptedException e){
                    empty=true;
                    throw new RuntimeException(e);
                }
            } 
            //for each element to insert this will be the yield return 

            emtpy=true;
        }

    }

    public boolean hasNext(){
        if(current!=null)return true;

        while(!empty){
            if( (current=queue.poll(100,TimeUnit.MILLISECONDS))!=null){//avoid deadlock when last element is already returned but empty wasn't written yet 
                return true;
            }
        }

        return false;
    }

    public boolean next(){
        if(!hasNext())throw new NoSuchElementException();
        T tmp = current;
        current=null;
        return tmp;
    }

    public void remove(){
        throw new UnsupportedOperationException();
    }


}

Answer 4

Java 始终提供一种机制来维护状态并在稍后的某个时间点继续执行：线程。我的库解决方案的基本思想是让 ConcurrentIterable 在一个线程中生成元素，并让 ConcurrentIterator 消耗元素em> 他们在另一个中，通过有界队列进行通信。 （这通常称为生产者/消费者模式。）

首先，这里是简化用法的演示：

public class CustomCollection<T> extends ConcurrentIterable<T>
{
    private T[] data;
    private int size;

    @Override
    protected void provideElements()
    {
        for (int i = 0; i < size; ++i)
        {
            provideElement(data[i]);
        }
    }
    // ...
}

请注意完全没有状态机。您所要做的就是从 ConcurrentIterable 派生并实现 provideElements 方法。在此方法中，您可以编写直接的代码，为集合中的每个元素调用 provideElement。

有时，客户端不会迭代整个集合，例如在线性搜索中。一旦检测到中止，您就可以通过检查 iterationAborted() 来停止提供元素：

    @Override
    protected void provideElements()
    {
        for (int i = 0; i < size && !iterationAborted(); ++i)
        {
            provideElement(data[i]);
        }
    }

只要您不关心，不检查 iterationAborted() 就完全可以了关于正在生成的附加元素。对于无限序列，必须检查iterationAborted()。

生产者如何检测消费者已经停止迭代？这是通过在消费者中对令牌进行强引用并在生产者中对同一令牌进行弱引用来实现的。当消费者停止迭代时，令牌就有资格进行垃圾回收，并且最终对生产者来说将变得不可见。从那时起，所有新元素都将被丢弃。

（如果没有这种预防措施，在某些情况下，有界队列最终可能会被填满，生产者将进入无限循环，并且所包含的元素永远不会被垃圾收集。）

现在介绍实现细节：

ConcurrentIterable.java

import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public abstract class ConcurrentIterable<T> implements Iterable<T>
{
    private static final int CAP = 1000;
    private final ThreadLocal<CommunicationChannel<T>> channels
    = new ThreadLocal<CommunicationChannel<T>>();

    @Override
    public Iterator<T> iterator()
    {
        BlockingQueue<Option<T>> queue = new ArrayBlockingQueue<Option<T>>(CAP);
        Object token = new Object();
        final CommunicationChannel<T> channel
        = new CommunicationChannel<T>(queue, token);
        new Thread(new Runnable()
        {
            @Override
            public void run()
            {
                channels.set(channel);
                provideElements();
                enqueueSentinel();
            }
        }).start();
        return new ConcurrentIterator<T>(queue, token);
    }

    protected abstract void provideElements();

    protected final boolean iterationAborted()
    {
        return channels.get().iterationAborted();
    }

    protected final void provideElement(T element)
    {
        enqueue(Option.some(element));
    }

    private void enqueueSentinel()
    {
        enqueue(Option.<T> none());
    }

    private void enqueue(Option<T> element)
    {
        try
        {
            while (!offer(element))
            {
                System.gc();
            }
        }
        catch (InterruptedException ignore)
        {
            ignore.printStackTrace();
        }
    }

    private boolean offer(Option<T> element) throws InterruptedException
    {
        CommunicationChannel<T> channel = channels.get();
        return channel.iterationAborted()
            || channel.queue.offer(element, 1, TimeUnit.SECONDS);
    }
}

CommunicationChannel.java

import java.lang.ref.WeakReference;
import java.util.concurrent.BlockingQueue;

public class CommunicationChannel<T>
{
    public final BlockingQueue<Option<T>> queue;
    private final WeakReference<Object> token;

    public CommunicationChannel(BlockingQueue<Option<T>> queue, Object token)
    {
        this.queue = queue;
        this.token = new WeakReference<Object>(token);
    }

    public boolean iterationAborted()
    {
        return token.get() == null;
    }
}

ConcurrentIterator。 java

import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.concurrent.BlockingQueue;

public class ConcurrentIterator<T> implements Iterator<T>
{
    private final BlockingQueue<Option<T>> queue;
    @SuppressWarnings("unused")
    private final Object token;
    private Option<T> next;

    public ConcurrentIterator(BlockingQueue<Option<T>> queue, Object token)
    {
        this.queue = queue;
        this.token = token;
    }

    @Override
    public boolean hasNext()
    {
        if (next == null)
        {
            try
            {
                next = queue.take();
            }
            catch (InterruptedException ignore)
            {
                ignore.printStackTrace();
            }
        }
        return next.present;
    }

    @Override
    public T next()
    {
        if (!hasNext()) throw new NoSuchElementException();
        T result = next.value;
        next = null;
        return result;
    }

    @Override
    public void remove()
    {
        throw new UnsupportedOperationException();
    }
}

Option.java

public class Option<T>
{
    public final T value;
    public final boolean present;

    private Option(T value, boolean present)
    {
        this.value = value;
        this.present = present;
    }

    public static <T> Option<T> some(T value)
    {
        return new Option<T>(value, true);
    }

    @SuppressWarnings("unchecked")
    public static <T> Option<T> none()
    {
        return none;
    }

    @SuppressWarnings({ "rawtypes", "unchecked" })
    private static final Option none = new Option(null, false);
}

让我知道你的想法！