Entwerfen einer Anwendung mit Redis als Datenspeicher. Was? Wieso den?

1) Einführung

Hallo zusammen! Viele Leute wissen, was Redis ist, und wenn Sie es nicht wissen, kann die offizielle Website Sie auf den neuesten Stand bringen.
Für die meisten Redis ist es ein Cache und manchmal eine Nachrichtenwarteschlange.
Aber was, wenn wir ein bisschen verrückt werden und versuchen, eine ganze Anwendung nur mit Redis als Datenspeicher zu entwerfen? Welche Aufgaben können wir mit Redis lösen?
Wir werden versuchen, diese Fragen in diesem Artikel zu beantworten.

Was werden wir hier nicht sehen?

• Jede Redis-Datenstruktur im Detail wird hier nicht sein. Für welche Zwecke sollten Sie spezielle Artikel oder Dokumentationen lesen.
• Hier wird es auch keinen produktionsreifen Code geben, den Sie in Ihrer Arbeit verwenden könnten.

Was werden wir hier sehen?

• Wir werden verschiedene Redis-Datenstrukturen verwenden, um verschiedene Aufgaben einer Dating-Anwendung zu implementieren.
• Hier finden Sie Codebeispiele für Kotlin + Spring Boot.

2) Erfahren Sie, wie Sie Benutzerprofile erstellen und abfragen.

• Lassen Sie uns zunächst lernen, wie man Benutzerprofile mit ihren Namen, Vorlieben usw. erstellt.

  Dazu benötigen wir einen einfachen Key-Value-Store. Wie es geht?

• Einfach. Ein Redis hat eine Datenstruktur – einen Hash. Im Wesentlichen ist dies nur eine vertraute Hash-Map für uns alle.

Befehle für die Redis-Abfragesprache finden Sie hier und hier.
Die Dokumentation hat sogar ein interaktives Fenster, um diese Befehle direkt auf der Seite auszuführen. Und die gesamte Befehlsliste finden Sie hier.
Ähnliche Links funktionieren für alle nachfolgenden Befehle, die wir berücksichtigen werden.

Im Code verwenden wir fast überall RedisTemplate. Dies ist eine grundlegende Sache für die Arbeit mit Redis im Spring-Ökosystem.

Der einzige Unterschied zur Map besteht darin, dass wir "field" als erstes Argument übergeben. Das „Feld“ ist der Name unseres Hashs.

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")
    }

Oben ist ein Beispiel dafür, wie es in Kotlin mit den Bibliotheken von Spring aussehen könnte.

Alle Codeteile aus diesem Artikel finden Sie auf Github.

3) Aktualisieren von Benutzervorlieben mithilfe von Redis-Listen.

• Toll!. Wir haben Benutzer und Informationen über Likes.

  Jetzt sollten wir einen Weg finden, wie wir diese Likes aktualisieren können.

  Wir gehen davon aus, dass Ereignisse sehr oft passieren können. Verwenden wir also einen asynchronen Ansatz mit einer Warteschlange. Und wir werden die Informationen aus der Warteschlange nach einem Zeitplan lesen.

• Redis hat eine Listendatenstruktur mit einem solchen Befehlssatz. Sie können Redis-Listen sowohl als FIFO-Warteschlange als auch als LIFO-Stack verwenden.

In Spring verwenden wir denselben Ansatz, um ListOperations von RedisTemplate abzurufen.

Wir müssen rechts schreiben. Denn hier simulieren wir eine FIFO-Warteschlange von rechts nach links.

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)
}

Jetzt werden wir unsere Arbeit planmäßig ausführen.

Wir übertragen einfach Informationen von einer Redis-Datenstruktur in eine andere. Das reicht uns als Beispiel.

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

Die Benutzeraktualisierung ist hier wirklich einfach. Begrüßen Sie HashOperation aus dem vorherigen Teil.

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))
    }

