App视频邀请

发表于 | 分类于 | 阅读次数:
字数统计: 1,980 字 | 阅读时长 ≈ 8 分钟

需求

项目中有这么一个需求：

当用户余额不足，1分钟后，机器人进行视频邀请，当用户点击接听时，则提示用户充值；当用户点击拒绝，3分钟后，再对该用户使用机器人进行视频邀请，当用户点击接听时，则提示用户充值；当用户点击拒绝，10分钟后，再次对该用户使用机器人进行视频邀请，当用户点击接听时，则提示用户充值；当用户点击拒绝，3次诱导充值结束。

当用户余额充足，1分钟后，推荐真实用户对该用户进行视频邀请，若该用户接听，则对真实用户发送视频邀请；当用户挂断，3分钟后，继续推荐真实用户进行视频邀请，若该用户接听，则对真实用户发送视频邀请，当用户挂断，10分钟后，继续推荐真实用户进行视频邀请。

当用户余额不够时，继续走余额不够的逻辑。

分析这个需求，难点无非就是三次时间间隔，开始考虑的是使用消息队列RocketMQ，但用RocketMQ有点大材小用的意思。后面考虑用Redis，如果Redis有对过期时间的监听，那岂不美哉，我擦，谷歌了一发，还真TM有。于是，就研究了一发，也是比较简单。

Redis对过期时间的监听是这样的：使用String类型，设置Key-Value，对该Key设置过期时间，当时间过期后，触发某个事件，这就是所谓的 对过期事件的监听。过期事件是通过Redis的发布订阅功能来进行分发。

事件类型

对于每个修改数据库的操作，键空间通知都会发送两种不同类型的事件消息：keyspace 和 keyevent。以 keyspace 为前缀的频道被称为键空间通知（key-space notification）， 而以 keyevent 为前缀的频道则被称为键事件通知（key-event notification）。

事件是用 keyspace@DB:KeyPattern 或者 keyevent@DB:OpsType 的格式来发布消息的。
DB表示在第几个库；KeyPattern则是表示需要监控的键模式（可以用通配符，如：key:）；OpsType则表示操作类型。因此，如果想要订阅特殊的Key上的事件，应该是订阅keyspace。
比如说，对 0 号数据库的键 mykey 执行 DEL 命令时， 系统将分发两条消息， 相当于执行以下两个 PUBLISH 命令：
PUBLISH keyspace@0:sampleKey del
PUBLISH keyevent@0:del sampleKey
订阅第一个频道 keyspace@0:mykey 可以接收 0 号数据库中所有修改键 mykey 的事件，而订阅第二个频道 keyevent@0:del 则可以接收 0 号数据库中所有执行 del 命令的键。

开启配置

键空间通知通常是不启用的，因为这个过程会产生额外消耗。所以在使用该特性之前，请确认一定是要用这个特性的，然后修改配置文件，或使用config配置。相关配置项如下：

输入的参数中至少要有一个 K 或者 E ， 否则的话， 不管其余的参数是什么， 都不会有任何通知被分发。上表中斜体的部分为通用的操作或者事件，而黑体则表示特定数据类型的操作。在redis的配置文件redis.conf中修改 notify-keyspace-events “Kx”，注意：这个双引号是一定要的，否则配置不成功，启动也不报错。例如，“Kx”表示想监控某个Key的失效事件。也可以在命令行通过config配置：CONFIG set notify-keyspace-events Ex （但非持久化）。

实现步骤

  1. 修改redis.conf配置文件中的 notify-keyspace-events “Kx”，redis默认是关闭的
  2. 对SpringBoot整合 Redis的发布订阅，指定监听类和监听类型

代码示例

pom依赖
1
2
3
4
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
redis工具类（部分）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;

/**
 * redis缓存客户端
 */
@Component
public class RedisCacheUtils<T> {

    @Autowired
    private  RedisTemplate<String, T> redisTemplate;

