一致性hash实现

import java.security.MessageDigest

import scala.collection.mutable

/**
  * Created by liufengquan on 2018/4/13.
  */
class ConsistentHash(virtualNodes: Int) {
  val servers: mutable.Set[String] = mutable.Set()

  /** store virtual node's hash */
  private var hashRing = List[Int]()
  /** virtual node's hash -> node identity*/
  private var hashServerMap = mutable.Map[Int, String]()

  /**
    * put a node, that is a server in pool
    * @param ip
    */
  def addNode(ip: String): Unit = {
    servers += ip
    (1 to virtualNodes).foreach(i => {
      val key = getKeyHash(ip + ":" + i)
      hashRing = key :: hashRing
      hashServerMap += key -> ip
    })

    hashRing = hashRing.sorted
  }

  /**
    * remove a node
    * @param ip
    */
  def removeNode(ip: String): Unit = {
    servers -= ip
    (1 to virtualNodes).foreach(i => {
      val key = getKeyHash(ip + "i")
      hashRing = hashRing.filter(_ != key)
      hashServerMap = hashServerMap.filter(_._1 != key)
    })
  }

  /**
    * get a node according to a key
    * @param key
    * @return
    */
  def getNode(key: String): String = {
    require(servers != null && servers.nonEmpty)

    val hash = getKeyHash(key)
    val avail = hashRing.filter(_ > hash)
    if (avail.nonEmpty) hashServerMap(avail.head)
    else hashServerMap(hashRing.head)
  }

  def getServers: String = servers mkString ":"

  /**
    * naive hashCode has bad randomness,
    * use md5 and do a little exchange
    * @param key
    * @return
    */
  private def getKeyHash(key: String): Int = {
    val digest = MessageDigest.getInstance("MD5")

    val bytes = digest.digest(key.getBytes("UTF-8"))

    bytes.map(_.toInt).reduceLeft(_ * 11 + _)
  }
}

object ConsistentHash {
  val servers = List("10.10.187.1", "10.10.187.2", "10.10.187.3", "10.10.187.4", "10.10.187.5")

  def consistentHashTest(replica: Int): Unit = {
    val consistentHash = new ConsistentHash(replica)

    var map = mutable.Map[String, List[String]]()
    val text = """The following examples show some safe operations to drop or change columns. Dropping the final column in a table lets Impala ignore the data causing any disruption to existing data files. Changing the type of a column works if existing data values can be safely converted to the new type. The type conversion rules depend on the file format of the underlying table. For example, in a text table, the same value can be interpreted as a STRING or a numeric value, while in a binary format such as Parquet, the rules are stricter and type conversions only work between certain sizes of integers."""

    servers.foreach(consistentHash.addNode)

    servers.foreach(map += _ -> List[String]())

    text.split(" ").foreach(s => {
      val server = consistentHash.getNode(s)
      map(server) = s :: map(server)
    })

    println(map mkString "\n")
  }

  def main(args: Array[String]): Unit = {
    require(args.length > 0)
    consistentHashTest(args(0).toInt)
  }
}