如何使用 spymemcached 动态添加 memcached 节点

Question

我有一个 Java 应用程序设置，它有多个 memcached 服务器节点与 spymemcached 客户端通信。

我想知道是否可以在运行时添加或删除服务器节点，而不会干扰所有现有的缓存节点（我知道应该更改一些节点）。

这是我知道（或理解）的：

可以在DefaultConnectionFactory中设置自定义哈希算法，这有助于我们使用一致性哈希，甚至可以使用内置的KetamaConnectionFactory。

所以我们应该能够添加或删除仅对一个或几个现有节点进行更改的节点。

是否可以使用spymemcached？

如果是，那怎么办？

谁能指出我正确的方向？

Answer 1

看起来NodeLocator.updateLocator(List<MemcachedNode> newNodes) 应该可以完成这项工作。

但是连接 MemcachedNode 有点困难。您必须覆盖MemcachedClient、MemcachedConnection 和DefaultConnectionFactory。

您想在MemcachedClient 添加或删除客户端是合理的，因此您添加了remove(MemcachedNode node) 和add(MemcachedNode node) 方法。

如果要删除，您应该断开节点（参见MemcachedConnection.shutdown()）并将其从NodeLocator.getAll() 中删除并调用NodeLocator.updateLocator(List<MemcachedNode> newNodes)。

如果添加，您应该通过 MemcachedConnection.createConnections(final Collection a) 连接节点，将其与NodeLocator.getAll() 合并并调用NodeLocator.updateLocator(List<MemcachedNode> newNodes)。

嗯，我从来没有尝试过，所以它可能行不通。祝你好运！

ExtMemCachedConnection.java 公共类 ExtMemCachedConnection 扩展 MemcachedConnection {

  protected final OperationFactory opFact;

  /**
   * Construct a memcached connection.
   *
   * @param bufSize the size of the buffer used for reading from the server
   * @param f       the factory that will provide an operation queue
   * @param a       the addresses of the servers to connect to
   * @throws java.io.IOException if a connection attempt fails early
   */
  public ExtendableMemcachedConnection(int bufSize, ConnectionFactory f,
                                       List<InetSocketAddress> a,
                                       Collection<ConnectionObserver> obs,
                                       FailureMode fm, OperationFactory opfactory)
      throws IOException {
    super(bufSize, f, a, obs, fm, opfactory);
    this.opFact = opfactory;
  }

  public void add(InetSocketAddress nodeAddress) throws IOException {
    final List<InetSocketAddress> nodeToAdd = new ArrayList<InetSocketAddress>(1);
    nodeToAdd.add(nodeAddress);
    List<MemcachedNode> newNodesList = createConnections(nodeToAdd);
    newNodesList.addAll(getLocator().getAll());
    getLocator().updateLocator(newNodesList);
  }

  //The node should be obtain from locator to ensure currentNode.equals(node) will return true
  public void remove(MemcachedNode node) throws IOException {
    for(MemcachedNode currentNode : getLocator().getAll()) {
      if(currentNode.equals(node)) {
        Collection<Operation> notCompletedOperations = currentNode.destroyInputQueue();
        if (currentNode.getChannel() != null) {
          currentNode.getChannel().close();
          currentNode.setSk(null);
          if (currentNode.getBytesRemainingToWrite() > 0) {
            getLogger().warn("Shut down with %d bytes remaining to write",
                             currentNode.getBytesRemainingToWrite());
          }
          getLogger().debug("Shut down channel %s", currentNode.getChannel());
        }
        //Unfortunatelly,  redistributeOperations is private so it cannot be used or override. I put copy/paste the implementation
        redistributeOperations(notCompletedOperations);
      }
    }
  }

  protected void redistributeOperations(Collection<Operation> ops) {
    for (Operation op : ops) {
      if (op.isCancelled() || op.isTimedOut()) {
        continue;
      }
      if (op instanceof KeyedOperation) {
        KeyedOperation ko = (KeyedOperation) op;
        int added = 0;
        for (String k : ko.getKeys()) {
          for (Operation newop : opFact.clone(ko)) {
            addOperation(k, newop);
            added++;
          }
        }
        assert added > 0 : "Didn't add any new operations when redistributing";
      } else {
        // Cancel things that don't have definite targets.
        op.cancel();
      }
    }
  }


}

ExtMemcachedClient.java

  public void add(InetSocketAddress nodeAddress) {
    if(mconn instanceof ExtMemcachedConnection) {
      ((ExtMemcachedConnection)mconn).add(nodeAddress);  
    }
  }

  public boolean remove(MemcachedNode node) {
    if(mconn instanceof ExtMemcachedConnection) {
      ((ExtMemcachedConnection)mconn).remove(nodeAddress);
    }
  }

ExtMemcachedConnectionfactory.java

  @Override
  public MemcachedConnection createConnection(List<InetSocketAddress> addrs) throws IOException {
    return new ExtendableMemcachedConnection(getReadBufSize(), this, addrs,
                                             getInitialObservers(), getFailureMode(), getOperationFactory());
  }

Answer 2

这个问题是很久以前针对 spymemcached 提出的，但我认为这个答案可能对其他寻找替代方案的人有所帮助。还有另一个名为 xmemcached 的 memcached 客户端。此客户端支持动态添加/删除 memcached 服务器。

https://code.google.com/p/xmemcached/

https://github.com/killme2008/xmemcached

memcachedClient.addServer(host, port);

memcachedClient.removeServer(host);