亿量级App Push系统(3)--推送服务

1.3 推送服务

 (1) 推送架构模型

采用异步方式，为业务方提供RPC推送服务接口，核心接口 IPushMsgService接到消息，通过MQ缓冲，进入策略过滤器，然后通过通道策略分发器，最后向渠道发送。

1.3.1 API接口设计

public interface IPushMsgService {
    /**
     * 发送模板push消息
     * @param templatePushRequest
     * @return
     */
    Results sendTemplatePush(TemplatePushRequest templatePushRequest);
}

1.3.2 发送过滤服务

(1) 过滤服务定义了很多策略，策略决定了消息是否能发送，所有的策略通过消息才能发送，如果被其中任意一条策略拦截消息将不能发送。

(2) 过滤策略接口设计，定义策略处理接口，并采用责任链模式，指定策略的执行的先后顺序

public interface StrategyChainHandler {
       /**
        * 处理消息, 返回成功继续执行下一个handle, 失败则过滤
        */
      <T> Results handle(T t, UserPushMsg userPushMsg);
}

(3) 策略过滤和业务联系比较紧密，规则实现会比较多，跟实际情况而定

@Component
public class WhiteListService implements StrategyChainHandler{
      @Override
      public <T> Results handle(T t, UserPushMsg userPushMsg) {
           Return new Result();
     }
}

(4) 策略执行器按照策略的顺序执行

@Component
public class StrategyChainProcessor {

      @Autowired
      private List<StrategyChainHandler> strategys;

      public <T> Results handle(T t, UserPushMsg userPushMsg) {
            List<StrategyChainHandler> currentHandlers = strategys.stream().filter(s->s.isSupported(t)).collect(Collectors.toList());
            for (StrategyChainHandler chain : currentHandlers) {

                 try {
                      Results results = chain.handle(t, userPushMsg);
                      if (results == null || ApiResultStatus.FAIL.getValue().equals(results.getResult())) {
                         return results;
                      }
                  } catch (Exception ex) {
                      log.error("通过handler:{}进行处理失败! context:{}", handler.getClass(), ex);
                  }
             }
            return new Results(ApiResultStatus.SUCCESS.getValue(), ApiResultStatus.SUCCESS.getValue(), ApiResultStatus.SUCCESS.getDesc());
     }
}

1.3.3 发送路由

(1) 通过策略过滤之后，进入路由发送阶段，路由会根据用户的上报渠道进行路由

(2) 一般用户会有两条路由，首先厂商路由，其次App自建长连接路由

(3) 发送路由接口设计

* 路由选择

public interface PushChannelChain  {

      /**
       * 路由选择器
       */
      <T> PushChannelChain select(T t);

     /**
      * 发送接口
      */
     <T> Results send(T t);
}

* 路由策略发送

@Component
public class StrategyDispatcher {

    @Autowired
     private PushChannelChain dispatcher

     public Results sendMessage(UserPushMsg userPushMsg) {
            Results results = dispatcher
                  .select(userPushMsg) 
                  .send(userPushMsg);
            return results;
     }
}

1.3.4 消息持久化

(1) 持久化策略

由于每天push上亿的发送量，对于存储来讲压力很大，采用分级存储模式。

* 一级存储Redis，按消息分类进行有效期存储，支持App手机端消息中心的查询服务

* 二级存储 Elasticsearch，按天索引存储，支持近期发送的消息查询，有存储天数的限制

* 三级存储 大数据仓库，存储期限不受限制，支持时间跨度的消息查询，大数据分析

(2) Redis存储的架构设计

* Redis采用多机房、多集群部署，集群单分片2.5G，分片数量40。

* 通过MurmurHash算法进行存取路由

public class RouteStrategyKeyHash implements RouteStrategy {

      @Override
       public int calculate(String businessKey, int count) {
             if (StringUtils.isEmpty(businessKey)) {
                 log.error("invalid parameter. businessKey:{}", businessKey);
                 throw new RuntimeException("invalid parameter”);
             }
             return (int) (Math.abs(MurmurHash.hash(businessKey)) % count);
    }
}

public class ClusterRedisClientFactory {

