一：背景

实现一个多维度的排行榜（已自然周为一个周期），考虑得分和时间维度。当得分一样时，获得此排名越早的排名越靠前
需要监听原始数据，这里分为三个动作：收到、已读、通过。根据三个动作进行各项数据指标的统计
- 用户当前自然周收到、查看、标记的数量
- 根据三个动作等进行多条件过滤，准备出各个条件下的文案提示

二：方案设计

针对自然周的定义，可以参考雪花算法的实现。通过设计一个固定不可变基准开始日期A，来将某个日期B化为距离基准日A的周数X来作为周期数来表示
针对排行榜的实现，我们可以采用Redis的ZSet来实现。key：固定标识 + 固定基准日A + 距离固定基准日A的周数X value：用户id score：可以参考雪花算法的实现
- 因为score要承担两个维度：得分和时间，所以采用64位的long来进行数据整合
- score 64位：首位可以默认0，用来做保留位
- score 64位：得分可以占位23位，代表最大得分：8388608(2^23)
- score 64位：当前时间距离基准日C的时间戳（毫秒）可以占位40位，代表可以持续34年（2^40）。因为排名是倒序排的，所以这个基准日C的选择得是距离今年34后的时间作为基准日C。这样计算时间戳差值Y的时候，就可以差值Y越大，排名越靠前
- 这样一个得分就拼接完成了：0 （标识位）+ 00000 00000 000 （真正得分位）+ 00000 00000 00000 00000 00000 00000 00000 00000（时间戳差值C）。因为真正的得分权重要比时间戳高，所以真正得分位靠前
针对得分的赋值，可以考虑乐观锁 + ZADD + LUA来实现，避免覆盖更新，导致score不正确
针对监听原始数据，可以考虑观察者模式 + 线程隔离实现。基于开闭原则，高内聚低耦合，使业务更加明浪
针对三个动作的数据源进行多个条件进行过滤，得出属于每个用户的个性化文案，可以考虑责任链实现。基于开闭原则，每个过滤条件一个实现类，当条件新增，减少或者变更时可以灵活的只更改当前过滤实现类就可以，能做到影响程度最低，复用程度高，耦合程度低。

三：具体实现

redis Zset Score的实现：

基础score格式准备：

/**
 * 得分位 最大得分8388608
 */
private static final int SCORE = 23;

/**
 * 时间戳：34年
 */
private static final int TIMESTAMP = 40;

/**
 * 得分占位最大值
 */
private static final long SCORE_MAX_SIZE = ~(-1L << SCORE);

/**
 * 时间戳占位最大值
 */
private static final long TIME_STAMP_MAX_SIZE = ~(-1L << TIMESTAMP);

/**
 * 获取真实score
 * @param redisScore redis存储得分
 * @return
 */
public static BigDecimal getRealScore(Long redisScore) {
    if (redisScore == null) {
        return BigDecimal.ZERO;
    }
    long score = getRedisRealScore(redisScore);
    return new BigDecimal(score).divide(BigDecimal.TEN, 2, BigDecimal.ROUND_HALF_UP);
}

/**
 * 获取redis真实score(扩大10倍)
 * @param redisScore redis存储得分
 * @return
 */
public static long getRedisRealScore(Long redisScore) {
    if (redisScore == null) {
        return 0;
    }
    return redisScore >> TIMESTAMP & SCORE_MAX_SIZE;
}

/**
 * 计算 时间戳
 * @param redisScore redis存储得分
 * @return
 */
public static long genTimeStamp(Long redisScore) {
    if (redisScore == null) {
        return 0;
    }
    return getFixedEndTimeStamp() - (redisScore & TIME_STAMP_MAX_SIZE);
}

/**
 * 计算增加 value 值
 * @param score 得分
 * @param betweenMs 相差毫秒
 */
public static Number incScoreValue(long score, long betweenMs) {
    return ((score & SCORE_MAX_SIZE) << TIMESTAMP) | (betweenMs & TIME_STAMP_MAX_SIZE);
}

/**
 * 获取固定时间(基准起始值，千万不要改动)
 *
 * @return
 */
public static DateTime getFixedStartTime() {
    return DateUtil.parse("2020-09-07 00:00:00", DatePattern.NORM_DATETIME_PATTERN);
}

/**
 * 获取固定时间戳(基准结束值，千万不要改动)
 *
 * @return
 */
public static long getFixedEndTimeStamp() {
    return DateUtil.offset(getFixedStartTime(), DateField.YEAR, 34).getTime();
}

