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

Question

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

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

这是我所知道的(或理解):

可以设置自定义哈希算法DefaultConnectionFactory,这有助于我们使用Consistent Hashing甚至使用内置的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(最终的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());
  }