Design af en applikation med Redis som datalager. Hvad? Hvorfor?

1) Introduktion

Hej allesammen! Mange mennesker ved, hvad Redis er, og hvis du ikke ved det, kan den officielle side bringe dig ajour.
For de fleste er Redis en cache og nogle gange en beskedkø.
Men hvad nu hvis vi går lidt amok og prøver at designe en hel applikation kun ved hjælp af Redis som datalagring? Hvilke opgaver kan vi løse med Redis?
Vi vil forsøge at besvare disse spørgsmål i denne artikel.

Hvad vil vi ikke se her?

• Hver Redis-datastruktur i detaljer vil ikke være her. Til hvilke formål bør du læse særlige artikler eller dokumentation.
• Her vil heller ikke være nogen produktionsklar kode, som du kan bruge i dit arbejde.

Hvad vil vi se her?

• Vi vil bruge forskellige Redis-datastrukturer til at implementere forskellige opgaver i en datingapplikation.
• Her er eksempler på Kotlin + Spring Boot-kode.

2) Lær at oprette og forespørge på brugerprofiler.

• For det første, lad os lære, hvordan man opretter brugerprofiler med deres navne, likes osv.

  For at gøre dette har vi brug for et simpelt nøgleværdilager. Hvordan gør man det?

• Simpelthen. En Redis har en datastruktur – en hash. I bund og grund er dette blot et velkendt hash-kort for os alle.

Redis forespørgselssprogkommandoer kan findes her og her.
Dokumentationen har endda et interaktivt vindue til at udføre disse kommandoer direkte på siden. Og hele kommandolisten kan findes her.
Lignende links virker for alle efterfølgende kommandoer, som vi vil overveje.

I koden bruger vi RedisTemplate næsten overalt. Dette er en grundlæggende ting for at arbejde med Redis i forårets økosystem.

Den ene forskel fra kortet her er, at vi passerer "mark" som det første argument. "Feltet" er vores hashs navn.

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

Ovenfor er et eksempel på, hvordan det kan se ud i Kotlin ved hjælp af Springs biblioteker.

Alle stykker kode fra den artikel kan du finde på Github.

3) Opdatering af brugernes likes ved hjælp af Redis-lister.

• Store!. Vi har brugere og information om likes.

  Nu skulle vi finde en måde, hvordan man opdaterer det like.

  Vi antager, at begivenheder kan ske meget ofte. Så lad os bruge en asynkron tilgang med en vis kø. Og vi læser informationen fra køen efter en tidsplan.

• Redis har en listedatastruktur med sådan et sæt kommandoer. Du kan bruge Redis-lister både som en FIFO-kø og som en LIFO-stak.

I foråret bruger vi den samme tilgang til at hente ListOperations fra RedisTemplate.

Vi skal skrive til højre. For her simulerer vi en FIFO-kø fra højre mod venstre.

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

Nu skal vi køre vores job efter planen.

Vi overfører simpelthen information fra én Redis-datastruktur til en anden. Dette er nok for os som eksempel.

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

Brugeropdatering er virkelig let her. Giv et hej til HashOperation fra forrige del.

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

Og nu viser vi, hvordan man får data fra listen. Det får vi fra venstre. For at få en masse data fra listen vil vi bruge et range metode.
Og der er en vigtig pointe. Områdemetoden vil kun hente data fra listen, men ikke slette dem.

Så vi er nødt til at bruge en anden metode til at slette data. trim gør det. (Og du kan have nogle spørgsmål der).

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

Og spørgsmålene er:

• Hvordan får man data fra listen ind i flere tråde?
• Og hvordan sikrer man, at dataene ikke går tabt i tilfælde af fejl? Fra kassen - ingenting. Du skal hente data fra listen i én tråd. Og du skal klare alle de nuancer, der opstår på egen hånd.

4) Afsendelse af push-meddelelser til brugere ved hjælp af pub/sub

• Fortsæt fremad!
  Vi har allerede brugerprofiler. Vi fandt ud af, hvordan vi skulle håndtere strømmen af ​​likes fra disse brugere.

  Men forestil dig det tilfælde, hvor du vil sende en push-meddelelse til en bruger i det øjeblik, vi fik et like.
  Hvad vil du gøre?

• Vi har allerede en asynkron proces til håndtering af likes, så lad os bare bygge push-meddelelser ind der. Vi vil selvfølgelig bruge WebSocket til det formål. Og vi kan bare sende det via WebSocket, hvor vi får et like. Men hvad hvis vi vil udføre langvarig kode, før vi sender? Eller hvad hvis vi ønsker at uddelegere arbejde med WebSocket til en anden komponent?
• Vi vil tage og overføre vores data igen fra én Redis-datastruktur (liste) til en anden (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))
    }
}

Den lytter, der binder til emnet, er placeret i konfigurationen.
Nu kan vi bare tage vores lytter ind i en separat tjeneste.

@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) At finde de nærmeste brugere gennem geooperationer.

• Vi er færdige med likes. Men hvad med muligheden for at finde de brugere, der er tættest på et givet punkt.

• GeoOperations vil hjælpe os med dette. Vi gemmer nøgleværdi-parrene, men nu er vores værdi brugerkoordinat. For at finde vil vi bruge [radius](https://redis.io/commands/georadius) metode. Vi videregiver bruger-id'et for at finde og selve søgeradius.

Redis returneringsresultat inklusive vores bruger-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) Opdatering af brugernes placering gennem streams

• Vi implementerede næsten alt, hvad vi har brug for. Men nu har vi igen en situation, hvor vi er nødt til at opdatere data, der kan ændre sig hurtigt.

  Så vi skal bruge en kø igen, men det ville være rart med noget mere skalerbart.

• Redis-streams kan hjælpe med at løse dette problem.
• Sandsynligvis kender du til Kafka og sandsynligvis kender du endda til Kafka-strømme, men det er ikke det samme som Redis-strømme. Men Kafka i sig selv er en ganske lignende ting som Redis-streams. Det er også en log ahead-datastruktur, der har forbrugergruppe og offset. Dette er en mere kompleks datastruktur, men den giver os mulighed for at få data parallelt og ved hjælp af en reaktiv tilgang.

Se Redis-streamdokumentationen for detaljer.

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