      public RedisClusterClient getClient(RouteStrategy routeStrategy, String businessKey) throws Exception {
            int index = routeStrategy.calculate(businessKey, RedisClientFactory.size());
            String routeKey = buildRedisClusterName(index);
            log.info("routeKey: {}", routeKey);
            return RedisClientFactory.get(routeKey);
      }
}

* 采用zSet数据结构，按人天维度进行存储,存取的天数和每天的最大存储量动态可调，但要保证zSet的原子操作，采用LUA脚本来保证原子操作，核心方法源码。

public void zAdd(String key, int number, long expire, byte[] value, Double score) throws Exception {

     RedisClusterClient redisClusterClient = clusterRedisClientFactory.getRouteKey(routeStrategy, key).getClient();
     String scriptSha = clusterRedisClientFactory.scriptSha(zAddStaticScript(), redisClusterClient);

     if (StringUtils.isNotBlank(scriptSha)) {
         List<byte[]> keys = new ArrayList();
         keys.add(key.getBytes(CompressUtils.UTF_8));

         List<byte[]> values = new ArrayList<>();
         values.add(value);
         values.add(new String(number + "").getBytes(CompressUtils.UTF_8)); // number ARGV[2]
         values.add(new String(expire + "").getBytes(CompressUtils.UTF_8)); // expire ARGV[3]
         values.add(new String(score + "").getBytes(CompressUtils.UTF_8)); // score ARGV[4]

         redisClusterClient.evalsha(scriptSha.getBytes(CompressUtils.UTF_8), keys, values);
      }
}

* 对存储数据进行压缩，推荐采用ZSTD

@Slf4j
public class CompressUtils {

    public static final Charset UTF_8 = Charset.forName("UTF-8");

    /**
     * 根据类型压缩
     *
     * @param src
     * @param compressType
     * @return
     * @throws Exception
     */
    public static byte[] compress(String src, CompressType compressType) {

        switch (compressType) {
            case ZSTD:
                return ArrayUtils.addAll(compressType.getValue(), Zstd.compress(src.getBytes(UTF_8)));
            case GZIP:
                return ZipUtils.gzip(src).getBytes(UTF_8);
            default:
                return src.getBytes(UTF_8);
        }

    }

    /**
     * 解压
     *
     * @param desc
     * @return
     */
    public static String decompress(byte[] desc) {
        if (desc == null || desc.length == 0) {
            return null;
        }

        if (isCompress(desc, CompressType.ZSTD.getValue())) {
            byte[] data = Arrays.copyOfRange(desc, CompressType.ZSTD.getValue().length, desc.length);
            return new String(Zstd.decompress(data, (int) Zstd.decompressedSize(data)),UTF_8);
        }

        if (isCompress(desc, CompressType.GZIP.getValue())) {
            byte[] data = Arrays.copyOfRange(desc, CompressType.GZIP.getValue().length, desc.length);
            return new String(data, UTF_8);
        }
        return new String(desc, UTF_8);
    }


    /**
     * 判断是否有压缩前缀
     *
     * @param desc
     * @param prefix
     * @return boolean
     */
    public static boolean isCompress(byte[] desc, byte[] prefix) {
        if (desc == null || desc.length < prefix.length) {
            return false;
        }

        for (int i = 0, len = prefix.length; i < len; i++) {
            if (desc[i] != prefix[i]) {
                return false;
            }
        }
        return true;
    }


    public enum CompressType {

        ZSTD("zstd", "#zstd#".getBytes(UTF_8)),
        GZIP("gzip", "#gzip#".getBytes(UTF_8));
        private String name;
        private byte[] value;

        CompressType(String name, byte[] value) {
            this.name = name;
            this.value = value;
        }

        public String getName() {
            return name;
        }

        public byte[] getValue() {
            return value;
        }
    }
}