    /**
     * 写入单个对象到缓存(可以设置有效时间)
     * @param key
     * @param value
     * @param expireTime 有效时间 单位秒
     * @return
     */
    public boolean set(final String key, T value, Long expireTime) {
        boolean result = false;
        try {
            ValueOperations<String, T> operations = redisTemplate.opsForValue();
            operations.set(key, value);
            redisTemplate.expire(key, expireTime, TimeUnit.SECONDS);
            result = true;
        } catch (Exception e) {
            throw e;
        }
        return result;
    }

    /**
     * 自增
     * @param key
     * @param by
     * @param seconds
     * @return
     */
    public Long incr(final String key, final long by,final long seconds) {
        Long count = redisTemplate.opsForValue().increment(key, by);
        redisTemplate.expire(key, seconds, TimeUnit.SECONDS);
        return count;
    }
}
监听配置
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import com.app.common.constants.SystemConstant;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

@Configuration
public class RedisLinstenerConfig {

    @Autowired
    private RedisConnectionFactory redisConnectionFactory;

    @Bean
    public ConsumerRedisListener consumerRedis() {
        return new ConsumerRedisListener();
    }

    @Bean
    public ChannelTopic topic() {
        return new ChannelTopic("__keyevent@0__:expired");
    }

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer() {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        container.addMessageListener(consumerRedis(),topic());
        return container;
    }
}

redis监听器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import com.app.cache.RedisCacheUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;

public class ConsumerRedisListener implements MessageListener {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedisCacheUtils redisCacheUtils;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        doBusiness(message);
    }

    /**
     * 打印 message body 内容
     * @param message
     */
    public void doBusiness(Message message) {
        Object value = stringRedisTemplate.getValueSerializer().deserialize(message.getBody());
        byte[] body = message.getBody();
        byte[] channel = message.getChannel();
        String topic = new String(channel);
        String itemValue = new String(body);
        System.out.println("itemValue-----------------------" + itemValue);
     
        // 如果key中包含^，则说明是 视频邀请的
        if(itemValue.contains("^")) {
            String[] keyArr = itemValue.split("\\^");
			String userId = keyArr[1];
			// 防止重复消费，设置一个过期时间
			Long num = redisCacheUtils.incr(userId + "_incr", 1L, 60L);
			if(StringUtils.isBlank(userId)) {
				return;
			}
			if(num == 1){
				// 处理逻辑，给App推送消息，调起视频呼叫
                 //…………
			}
        }
    }
}

看了上面的代码可能有点懵，貌似和上述所说的时间间隔并没有什么瓜葛，然而并不是。首先，当用户当日首次登陆App时，客户端用调用一个接口，表示用户进入App，我会在接口中判断用户是不是当日首次登陆，如果是，则使用”video” + “^” + 用户的ID + “^” + 180 作为一个Key，value无所谓，并对该key设置60秒的过期时间，当该key过期，则会进入到redis监听中，并对客户端推送消息，其中，消息体中包含一个关键字段，此关键字段就是下次需要间隔多久来发起视频邀请，即之前过期Key后面跟随的180，当客户端点击挂断，调用挂断接口时，就将此字段传过来，然后 使用”video” + “^” + 用户的ID + “^” + 600 作为一个Key，并对该key设置180秒的过期时间，后面逻辑同理……

然而，因为项目是分布式项目，会部署多个节点，这样就存在重复订阅，因为这一部分数据老大要求不能存到数据库，所以使用了redis 的incr来记录进入过期监听器的次数，并设置过期时间为60秒，这样 多个节点即使重复订阅，也会只有一个订阅者可以处理逻辑，即对客户端推送消息，这里的推送消息使用的是融云的IM，后续对该IM进行分析。

欢迎关注我的公众号~ 搜索公众号： 翻身码农把歌唱 或者 扫描下方二维码：

img

坚持原创技术分享，您的支持将鼓励我继续创作！
-------------本文结束感谢您的阅读-------------
分享到：
0%