@Override
  public <T> BulkFuture<Map<String, T>> asyncGetBulk(Iterator<String> keyIter,
      Iterator<Transcoder<T>> tcIter) {
    final Map<String, Future<T>> m = new ConcurrentHashMap<String, Future<T>>();

    // This map does not need to be a ConcurrentHashMap
    // because it is fully populated when it is used and
    // used only to read the transcoder for a key.
    final Map<String, Transcoder<T>> tcMap =
        new HashMap<String, Transcoder<T>>();

    // Break the gets down into groups by key
    final Map<MemcachedNode, Collection<String>> chunks =
        new HashMap<MemcachedNode, Collection<String>>();
    final NodeLocator locator = mconn.getLocator();

    while (keyIter.hasNext() && tcIter.hasNext()) {
      String key = keyIter.next();
      tcMap.put(key, tcIter.next());
      StringUtils.validateKey(key, opFact instanceof BinaryOperationFactory);
      final MemcachedNode primaryNode = locator.getPrimary(key);
      MemcachedNode node = null;
      if (primaryNode.isActive()) {
        node = primaryNode;
      } else {
        for (Iterator<MemcachedNode> i = locator.getSequence(key); node == null
            && i.hasNext();) {
          MemcachedNode n = i.next();
          if (n.isActive()) {
            node = n;
          }
        }
        if (node == null) {
          node = primaryNode;
        }
      }
      assert node != null : "Didn't find a node for " + key;
      Collection<String> ks = chunks.get(node);
      if (ks == null) {
        ks = new ArrayList<String>();
        chunks.put(node, ks);
      }
      ks.add(key);
    }

    final AtomicInteger pendingChunks = new AtomicInteger(chunks.size());
    int initialLatchCount = chunks.isEmpty() ? 0 : 1;
    final CountDownLatch latch = new CountDownLatch(initialLatchCount);
    final Collection<Operation> ops = new ArrayList<Operation>(chunks.size());
    final BulkGetFuture<T> rv = new BulkGetFuture<T>(m, ops, latch, executorService);

    GetOperation.Callback cb = new GetOperation.Callback() {
      @Override
      @SuppressWarnings("synthetic-access")
      public void receivedStatus(OperationStatus status) {
        if (status.getStatusCode() == StatusCode.ERR_NOT_MY_VBUCKET) {
          pendingChunks.addAndGet(Integer.parseInt(status.getMessage()));
        }
        rv.setStatus(status);
      }

      @Override
      public void gotData(String k, int flags, byte[] data) {
        Transcoder<T> tc = tcMap.get(k);
        m.put(k,
            tcService.decode(tc, new CachedData(flags, data, tc.getMaxSize())));
      }

      @Override
      public void complete() {
        if (pendingChunks.decrementAndGet() <= 0) {
          latch.countDown();
          rv.signalComplete();
        }
      }
    };

    // Now that we know how many servers it breaks down into, and the latch
    // is all set up, convert all of these strings collections to operations
    final Map<MemcachedNode, Operation> mops =
        new HashMap<MemcachedNode, Operation>();

    for (Map.Entry<MemcachedNode, Collection<String>> me : chunks.entrySet()) {
      Operation op = opFact.get(me.getValue(), cb);
      mops.put(me.getKey(), op);
      ops.add(op);
    }
    assert mops.size() == chunks.size();
    mconn.checkState();
    mconn.addOperations(mops);
    return rv;
  }
Asynchronously get a bunch of objects from the cache.

@param <T>
@param keyIter Iterator that produces keys.
@param tcIter an iterator of transcoders to serialize and unserialize
         values; the transcoders are matched with the keys in the same
         order. The minimum of the key collection length and number of
         transcoders is used and no exception is thrown if they do not
         match
@return a Future result of that fetch
@throws IllegalStateException in the rare circumstance where queue is too
          full to accept any more requests
src/main/java/net/spy/memcached/MemcachedClient.java