redis调用：

// 当前时间
Date now = new Date();
// 相差秒数
long betweenMs = fixedEndTimeStamp - currentTime.getTime();
// 获取该行业该期数对应的排行榜key
String weekRankingKey = MessageFormat.format("WEEK_RANKING:{0}:{2}", getFixedStartTime().toString(DatePattern.PURE_DATE_PATTERN), getFixedPeriod(now));
incrScore(weekRankingKey, String.valueOf(bUid), rankingScore, betweenMs);

/**
 * 设置多维度登封值
 *
 * @param key       zset key
 * @param value     zset value
 * @param getScore  此次获取的分数
 * @param betweenMs 与固定时间相差毫秒数
 * @return
 */
private boolean incrScore(String key, String value, long getScore, long betweenMs) {
    Long oldScore = null;
    long newScore;
    long totalScore;

    do {
        Double zScore = redisClient.zscore(key, value);
        if (zScore != null) {
            oldScore = zScore.longValue();
            long redisRealScore = getRedisRealScore(oldScore);
            totalScore = redisRealScore + getScore;
        } else {
            totalScore =  getScore;
        }
        // 生成新值
        newScore = incScoreValue(totalScore, betweenMs).longValue();
    } while (!compareAndSetScore(key, value, oldScore, newScore));

    return true;
}

private static String LUA_SCRIPT = "if ( (ARGV[2] == '' or ARGV[2] == nil)  and ((not (redis.call('zscore', KEYS[1], ARGV[1]))) ) or redis.call('zscore', KEYS[1], ARGV[1]) == ARGV[2]) \n" +
            "    then \n" +
            "redis.call('zadd',KEYS[1],ARGV[3],ARGV[1])\n" +
            "            redis.call('EXPIRE', KEYS[1], tonumber(ARGV[4]))\n" +
            "            return 1\n" +
            "            else\n" +
            "            return 0\n" +
            "    end";
}

/**
 * 1个月
 */
public static final int EXPIRE_ONE_MONTH = 60 * 60 * 24 * 30;

/**
 * CAS 设置score
 *
 * @param key
 * @param value
 * @param oldScore
 * @param newScore
 * @return
 */
private boolean compareAndSetScore(String key, String value, Long oldScore, long newScore) {
    Long execute = 0L;
    try {
        execute = redisClient.execute(workCallback -> {
            List<String> args = new ArrayList<>();
            args.add(value);
            args.add(Convert.toStr(oldScore, ""));
            args.add(Convert.toStr(newScore, ""));
            args.add(String.valueOf(EXPIRE_ONE_MONTH));
            return (Long) workCallback.eval(LUA_SCRIPT, Lists.newArrayList(key), args);
        });
    } catch (Exception e) {
        log.error("compareAndSetScore Exception", e);
    }

    return execute == 1L;
}

观察者模式 + 线程隔离监听

可以采用java自带的Observer和Observable来实现

public class ActionObservable extends Observable {

    private ActionObservable() {
    }

    private static volatile ActionObservable actionObservable = null;

    public static ActionObservable getInstance() {
        if (actionObservable == null) {
            synchronized (ActionObservable.class) {
                if (actionObservable == null) {
                    actionObservable = new ActionObservable();
                }
            }
        }
        return actionObservable;
    }

    /**
     * 初始化订阅者
     */
    public void initLoginObserver() {
        addObserver(new RankingObserver());
        addObserver(new OwnerTitleObserver());
    }

    public void loginNoticeAll(Dto dto) {
        setChanged();
        notifyObservers(dto);
    }
}

public abstract class AbsObserver implements Observer {

    private final Logger log = Logger.getLogger(AbsObserver.class);

    private static ThreadPoolExecutor threadPoolExecutor;

    static {
        int nThreads = Runtime.getRuntime().availableProcessors() * 2 + 1;
        ThreadPoolExecutor.CallerRunsPolicy callerRunsPolicy = new ThreadPoolExecutor.CallerRunsPolicy();
        threadPoolExecutor = new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(2048), callerRunsPolicy);
    }

    @Override
    public void update(Observable o, Object arg) {
        if (o instanceof ActionObservable) {
            if (arg instanceof Dto) {
                final Dto param = (Dto) arg;
                try {
                    threadPoolExecutor.execute(() -> change(param));
                } catch (Exception e) {
                    log.error("AbsObserver-change Exception param: " + JSON.toJSONString(param), e);
                }
            }
        }
    }

    /**
     * 接受订阅消息后执行
     *
     * @param dto
     */
    protected abstract void change(Dto dto);

}

