Progettazione di un'applicazione con Redis come archivio dati. Che cosa? Come mai?

1) Introduzione

Ciao a tutti! Molte persone sanno cos'è Redis e, se non lo sai, il sito ufficiale può aggiornarti.
Per la maggior parte dei Redis è una cache e talvolta una coda di messaggi.
Ma cosa succede se impazziamo e proviamo a progettare un'intera applicazione utilizzando solo Redis come archiviazione dati? Quali compiti possiamo risolvere con Redis?
Cercheremo di rispondere a queste domande, in questo articolo.

Cosa non vedremo qui?

• Ogni struttura di dati Redis in dettaglio non sarà qui. Per quali scopi dovresti leggere articoli o documentazione speciali.
• Qui non ci sarà nemmeno codice pronto per la produzione che potresti usare nel tuo lavoro.

Cosa vedremo qui?

• Utilizzeremo varie strutture di dati Redis per implementare diverse attività di un'applicazione di appuntamenti.
• Qui ci saranno esempi di codice Kotlin + Spring Boot.

2) Impara a creare e interrogare i profili utente.

• Per il primo, impariamo come creare profili utente con i loro nomi, Mi piace, ecc.

  Per fare ciò, abbiamo bisogno di un semplice archivio di valori-chiave. Come farlo?

• Semplicemente. Un Redis ha una struttura dati:un hash. In sostanza, questa è solo una mappa hash familiare per tutti noi.

I comandi del linguaggio di query Redis possono essere trovati qui e qui.
La documentazione ha anche una finestra interattiva per eseguire questi comandi direttamente sulla pagina. E l'intera lista dei comandi può essere trovata qui.
Collegamenti simili funzionano per tutti i comandi successivi che prenderemo in considerazione.

Nel codice utilizziamo RedisTemplate quasi ovunque. Questa è una cosa fondamentale per lavorare con Redis nell'ecosistema Spring.

L'unica differenza rispetto alla mappa qui è che passiamo "campo" come primo argomento. Il "campo" è il nome del nostro hash.

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

Sopra c'è un esempio di come potrebbe apparire in Kotlin usando le librerie di Spring.

Tutti i pezzi di codice di quell'articolo li puoi trovare su Github.

3) Aggiornare i Mi piace degli utenti utilizzando le liste Redis.

• Grande!. Abbiamo utenti e informazioni sui Mi piace.

  Ora dovremmo trovare un modo per aggiornare i Mi piace.

  Partiamo dal presupposto che gli eventi possono accadere molto spesso. Quindi usiamo un approccio asincrono con qualche coda. E leggeremo le informazioni dalla coda in base a un programma.

• Redis ha una struttura dati elenco con un tale insieme di comandi. Puoi utilizzare elenchi Redis sia come coda FIFO che come stack LIFO.

In primavera utilizziamo lo stesso approccio per ottenere ListOperations da RedisTemplate.

Dobbiamo scrivere a destra. Perché qui stiamo simulando una coda FIFO da destra a sinistra.

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

Ora eseguiremo il nostro lavoro nei tempi previsti.

Stiamo semplicemente trasferendo informazioni da una struttura dati Redis a un'altra. Questo ci basta come esempio.

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

L'aggiornamento degli utenti è davvero facile qui. Saluta HashOperation dalla parte precedente.

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

E ora mostriamo come ottenere i dati dall'elenco. Lo stiamo capendo da sinistra. Per ottenere una serie di dati dall'elenco utilizzeremo un range metodo.
E c'è un punto importante. Il metodo range otterrà solo i dati dall'elenco, ma non lo cancellerà.

Quindi dobbiamo usare un altro metodo per eliminare i dati. trim fallo. (E puoi avere alcune domande lì).

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

E le domande sono:

• Come ottenere i dati dall'elenco in più thread?
• E come garantire che i dati non vadano persi in caso di errore?Dalla scatola:niente. Devi ottenere i dati dall'elenco in un thread. E devi gestire da solo tutte le sfumature che si presentano.

4) Invio di notifiche push agli utenti tramite pub/sub

• Continuare ad andare avanti!
  Abbiamo già profili utente. Abbiamo capito come gestire il flusso di Mi piace di questi utenti.

  Ma immagina il caso in cui vuoi inviare una notifica push a un utente nel momento in cui abbiamo ricevuto un Mi piace.
  Cosa farai?

• Abbiamo già un processo asincrono per la gestione dei Mi piace, quindi costruiamo l'invio di notifiche push lì dentro. Useremo WebSocket per questo scopo, ovviamente. E possiamo semplicemente inviarlo tramite WebSocket dove riceviamo un Mi piace. Ma cosa succede se vogliamo eseguire codice di lunga durata prima di inviare? O se volessimo delegare il lavoro con WebSocket a un altro componente?
• Preleveremo e trasferiremo nuovamente i nostri dati da una struttura dati Redis (elenco) a un'altra (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))
    }
}

L'associazione del listener all'argomento si trova nella configurazione.
Ora possiamo semplicemente portare il nostro ascoltatore in un servizio separato.

@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) Trovare gli utenti più vicini tramite operazioni geografiche.

• Abbiamo finito con i Mi piace. Ma che dire della possibilità di trovare gli utenti più vicini a un determinato punto.

• GeoOperations ci aiuterà in questo. Conserveremo le coppie chiave-valore, ma ora il nostro valore è la coordinata utente. Per trovarlo useremo il [radius](https://redis.io/commands/georadius) metodo. Passiamo l'ID utente da trovare e il raggio di ricerca stesso.

Risultato restituito Redis incluso il nostro ID utente.

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) Aggiornamento della posizione degli utenti tramite stream

• Abbiamo implementato quasi tutto ciò di cui abbiamo bisogno. Ma ora abbiamo di nuovo una situazione in cui dobbiamo aggiornare i dati che potrebbero essere modificati rapidamente.

  Quindi dobbiamo usare di nuovo una coda, ma sarebbe bello avere qualcosa di più scalabile.

• Gli stream Redis possono aiutare a risolvere questo problema.
• Probabilmente conosci Kafka e probabilmente conosci anche i flussi Kafka, ma non è la stessa cosa dei flussi Redis. Ma lo stesso Kafka è una cosa abbastanza simile allo streaming di Redis. È anche una struttura di dati log ahead con gruppo di consumatori e offset. Questa è una struttura di dati più complessa, ma ci consente di ottenere dati in parallelo e utilizzando un approccio reattivo.

Consulta la documentazione del flusso Redis per i dettagli.

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