day15短信服务专题

1、设计一个通用的短信服务平台，需要哪些考虑点（难点）？

1、短信模板和签名如何申请: 需要事先在三方平台开通短信账号，在平台中申请签名和模板，通过审核后才可使用 ；
2、在高并发情况下如果保证系统的稳定性（高可用）:
搭建短信服务集群，提升服务的的可用性；
使用spring-cloud-stream-rabbitMQ来实现异步发送【流量削峰】来保证其发送稳定性 ；
3、后续对接新渠道的扩展性（高可扩）: 需要使用工厂加代理模式构建易于扩展的平台 ；
4、短信发送失败的补发机制（集群容错）: 如果三方短信平台欠费或者其他原因导致短信失败后，我们应自动切换其他短信平台继续发送；
5、负载均衡：集群解决负载均衡，算法是由我们自己实现，轮询、随机、一致性hash算法等；
6、渠道路由：类似支付，我们采用适配器设计模式，初始化各个平台实现类到IOC容器中，并存储到一个Map中，根据渠道类型，获取对应的实现类；

2、发送短信的业务流程

具体业务流程：
1、业务系统调用短信业务系统，此时直接返回发送结果
2、短信业务系统生产发送短信消息到rabbitMQ,此处是做的削峰操作，防止短信发送量过大
3、短信监听从rabbitMQ中拉取短信消息
4、调用SmsSendAdapter适配器过滤黑名单
5、通过负载均衡选择对应渠道
6、查询渠道消息，选择模板
7、查询签名消息
8、通过模板变量兑换参数
9、推送短信消息到三方平台
10、轮询三方平台查看短信发送结果
11、短信业务平台同步发送结果

3、自定义负载均衡算法，及优化过程，以（权重）轮询算法为例（重点）

方案1：
1、获取短信模板对应的渠道及权重比例；
2、使用synchronized同步锁加锁；
3、以频道列表的长度为判断，当索引大于等于集合长度时，重置索引为0；否则，获取当前渠道名称，索引自增
缺点：synchronized同步锁加锁性能较差；

优化1：
采用java.util.concurrent.atomic.AtomicInteger（juc）解决线程安全问题；AtomicInterger基于CAS算法实现占用cpu，使用线程调度解决。使用AtomicInteger.incrementAndGet()替换pos++

优化2：
参考springcloud ribbon负载均衡算法，优化当前代码

 private int incrementAndGetModulo(int modulo) {
        for (;;) {  //自旋锁
            int current = nextChannelCyclicCounter.get();
            int next = (current + 1) % modulo;  // 避免数组越界
            if (nextChannelCyclicCounter.compareAndSet(current, next))  // CAS 思想
                return next;
        }
    }

/**
 * @ClassName RoundRobin.java
 * @Description 轮询（Round Robin）法
 * 自定义负载均衡算法：
 * * 防止数组越界 （按位与, 对集合大小取模运算）
 * * 按照权重设置Map
 *     * key： 渠道名称
 *     * value： 权重
 * 思想： 取value权重的值，值对应的渠道在集合里添加对应的权重的数量
 * 如：
 * MAP： {"ali":2,"baidu":1,"tencent":"2"}
 * List: ["ali","ali","baidu","tencent","tencent"]
 * 在按照轮询/随机算法在获取对应集合下标的值即可
 */
@Component
public class RoundRobinSend extends BaseSendLoadBalancer {
    private static Integer pos = 0;
    @Override
    public String chooseChannel(List<SmsTemplate> SmsTemplates, Set<String> mobile) {
        //获得当前模板对应的渠道
        Map<String, String> channelMap = super.getChannelList(SmsTemplates);
        // 取得通道地址List
        Set<String> keySet = channelMap.keySet();
        ArrayList<String> keyList = new ArrayList<String>();
        keyList.addAll(keySet);
        String channelName = null;
        synchronized (pos) {
            if (pos >= keySet.size())
                pos = 0;
            channelName = keyList.get(pos);
            pos ++;
        }
        return channelName;
    }
 /**
 * 优化1： synchronized大量同步锁，性能不高，使用 AtomicLong对象解决线程安全问题
 * 存在的问题： AtomicInteger 基于CAS实现占用CPU， 使用线程调度解决
 */
    private static AtomicInteger pos = new AtomicInteger(0);
    @Override
    public String chooseChannel(List<SmsTemplate> smsTemplates, Set<String> mobile) {
        //获得当前模板对应的渠道 ALIYUN_SMS 2
        Map<String, String> channelMap = super.getChannelList(smsTemplates);
        // 取得通道地址List 3
        Set<String> keySet = channelMap.keySet();
        ArrayList<String> keyList = new ArrayList<String>(keySet);
        String channelName = null;
        if (pos.incrementAndGet() >= keySet.size())
            pos.set(0);
        channelName = keyList.get(pos.get());
        return channelName;
    }
 /**
 * 优化2： 参考SpringCloud Ribbon 负载均衡算法，优化当前代码
 */
     private AtomicInteger nextChannelCyclicCounter;
    public RoundRobinSend2() {
        nextChannelCyclicCounter = new AtomicInteger(0);
    }
    public String chooseChannel(List<SmsTemplate> smsTemplates, Set<String> mobile) {
        if (EmptyUtil.isNullOrEmpty(smsTemplates)) {
            log.warn("no load balancer");
            return null;
        }
        //获取所有的渠道列表
        Map<String, String> channelMap = super.getChannelList(smsTemplates);
        Set<String> keySet = channelMap.keySet();
        ArrayList<String> keyList = new ArrayList<>(keySet);
        String channelName = null;
        int count = 0;
        //循环10次后自动跳出循环
        while (channelName == null && count++ < 10) {
            // 得到所有的渠道集合大小
            int upCount = keyList.size();
            if (upCount == 0) {
                log.warn("No up servers available from load balancer: " + smsTemplates);
                return null;
            }
            //轮询算法实现
            int nextServerIndex = incrementAndGetModulo(upCount);
            channelName = keyList.get(nextServerIndex);
            if (channelName == null) {
                /* Transient. */
                Thread.yield();
                continue;
            }
        }
        if (count >= 10) {
            log.warn("No available alive servers after 10 tries from load balancer: ");
        }
        return channelName;
    }
    /**
     * Inspired by the implementation of {@link AtomicInteger#incrementAndGet()}.
     * @param modulo 集合大小
     * @return 获取集合中下标
     */
    private int incrementAndGetModulo(int modulo) {
        for (;;) {  //自旋锁
            int current = nextChannelCyclicCounter.get();
            int next = (current + 1) % modulo;  // 避免数组越界
            if (nextChannelCyclicCounter.compareAndSet(current, next))  // CAS 思想
                return next;
        }
    }
}