Und jetzt zeigen wir, wie man Daten aus der Liste bekommt. Das bekommen wir von links. Um eine Reihe von Daten aus der Liste zu erhalten, verwenden wir einen range Methode.
Und es gibt einen wichtigen Punkt. Die Bereichsmethode holt nur Daten aus der Liste, löscht sie aber nicht.

Wir müssen also eine andere Methode verwenden, um Daten zu löschen. trim Tu es. (Und Sie können dort einige Fragen haben).

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)
}
}

Und die Fragen sind:

• Wie bekomme ich Daten aus der Liste in mehrere Threads?
• Und wie stellen Sie sicher, dass die Daten im Fehlerfall nicht verloren gehen?Aus der Box - nichts. Sie müssen Daten aus der Liste in einem Thread abrufen. Und mit allen Nuancen, die sich ergeben, müssen Sie selbst fertig werden.

4) Senden von Push-Benachrichtigungen an Benutzer mit pub/sub

• Gib niemals auf!
  Wir haben bereits Benutzerprofile. Wir haben herausgefunden, wie wir mit dem Strom von Likes von diesen Benutzern umgehen können.

  Aber stellen Sie sich den Fall vor, in dem Sie eine Push-Benachrichtigung an einen Benutzer senden möchten, sobald wir ein Like erhalten haben.
  Was wirst du tun?

• Wir haben bereits einen asynchronen Prozess zum Umgang mit Likes, also bauen wir einfach das Senden von Push-Benachrichtigungen dort ein. Wir werden natürlich WebSocket für diesen Zweck verwenden. Und wir können es einfach über WebSocket senden, wo wir ein Like bekommen. Aber was ist, wenn wir lange laufenden Code vor dem Senden ausführen möchten? Oder was ist, wenn wir die Arbeit mit WebSocket an eine andere Komponente delegieren möchten?
• Wir werden unsere Daten erneut von einer Redis-Datenstruktur (Liste) in eine andere (Pub/Sub) übernehmen und übertragen.

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))
    }
}

Die Listenerbindung zum Topic befindet sich in der Konfiguration.
Jetzt können wir unseren Zuhörer einfach in einen separaten Dienst bringen.

@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) Finden der nächstgelegenen Benutzer durch Geooperationen.

• Wir sind fertig mit Likes. Aber was ist mit der Möglichkeit, die Benutzer zu finden, die einem bestimmten Punkt am nächsten sind?

• GeoOperations wird uns dabei helfen. Wir werden die Schlüssel-Wert-Paare speichern, aber jetzt ist unser Wert die Benutzerkoordinate. Um ihn zu finden, verwenden wir den [radius](https://redis.io/commands/georadius) Methode. Wir übergeben die zu findende Benutzer-ID und den Suchradius selbst.

Redis-Rückgabeergebnis einschließlich unserer Benutzer-ID.

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) Aktualisieren des Standorts von Benutzern durch Streams

• Wir haben fast alles implementiert, was wir brauchen. Aber jetzt haben wir wieder eine Situation, in der wir Daten aktualisieren müssen, die sich schnell ändern könnten.

  Wir müssen also wieder eine Warteschlange verwenden, aber es wäre schön, etwas Skalierbareres zu haben.

• Redis-Streams können helfen, dieses Problem zu lösen.
• Wahrscheinlich kennen Sie Kafka und wahrscheinlich kennen Sie sogar Kafka-Streams, aber es ist nicht dasselbe wie Redis-Streams. Aber Kafka selbst ist eine ziemlich ähnliche Sache wie Redis-Streams. Es ist auch eine Log-Ahead-Datenstruktur, die Verbrauchergruppe und Offset hat. Dies ist eine komplexere Datenstruktur, aber sie ermöglicht es uns, Daten parallel und mit einem reaktiven Ansatz zu erhalten.

Weitere Informationen finden Sie in der Redis-Stream-Dokumentation.

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)
}