@Slf4j
public class OwnerTitleObserver extends AbsObserver {

    @Override
    protected void change(Dto dto) {
        // 个性化文案统计
    }

}

@Slf4j
public class RankingObserver extends AbsObserver {

    @Override
    protected void change(Dto dto) {
         // 进行排名
    }

}

针对三个动作的数据源进行多个条件进行责任链过滤

责任链底层抽象

@Slf4j
public abstract class AbsFilter<T> {

    /**
     * 下一个处理链
     */
    protected AbsFilter nextFilter;

    public AbsFilter setNextFilter(AbsFilter nextFilter) {
        return this.nextFilter = nextFilter;
    }

    public void filter(T param) {
        if (param == null) {
           return;
        }
        int order = getOrder();
        if (order == 1) {
            boolean isDeal = handlerFirstBefore(param);
            if (!isDeal) {
                return;
            }
        }
        boolean handlerRes = handler(param);
        if (handlerRes) {
            if (nextFilter != null) {
                // 调用下一个链
                nextFilter.filter(param);
            }
        } else {
            handlerAfterFalse(param);
        }
    }

    /**
     * 处理逻辑
     *
     * @param param
     * @return
     */
    abstract protected boolean handler(T param);

    /**
     * 前置处理(只处理一次)
     * @param param
     * @return 是否继续处理
     */
    protected boolean handlerFirstBefore(T param) {
        // 可以进行参数校验
        ValidateUtils.validate(param);
        return true;
    }

    /**
     * 后置处理(只处理handler返回false的, 只处理一次)
     * @param param
     * @return
     */
    protected void handlerAfterFalse(T param) {}

    /**
     * 自定义排序 越小越靠前 从1开始
     * @return
     */
    protected abstract int getOrder();

}

责任链业务底层抽象

@Slf4j
public abstract class AbsOwnerTitleFilter extends AbsFilter<Dto> {

    @Override
    protected boolean handlerFirstBefore(Dto param) {
        super.handlerFirstBefore(param);
        return false;
    }

    protected void commonDeal(Dto dto) {
        // 公用处理
        // 进行个性化文案保存
    }

    @Override
    protected int getOrder() {
        return getCurrentOwnerTitleEnum().getOrder();
    }

    /**
     * 获取当前代表的个性化称号文案枚举 可以自定义，包含文案，排序，类型等字段
     *
     * @return
     */
    protected abstract OwnerTitleEnum getCurrentOwnerTitleEnum();

    @Override
    protected void handlerAfterFalse(Dto dto) {
        commonDeal(param);
    }
}

具体的过滤条件调用（示例）

@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class OwnerTitleSevenFilter extends AbsOwnerTitleFilter {

    @Override
    protected boolean handler(Dto dto) {
        // 进行业务处理，返回false则链路完成，不在进行下一链路调用，否则继续调用下一链路
        return false;
    }

    @Override
    protected OwnerTitleEnum getCurrentOwnerTitleEnum() {
        return OwnerTitleEnum.SENEN;
    }
}

调用入口可以进行调用

private OwnerTitleOneFilter ownerTitleOneFilter = Singleton.get(OwnerTitleOneFilter.class);

// 构造方法初始化
private RankingServiceImpl() {
    // 责任链过滤
    ownerTitleOneFilter
            .setNextFilter(ownerTitleTwoFilter)
            .setNextFilter(ownerTitleThreeFilter)
            .setNextFilter(ownerTitleFourFilter)
            .setNextFilter(ownerTitleFiveFilter)
            .setNextFilter(ownerTitleSixFilter)
            .setNextFilter(ownerTitleSevenFilter);
}

总结：

此次需求主要挑战在于

redis zset的多维度排序。可以参考其他框架的实现，比如这次就复用了雪花算法的一些思想，因此多看源码，我们看的更多的是思想和架构，以便能够在其他地方复用，而不是只是背。
设计模式的有效使用，可以大大降低系统的耦合度。我们不想写过多if else的原因很简单，是为了代码清晰和可扩展性强，毕竟我们都不想在一个屎山一样的代码中进行编辑，更多的是新写一个类进行我们自己的代码编辑，也能降低错误的发生。

chinotan

一：背景

一：背景

二：方案设计

三：具体实现

redis Zset Score的实现：

基础score格式准备：

redis调用：

观察者模式 + 线程隔离 监听

针对三个动作的数据源进行多个条件进行责任链过滤

总结：

观察者模式 + 线程隔离监听