Projektowanie aplikacji z Redis jako magazynem danych. Co? Czemu?

1) Wprowadzenie

Witam wszystkich! Wiele osób wie, czym jest Redis, a jeśli nie wiesz, oficjalna strona może Cię poinformować.
Dla większości Redis to pamięć podręczna, a czasem kolejka wiadomości.
Ale co, jeśli trochę zwariujemy i spróbujemy zaprojektować całą aplikację, używając tylko Redis jako magazynu danych? Jakie zadania możemy rozwiązać z Redisem?
Postaramy się odpowiedzieć na te pytania w tym artykule.

Czego tu nie zobaczymy?

• Nie będzie tu żadnej szczegółowej struktury danych Redis. W jakim celu powinieneś przeczytać specjalne artykuły lub dokumentację.
• Tutaj również nie będzie kodu gotowego do produkcji, którego mógłbyś użyć w swojej pracy.

Co tu zobaczymy?

• Będziemy używać różnych struktur danych Redis do realizacji różnych zadań aplikacji randkowej.
• Oto przykłady kodu Kotlin + Spring Boot.

2) Naucz się tworzyć i sprawdzać profile użytkowników.

• Po pierwsze, nauczmy się tworzyć profile użytkowników z ich imionami, polubieniami itp.

  Aby to zrobić, potrzebujemy prostego magazynu klucz-wartość. Jak to zrobić?

• Po prostu. Redis ma strukturę danych - hash. Zasadniczo jest to po prostu znajoma mapa skrótów dla nas wszystkich.

Polecenia języka zapytań Redis można znaleźć tutaj i tutaj.
Dokumentacja zawiera nawet interaktywne okno do wykonywania tych poleceń bezpośrednio na stronie. A całą listę poleceń znajdziesz tutaj.
Podobne linki działają dla wszystkich kolejnych poleceń, które rozważymy.

W kodzie używamy RedisTemplate prawie wszędzie. To podstawa pracy z Redis w ekosystemie Spring.

Jedyną różnicą w stosunku do mapy jest to, że jako pierwszy argument przekazujemy "pole". „Pole” to nazwa naszego skrótu.

fun addUser(user: User) {
        val hashOps: HashOperations<String, String, User> = userRedisTemplate.opsForHash()
        hashOps.put(Constants.USERS, user.name, user)
    }

fun getUser(userId: String): User {
        val userOps: HashOperations<String, String, User> = userRedisTemplate.opsForHash()
        return userOps.get(Constants.USERS, userId)?: throw NotFoundException("Not found user by $userId")
    }

Powyżej znajduje się przykład tego, jak może wyglądać w Kotlinie przy użyciu bibliotek Spring.

Wszystkie fragmenty kodu z tego artykułu można znaleźć na Github.

3) Aktualizowanie polubień użytkowników za pomocą list Redis.

• Świetny!. Mamy użytkowników i informacje o polubieniach.

  Teraz powinniśmy znaleźć sposób, jak zaktualizować to polubienia.

  Zakładamy, że wydarzenia zdarzają się bardzo często. Użyjmy więc podejścia asynchronicznego z pewną kolejką. A informacje z kolejki odczytamy zgodnie z harmonogramem.

• Redis ma strukturę danych list z takim zestawem poleceń. Możesz używać list Redis zarówno jako kolejki FIFO, jak i stosu LIFO.

Na wiosnę używamy tego samego podejścia do pobierania ListOperations z RedisTemplate.

Musimy napisać w prawo. Ponieważ tutaj symulujemy kolejkę FIFO od prawej do lewej.

fun putUserLike(userFrom: String, userTo: String, like: Boolean) {
        val userLike = UserLike(userFrom, userTo, like)
        val listOps: ListOperations<String, UserLike> = userLikeRedisTemplate.opsForList()
        listOps.rightPush(Constants.USER_LIKES, userLike)
}

Teraz będziemy wykonywać naszą pracę zgodnie z harmonogramem.

Po prostu przenosimy informacje z jednej struktury danych Redis do drugiej. To nam wystarczy jako przykład.

fun processUserLikes() {
        val userLikes = getUserLikesLast(USERS_BATCH_LIMIT).filter{ it.isLike}
        userLikes.forEach{updateUserLike(it)}
}

Aktualizacja użytkowników jest tutaj naprawdę łatwa. Pozdrów HashOperation z poprzedniej części.

private fun updateUserLike(userLike: UserLike) {
        val userOps: HashOperations<String, String, User> = userLikeRedisTemplate.opsForHash()
        val fromUser = userOps.get(Constants.USERS, userLike.fromUserId)?: throw UserNotFoundException(userLike.fromUserId)
        fromUser.fromLikes.add(userLike)
        val toUser = userOps.get(Constants.USERS, userLike.toUserId)?: throw UserNotFoundException(userLike.toUserId)
        toUser.fromLikes.add(userLike)

        userOps.putAll(Constants.USERS, mapOf(userLike.fromUserId to fromUser, userLike.toUserId to toUser))
    }

A teraz pokazujemy, jak pobrać dane z listy. Dostajemy to z lewej strony. Aby pobrać kilka danych z listy, użyjemy range metoda.
I jest ważny punkt. Metoda zakresu pobierze tylko dane z listy, ale ich nie usunie.

Musimy więc użyć innej metody, aby usunąć dane. trim Zrób to. (I możesz mieć tam kilka pytań).

private fun getUserLikesLast(number: Long): List<UserLike> {
        val listOps: ListOperations<String, UserLike> = userLikeRedisTemplate.opsForList()
        return (listOps.range(Constants.USER_LIKES, 0, number)?:mutableListOf()).filterIsInstance(UserLike::class.java)
            .also{
listOps.trim(Constants.USER_LIKES, number, -1)
}
}

