Projetando um aplicativo com o Redis como um armazenamento de dados. Que? Por quê?

1) Introdução

O que não veremos aqui?

• Todas as estruturas de dados do Redis em detalhes não estarão aqui. Para quais propósitos você deve ler artigos ou documentação especial.
• Aqui também não haverá código pronto para produção que você possa usar em seu trabalho.

O que veremos aqui?

• Usaremos várias estruturas de dados Redis para implementar diferentes tarefas de um aplicativo de namoro.
• Aqui estão exemplos de código Kotlin + Spring Boot.

2) Aprenda a criar e consultar perfis de usuários.

• 
  Para o primeiro, vamos aprender como criar perfis de usuários com seus nomes, curtidas, etc.
  
  Para fazer isso, precisamos de um armazenamento de valor-chave simples. Como fazer isso?
  

• Simplesmente. Um Redis tem uma estrutura de dados - um hash. Em essência, este é apenas um mapa de hash familiar para todos nós.

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

3) Atualizar curtidas de usuários usando listas do Redis.

• 
  Excelente!. Temos usuários e informações sobre curtidas.
  
  Agora devemos encontrar uma maneira de atualizar esses gostos.
  
  Assumimos que os eventos podem acontecer com muita frequência. Então, vamos usar uma abordagem assíncrona com alguma fila. E vamos ler as informações da fila em uma programação.
  

• O Redis tem uma estrutura de dados de lista com esse conjunto de comandos. Você pode usar as listas do Redis como uma fila FIFO e uma pilha LIFO.

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

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

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

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

• Como obter dados da lista em vários encadeamentos?
• E como garantir que os dados não serão perdidos em caso de erro? Da caixa - nada. Você tem que obter dados da lista em um segmento. E você precisa lidar com todas as nuances que surgem por conta própria.

4) Envio de notificações push para usuários usando pub/sub

• 
  Siga em Frente!
  Já temos perfis de usuários. Descobrimos como lidar com o fluxo de curtidas desses usuários.
  
  Mas imagine o caso em que você deseja enviar uma notificação push para um usuário no momento em que recebemos uma curtida.
  O que você vai fazer?
  

• Já temos um processo assíncrono para lidar com curtidas, então vamos apenas criar o envio de notificações push nele. Usaremos o WebSocket para esse propósito, é claro. E podemos apenas enviá-lo via WebSocket, onde recebemos um like. Mas e se quisermos executar um código de longa duração antes de enviar? Ou se quisermos delegar o trabalho com o WebSocket para outro componente?
• Tomaremos e transferiremos nossos dados novamente de uma estrutura de dados Redis (lista) para outra (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))
    }
}

@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 os usuários mais próximos por meio de operações geográficas.

• Acabamos com as curtidas. Mas e a capacidade de encontrar os usuários mais próximos de um determinado ponto.

• GeoOperations nos ajudará com isso. Armazenaremos os pares chave-valor, mas agora nosso valor é a coordenada do usuário. Para encontrar, usaremos o [radius](https://redis.io/commands/georadius) método. Passamos o ID do usuário para localizar e o próprio raio de pesquisa.

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) Atualizando a localização dos usuários por meio de streams

• 
  Implementamos quase tudo que precisamos. Mas agora temos novamente uma situação em que precisamos atualizar dados que podem ser modificados rapidamente.
  
  Então temos que usar uma fila novamente, mas seria bom ter algo mais escalável.
  

• Os streams do Redis podem ajudar a resolver esse problema.
• Provavelmente você conhece o Kafka e provavelmente até conhece os fluxos do Kafka, mas não é o mesmo que os fluxos do Redis. Mas o próprio Kafka é uma coisa bastante semelhante aos fluxos do Redis. É também uma estrutura de dados de log ahead que possui grupo de consumidores e deslocamento. Esta é uma estrutura de dados mais complexa, mas nos permite obter dados em paralelo e usando uma abordagem reativa.

O Spring possui ReactiveRedisTemplate e RedisTemplate para trabalhar com estruturas de dados Redis. Seria mais conveniente usarmos RedisTemplate para escrever o valor e ReactiveRedisTemplate para leitura. Se falamos de fluxos. Mas nesses casos, nada funcionará.
Se alguém souber por que funciona assim, por causa do Spring ou Redis, escreva nos comentários.

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

7) Conte sessões únicas usando o HyperLogLog.

• E, finalmente, vamos imaginar que precisamos calcular quantos usuários entraram no aplicativo por dia.
• Além disso, lembre-se de que podemos ter muitos usuários. Portanto, uma opção simples usando um mapa de hash não é adequada para nós porque consumirá muita memória. Como podemos fazer isso usando menos recursos?
• Uma estrutura de dados probabilística HyperLogLog entra em jogo lá. Você pode ler mais sobre isso na página da Wikipedia. Um recurso importante é que essa estrutura de dados nos permite resolver o problema usando significativamente menos memória do que a opção com um mapa de hash.

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

8) Conclusão