Использование MapMaker # makeComputingMap для предотвращения одновременных RPC для одних и тех же данных - PullRequest
Новая
       24

Использование MapMaker # makeComputingMap для предотвращения одновременных RPC для одних и тех же данных

2 голосов
/ 28 октября 2010

У нас медленный бэкэнд-сервер, который перегружается нагрузкой, и мы хотели бы, чтобы сервер Scala среднего уровня имел только один невыполненный запрос к бэкэнду для каждого уникального поиска.

Бэкэнд-сервер хранит только неизменяемые данные, но после добавления новых данных серверы среднего уровня будут запрашивать новейшие данные от имени клиентов, а бэкэнд-сервер испытывает трудности с загрузкой.Неизменяемые данные кэшируются в memcached с использованием уникальных ключей, сгенерированных при записи, но скорость записи высока, поэтому мы получаем низкую частоту обращений в memcached.

Одна из идей, которые у меня есть, - использовать Google Guava MapMaker # makeComputingMap ()чтобы обернуть фактический поиск и после возврата ConcurrentMap # get (), средний уровень сохранит результат и просто удалит ключ из карты.

Это кажется немного расточительным, хотя код очень простнапишите, см. ниже пример того, о чем я думаю.

Существует ли более естественная структура данных, библиотека или часть Guava, которая решит эту проблему? 

import com.google.common.collect.MapMaker

object Test
{
  val computer: com.google.common.base.Function[Int,Long] =
  {
    new com.google.common.base.Function[Int,Long] {
      override
      def apply(i: Int): Long =
      {
        val l = System.currentTimeMillis + i
        System.err.println("For " + i + " returning " + l)
        Thread.sleep(2000)
        l
      }
    }
  }

  val map =
  {
    new MapMaker().makeComputingMap[Int,Long](computer)
  }

  def get(k: Int): Long =
  {
    val l = map.get(k)
    map.remove(k)
    l
  }

  def main(args: Array[String]): Unit =
  {
    val t1 = new Thread() {
      override def run(): Unit =
      {
        System.err.println(get(123))
      }
    }

    val t2 = new Thread() {
      override def run(): Unit =
      {
        System.err.println(get(123))
      }
    }

    t1.start()
    t2.start()
    t1.join()
    t2.join()

    System.err.println(get(123))
  }
}

Ответы [ 2 ]

4 голосов
/ 28 октября 2010

Я не уверен, почему вы реализуете удаление самостоятельно, почему бы просто не иметь слабые или мягкие значения и позволить GC очистить для вас? 

new MapMaker().weakValues().makeComputingMap[Int, Long](computer)
2 голосов
/ 29 октября 2010

Я думаю, что вы делаете, вполне разумно. Вы только используете структуру, чтобы получить чередование блокировок на ключе, чтобы гарантировать, что доступ к тому же самому ключу конфликтует. Не беспокойтесь, что вам не нужно сопоставление значений для каждого ключа. ConcurrentHashMap и friends - единственная структура в библиотеках Java + Guava, которая предлагает вам чинить блокировки.

Это вызывает некоторые незначительные накладные расходы времени выполнения, плюс размер хеш-таблицы, который вам не нужен (который может даже возрасти, если доступ к одному и тому же сегменту накапливается и удаляется () не поддерживается).

Если вы хотите сделать его как можно более дешевым, вы можете самостоятельно написать код для снятия блокировки. В основном это Object[] (или Array[AnyRef] :)) из N блокировок (N = уровень параллелизма), и вы просто отображаете хэш ключа поиска в этот массив и блокируете. Еще одним преимуществом этого является то, что вам действительно не нужно делать трюки с хеш-кодом, которые требуются CHM, потому что последний должен разделить хеш-код на одну часть, чтобы выбрать блокировку, а другую - для нужд хеш-таблицы, но вы можете используйте все это только для выбора блокировки.

edit : зарисовка моего комментария ниже: 

val concurrencyLevel = 16
val locks = (for (i <- 0 to concurrencyLevel) yield new AnyRef).toArray

def access(key: K): V = {
   val lock = locks(key.hashCode % locks.size)
   lock synchronized {
     val valueFromCache = cache.lookup(key)
     valueFromCache match {
       case Some(v) => return v
       case None =>
         val valueFromBackend = backendServer.lookup(key)
         cache.put(key, valueFromBackend)
         return valueFromBackend
     }
   }
}

(Кстати, нужен ли вызов toArray? Или возвращенный IndexSeq уже быстро доступен по индексу?)

...