A pytania to:

• Jak pobrać dane z listy do kilku wątków?
• A jak zapewnić, że dane nie zostaną utracone w przypadku błędu? Od samego początku – nic. Musisz zebrać dane z listy w jednym wątku. I musisz sam radzić sobie ze wszystkimi niuansami, które pojawiają się.

4) Wysyłanie powiadomień push do użytkowników za pomocą pub/sub

• Idź dalej!
  Mamy już profile użytkowników. Wymyśliliśmy, jak obsłużyć strumień polubień od tych użytkowników.

  Ale wyobraź sobie przypadek, w którym chcesz wysłać powiadomienie push do użytkownika w momencie, gdy otrzymamy polubienie.
  Co zamierzasz zrobić?

• Mamy już asynchroniczny proces obsługi polubień, więc po prostu zbudujmy tam wysyłanie powiadomień push. Oczywiście do tego celu użyjemy WebSocket. I możemy po prostu wysłać go przez WebSocket, gdzie otrzymamy lajka. Ale co, jeśli chcemy wykonać długo działający kod przed wysłaniem? A co, jeśli chcemy delegować pracę z WebSocket do innego komponentu?
• Pobierzemy i przeniesiemy nasze dane ponownie z jednej struktury danych Redis (listy) do innej (pub/sub).

fun processUserLikes() {
        val userLikes = getUserLikesLast(USERS_BATCH_LIMIT).filter{ it.isLike}
                pushLikesToUsers(userLikes)
        userLikes.forEach{updateUserLike(it)}
}

private fun pushLikesToUsers(userLikes: List<UserLike>) {
  GlobalScope.launch(Dispatchers.IO){
        userLikes.forEach {
            pushProducer.publish(it)
        }
  }
}

@Component
class PushProducer(val redisTemplate: RedisTemplate<String, String>, val pushTopic: ChannelTopic, val objectMapper: ObjectMapper) {

    fun publish(userLike: UserLike) {
        redisTemplate.convertAndSend(pushTopic.topic, objectMapper.writeValueAsString(userLike))
    }
}

Powiązanie odbiornika z tematem znajduje się w konfiguracji.
Teraz możemy po prostu zabrać naszego słuchacza do osobnego serwisu.

@Component
class PushListener(val objectMapper: ObjectMapper): MessageListener {
    private val log = KotlinLogging.logger {}

    override fun onMessage(userLikeMessage: Message, pattern: ByteArray?) {
        // websocket functionality would be here
        log.info("Received: ${objectMapper.readValue(userLikeMessage.body, UserLike::class.java)}")
    }
}

5) Znajdowanie najbliższych użytkowników poprzez operacje geograficzne.

• Skończyliśmy z polubieniami. Ale co z możliwością znalezienia użytkowników znajdujących się najbliżej danego punktu.

• GeoOperations nam w tym pomoże. Będziemy przechowywać pary klucz-wartość, ale teraz naszą wartością jest współrzędna użytkownika. Aby znaleźć, użyjemy [radius](https://redis.io/commands/georadius) metoda. Przekazujemy identyfikator użytkownika, aby znaleźć i sam promień wyszukiwania.

Redis zwraca wynik zawierający nasz identyfikator użytkownika.

fun getNearUserIds(userId: String, distance: Double = 1000.0): List<String> {
    val geoOps: GeoOperations<String, String> = stringRedisTemplate.opsForGeo()
    return geoOps.radius(USER_GEO_POINT, userId, Distance(distance, RedisGeoCommands.DistanceUnit.KILOMETERS))
        ?.content?.map{ it.content.name}?.filter{ it!= userId}?:listOf()
}

6) Aktualizacja lokalizacji użytkowników za pośrednictwem strumieni

• Wdrożyliśmy prawie wszystko, czego potrzebujemy. Ale teraz znowu mamy sytuację, w której musimy zaktualizować dane, które mogą się szybko zmienić.

  Znowu musimy więc skorzystać z kolejki, ale fajnie byłoby mieć coś bardziej skalowalnego.

• Strumienie Redis mogą pomóc w rozwiązaniu tego problemu.
• Prawdopodobnie wiesz o Kafce i prawdopodobnie wiesz nawet o strumieniach Kafka, ale to nie to samo, co strumienie Redis. Ale sama Kafka jest dość podobna do streamów Redis. Jest to również struktura danych z wyprzedzeniem logicznym, która ma grupę konsumentów i przesunięcie. Jest to bardziej złożona struktura danych, ale pozwala na równoległe pozyskiwanie danych i podejście reaktywne.

Zobacz dokumentację strumienia Redis, aby uzyskać szczegółowe informacje.

fun publishUserPoint(userPoint: UserPoint) {
    val userPointRecord = ObjectRecord.create(USER_GEO_STREAM_NAME, userPoint)
    reactiveRedisTemplate
        .opsForStream<String, Any>()
        .add(userPointRecord)
        .subscribe{println("Send RecordId: $it")}
}

@Service
class UserPointsConsumer(
    private val userGeoService: UserGeoService
): StreamListener<String, ObjectRecord<String, UserPoint>> {

    override fun onMessage(record: ObjectRecord<String, UserPoint>) {
        userGeoService.addUserPoint(record.value)
    }
}

fun uniqueActivitiesPerDay(): Long {
    val hyperLogLogOps: HyperLogLogOperations<String, String> = stringRedisTemplate.opsForHyperLogLog()
    return hyperLogLogOps.size(Constants.TODAY_ACTIVITIES)
}

fun userOpenApp(userId: String): Long {
    val hyperLogLogOps: HyperLogLogOperations<String, String> = stringRedisTemplate.opsForHyperLogLog()
    return hyperLogLogOps.add(Constants.TODAY_ACTIVITIES, userId)
}