Diseño de una aplicación con Redis como almacén de datos. ¿Qué? ¿Por qué?

1) Introducción

¡Hola a todos! Mucha gente sabe qué es Redis y, si no lo sabe, el sitio oficial puede actualizarlo.
Para la mayoría, Redis es un caché y, a veces, una cola de mensajes.
Pero, ¿y si nos volvemos un poco locos y tratamos de diseñar una aplicación completa usando solo Redis como almacenamiento de datos? ¿Qué tareas podemos resolver con Redis?
Intentaremos responder a estas preguntas en este artículo.

¿Qué no veremos aquí?

• Todas las estructuras de datos de Redis en detalle no estarán aquí. Con qué fines debe leer artículos o documentación especiales.
• Aquí tampoco habrá código listo para producción que pueda usar en su trabajo.

¿Qué veremos aquí?

• Usaremos varias estructuras de datos de Redis para implementar diferentes tareas de una aplicación de citas.
• Aquí habrá ejemplos de código de Kotlin + Spring Boot.

2) Aprende a crear y consultar perfiles de usuario.

• Para el primero, aprendamos cómo crear perfiles de usuario con sus nombres, gustos, etc.

  Para hacer esto, necesitamos un almacén de clave-valor simple. ¿Cómo hacerlo?

• Simplemente. Un Redis tiene una estructura de datos:un hash. En esencia, este es solo un mapa hash familiar para todos nosotros.

Los comandos de lenguaje de consulta de Redis se pueden encontrar aquí y aquí.
La documentación incluso tiene una ventana interactiva para ejecutar estos comandos directamente en la página. Y la lista completa de comandos se puede encontrar aquí.
Enlaces similares funcionan para todos los comandos subsiguientes que consideraremos.

En el código, usamos RedisTemplate en casi todas partes. Esto es algo básico para trabajar con Redis en el ecosistema Spring.

La única diferencia con el mapa aquí es que pasamos "campo" como primer argumento. El "campo" es el nombre de nuestro 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")
    }

Arriba hay un ejemplo de cómo podría verse en Kotlin usando las bibliotecas de Spring.

Todos los fragmentos de código de ese artículo se pueden encontrar en Github.

3) Actualización de los gustos de los usuarios mediante listas de Redis.

• ¡Estupendo!. Tenemos usuarios e información sobre me gusta.

  Ahora deberíamos encontrar una forma de actualizar esos Me gusta.

  Asumimos que los eventos pueden ocurrir muy a menudo. Así que usemos un enfoque asíncrono con alguna cola. Y leeremos la información de la cola en un horario.

• Redis tiene una estructura de datos de lista con un conjunto de comandos de este tipo. Puede usar las listas de Redis como una cola FIFO y como una pila LIFO.

En Spring usamos el mismo enfoque para obtener ListOperations de RedisTemplate.

Tenemos que escribir a la derecha. Porque aquí estamos simulando una cola FIFO de derecha a izquierda.

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

Ahora vamos a ejecutar nuestro trabajo según lo programado.

Simplemente estamos transfiriendo información de una estructura de datos de Redis a otra. Esto nos basta como ejemplo.

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

La actualización de usuarios es realmente fácil aquí. Saluda a HashOperation de la parte anterior.

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

Y ahora mostramos cómo obtener datos de la lista. Estamos obteniendo eso de la izquierda. Para obtener un montón de datos de la lista, usaremos un range método.
Y hay un punto importante. El método de rango solo obtendrá datos de la lista, pero no los eliminará.

Así que tenemos que usar otro método para eliminar datos. trim hazlo. (Y usted puede tener algunas preguntas allí).

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

Y las preguntas son:

• ¿Cómo obtener datos de la lista en varios hilos?
• ¿Y cómo asegurarse de que los datos no se pierdan en caso de error? Desde la caja:nada. Tienes que obtener datos de la lista en un hilo. Y tienes que manejar todos los matices que surgen por tu cuenta.

4) Envío de notificaciones automáticas a los usuarios que usan pub/sub

• ¡Seguir avanzando!
  Ya tenemos perfiles de usuario. Descubrimos cómo manejar el flujo de Me gusta de estos usuarios.

  Pero imagine el caso en el que desea enviar una notificación automática a un usuario en el momento en que obtuvimos un Me gusta.
  ¿Qué vas a hacer?

• Ya tenemos un proceso asíncrono para gestionar los Me gusta, así que vamos a crear el envío de notificaciones automáticas allí. Usaremos WebSocket para ese fin, por supuesto. Y podemos enviarlo a través de WebSocket donde obtenemos un Me gusta. Pero, ¿y si queremos ejecutar un código de ejecución prolongada antes de enviarlo? ¿O si queremos delegar el trabajo con WebSocket a otro componente?
• Tomaremos y transferiremos nuestros datos nuevamente de una estructura de datos de Redis (lista) a otra (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))
    }
}

El enlace del oyente al tema se encuentra en la configuración.
Ahora, podemos llevar a nuestro oyente a un servicio separado.

@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) Encontrar los usuarios más cercanos a través de operaciones geográficas.

• Hemos terminado con los Me gusta. Pero, ¿qué pasa con la capacidad de encontrar a los usuarios más cercanos a un punto determinado?

• GeoOperations nos ayudará con esto. Guardaremos los pares clave-valor, pero ahora nuestro valor es la coordenada del usuario. Para encontrar usaremos el [radius](https://redis.io/commands/georadius) método. Pasamos el id de usuario a buscar y el propio radio de búsqueda.

Resultado de retorno de Redis, incluida nuestra identificación de usuario.

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) Actualización de la ubicación de los usuarios a través de flujos

• Implementamos casi todo lo que necesitamos. Pero ahora volvemos a tener una situación en la que tenemos que actualizar datos que podrían modificarse rápidamente.

  Así que tenemos que volver a usar una cola, pero sería bueno tener algo más escalable.

• Las secuencias de Redis pueden ayudar a resolver este problema.
• Probablemente conozca Kafka y probablemente incluso conozca las transmisiones de Kafka, pero no es lo mismo que las transmisiones de Redis. Pero Kafka en sí es algo bastante similar a las transmisiones de Redis. También es una estructura de datos de registro anticipado que tiene un grupo de consumidores y una compensación. Esta es una estructura de datos más compleja, pero nos permite obtener datos en paralelo y utilizando un enfoque reactivo.

Consulte la documentación de transmisión de Redis para obtener más detalles.

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