package com.ydl.sms.entity.base;

import com.baomidou.mybatisplus.annotation.TableId;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;

import java.io.Serializable;
import java.time.LocalDateTime;

/**
 * 基础实体类，所有实体都需要继承
 */
@Data
public abstract class BaseEntity implements Serializable {

    @TableId
    @ApiModelProperty(value = "主键")
    private String id;

    @ApiModelProperty(value = "创建时间")
    private LocalDateTime createTime;

    @ApiModelProperty(value = "创建人")
    private String createUser = "0";

    @ApiModelProperty(value = "修改时间")
    private LocalDateTime updateTime;

    @ApiModelProperty(value = "修改人")
    private String updateUser;

    @ApiModelProperty(value = "逻辑删除：0删除")
    private Integer isDelete;

}

@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface DefaultParams {
}

/**
 * 通过切面方式，自定义注解，实现实体基础数据的注入（创建者、创建时间、修改者、修改时间）
 */
@Component
@Aspect
@Slf4j
public class DefaultParamsAspect {
    @SneakyThrows
    @Before("@annotation(com.itheima.sms.annotation.DefaultParams)")
    public void beforeEvent(JoinPoint point) {
        //从threadlocal中获取用户id
        Long userId = BaseContextHandler.getUserId();
        if (userId == null) {
            userId = 0L;
        }
        Object[] args = point.getArgs();
        for (int i = 0; i < args.length; i++) {
            Class<?> classes = args[i].getClass();
            Object id = null;
            Method method = getMethod(classes, "getId");
            if (null != method) {
                id = method.invoke(args[i]);
            }

            //请求操作的对象的id为空时，为创建操作
            if (null == id) {
                method = getMethod(classes, "setCreateUser", String.class);
                if (null != method) {
                    method.invoke(args[i], userId.toString());
                }
                method = getMethod(classes, "setCreateTime", LocalDateTime.class);
                if (null != method) {
                    method.invoke(args[i], LocalDateTime.now());
                }
            }

            //新建修改更新
            method = getMethod(classes, "setUydlateUser", String.class);
            if (null != method) {
                method.invoke(args[i], userId.toString());
            }
            method = getMethod(classes, "setUydlateTime", LocalDateTime.class);
            if (null != method) {
                method.invoke(args[i], LocalDateTime.now());
            }
        }
    }

    private Method getMethod(Class classes, String name, Class... types) {
        try {
            return classes.getMethod(name, types);
        } catch (NoSuchMethodException e) {
            return null;
        }
    }
}

@PostMapping
@ApiOperation("保存")
@DefaultParams
public R save(@RequestBody ObjectEntity entity) {
    //具体实现逻辑...
    return R.success();
}

package com.ydl.sms.listener;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;

/**
 * 自定义消息监听器，用于监听Redis频道中的消息
 */
@Component
@Slf4j
public class MyListener implements MessageListener {
    /**
     * 监听方法
     * @param message
     * @param pattern
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        log.info("接收到消息：" + message);
    }
}

package com.ydl.sms.config;

import com.ydl.sms.listener.MyListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
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.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

/**
 * 订阅发布模式的容器配置
 */
@Configuration
@AutoConfigureAfter({MyListener.class})
public class SubscriberConfig {

    @Autowired
    private MyListener myListener;

    /**
     * 创建消息监听容器
     *
     * @param redisConnectionFactory
     * @return
     */
    @Bean
    public RedisMessageListenerContainer getRedisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory) {
        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
        redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);

        //可以添加多个监听订阅频道
        //当前监听的是通道：MYTOPIC
        redisMessageListenerContainer.addMessageListener(new MessageListenerAdapter(myListener), new PatternTopic("MYTOPIC"));

        return redisMessageListenerContainer;
    }
}

package com.ydl.test;

import com.ydl.sms.SmsManageApplication;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = SmsManageApplication.class)
public class RedisTest {
    @Autowired
    private RedisTemplate redisTemplate;

    //@Test
    public void test1(){
        for (int i = 0; i < 10; i++) {
            redisTemplate.convertAndSend("MYTOPIC","im itlilaoshi");
        }
    }

}

//发送消息，通知短信发送服务更新缓存中的通道优先级
public void sendUpdateMessage() {
    //获得所有注册到redis中的短信发送服务实例
    Map map = redisTemplate.opsForHash().entries("SERVER_ID_HASH");
    log.info("全部服务：{}", map);
    Long current = System.currentTimeMillis();

    for (Object key : map.keySet()) {
        long valueLong = Long.parseLong(map.get(key).toString());

        //五分钟内报告，说明短信发送服务状态正常
        if (current - valueLong < (1000 * 60 * 5)) {
            //删除redis中缓存的可用通道，因为通道优先级发生变化，redis中缓存的可用通道需要重新加载
            redisTemplate.delete("listForConnect");

            //发布消息
            redisTemplate.convertAndSend("TOPIC_HIGH_SERVER",
                                         ServerTopic
                                         .builder()
                                         .option(ServerTopic.INIT_CONNECT)
                                         .value(key.toString())
                                         .build()
                                         .toString()
                                        );
            log.info("发送消息通知短信发送服务重新构建通道");
            return;
        }
    }
}

package com.ydl.test;

import com.ydl.sms.SmsApiApplication;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.concurrent.TimeUnit;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = SmsApiApplication.class)
public class RedisListTest {
    @Autowired
    RedisTemplate redisTemplate;

    //消息发送者
    @Test
    public void testPush(){
        for (int i = 0; i < 10; i++) {
            redisTemplate.opsForList().leftPush("ydllist123", "msg"+i);
        }
    }

    //消息消费者
    @Test
    public void testPop(){
        for (int i = 0; i < 11; i++) {
            Object value = redisTemplate.opsForList().rightPop("ydllist123");
            System.out.println("取出的值是："+value);
        }
    }

    //消费者阻塞监听队列
    @Test
    public void testPopBlock(){
       while (true){
           Object value = redisTemplate.opsForList().rightPop("ydllist123", 10L, TimeUnit.SECONDS);
           //业务逻辑
           System.out.println("取出的值是："+value);
       }
    }
}

//缺啥补啥
    /**
     * 根据短信模板分类 并分发
     *
     * @param templateEntity
     * @param smsSendDTO
     * @param platformEntity
     */
    private void pushSmsMessage(TemplateEntity templateEntity, SmsSendDTO smsSendDTO, PlatformEntity platformEntity) {
        // TODO 短信接收服务：将短信信息保存到数据库或者Redis队列
        ReceiveLogEntity receiveLogEntity = new ReceiveLogEntity();
        receiveLogEntity.setApiLogId(UUID.randomUUID().toString());
        long start = System.currentTimeMillis();
        try {
            //1、进行短信分类，分为实时发送短信和定时发送短信
            String sendTime = smsSendDTO.getSendTime();

            String request = JSON.toJSONString(smsSendDTO);
            if (StringUtils.isNotEmpty(sendTime)) {
                //2、如果是定时发送短信则将消息保存到Mysql数据库
                TimingPushEntity timingPushEntity = new TimingPushEntity();
                timingPushEntity.setMobile(smsSendDTO.getMobile());
                timingPushEntity.setTemplate(smsSendDTO.getTemplate());
                timingPushEntity.setRequest(request);
                timingPushEntity.setSignature(smsSendDTO.getSignature());
                timingPushEntity.setTiming(sendTime);

                timingPushService.save(timingPushEntity);
            } else {
                //3、如果是实时发送短信则将消息保存到Redis队列，判断短信模板类型，如果是验证码类型则将消息保存到高优先级队列TOPIC_HIGH_SMS，如果是其他类型则将消息保存到普通队列TOPIC_GENERAL_SMS

                if (templateEntity.getType() == TemplateType.VERIFICATION.getCode()) {
                    //您的验证码是${code}
                    redisTemplate.opsForList().leftPush("TOPIC_HIGH_SMS", request);
                } else {
                    //理财产品${product}
                    redisTemplate.opsForList().leftPush("TOPIC_GENERAL_SMS", request);
                }
            }
            receiveLogEntity.setStatus(1);
        } catch (Exception e) {
            receiveLogEntity.setStatus(0);
            receiveLogEntity.setError(ExceptionUtils.getErrorStackTrace(e));
        } finally {
            //4、保存短信接收日志到Mysql数据库
            receiveLogEntity.setPlatformId(platformEntity.getId());
            receiveLogEntity.setPlatformName(platformEntity.getName());
            receiveLogEntity.setBusiness(smsSendDTO.getBatchCode());
            receiveLogEntity.setConfigIds(StringUtils.join(smsSendDTO.getConfigIds(), ",")); //[1,2,3]--->1,2,3
            receiveLogEntity.setTemplate(smsSendDTO.getTemplate());
            receiveLogEntity.setSignature(smsSendDTO.getSignature());
            receiveLogEntity.setMobile(smsSendDTO.getMobile());
            receiveLogEntity.setRequest(JSON.toJSONString(smsSendDTO.getParams()));
            receiveLogEntity.setUseTime(System.currentTimeMillis() - start);

            receiveLogMapper.insert(receiveLogEntity);
        }
    }

//业务逻辑
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        // TODO 短信接收服务：接收应用系统的报文并解析，调用Service将消息保存到消息缓冲区
        //就是和controller一样 调用service层即可
        String restMsg="success";
        log.info("tcp接口接受到消息:"+msg);
        try {
            //1sring msg--> SmsParamsDTO
            SmsParamsDTO smsParamsDTO = parseMessage(msg);
            if(null==smsParamsDTO){
                log.info("报文解析失败！");
                return;
            }
            //2调用service层即可
            SpringUtils.getBean(SmsSendService.class).send(smsParamsDTO);

        }catch (Exception e){
            log.error("netty发送时，报错了！",e);
            restMsg=e.getMessage();
        }

        log.info("回推报文============="+restMsg);
        ctx.writeAndFlush(restMsg+"\n"); //为什么要加\n 不加客户端接收不到消息
    }

/**
     * 通过HttpClient发送post请求，请求短信接收服务HTTP接口
     *
     * @param baseParamsDTO
     * @param url
     * @return
     */
    private R send(BaseParamsDTO baseParamsDTO, String url) {
        //设置此平台的秘钥
        baseParamsDTO.setAccessKeyId(accessKeyId);

        //电商 平台是否认证
        if (auth) {
            //key secret 都要有
            if (StringUtils.isBlank(accessKeyId) || StringUtils.isBlank(accessKeySecret)) {
                return R.fail("accessKeyId或accessKeySecret不能为空");
            }
        }

        baseParamsDTO.setTimestamp(System.currentTimeMillis() + "");
        baseParamsDTO.setEncryption(SmsEncryptionUtils.encode(baseParamsDTO.getTimestamp(), baseParamsDTO.getAccessKeyId(), accessKeySecret));

        if (StringUtils.isBlank(domain)) {
            return R.fail("domain不能为空");
        }

        //1 httpclien okhttp
        CloseableHttpClient httpClient = HttpClients.createDefault();
        //构造post请求
        HttpPost post = new HttpPost(url);
        //设置请求头
        post.setHeader("Content-Type", "application/json; charset=UTF-8");
        //设置请求体
        StringEntity stringEntity = new StringEntity(JSON.toJSONString(baseParamsDTO), "UTF-8");
        post.setEntity(stringEntity);

        try {
            //发送post请求
            CloseableHttpResponse response = httpClient.execute(post);
            HttpEntity responseEntity = response.getEntity();

            //解析响应码
            if (response.getStatusLine().getStatusCode() == 200) {
                //发送成功
                log.info("httpRequest access success, StatusCode is:{}", response.getStatusLine().getStatusCode());
                String responseEntityStr = EntityUtils.toString(responseEntity);
                log.info("responseContent is :" + responseEntityStr);

                return JSON.parseObject(responseEntityStr, R.class);
            } else {
                //发送失败
                log.error("httpRequest access fail ,StatusCode is:{}", response.getStatusLine().getStatusCode());
                return R.fail("status is " + response.getStatusLine().getStatusCode());
            }

        } catch (Exception e) {
            log.error("error", e);
            return R.fail(e.getMessage());
        } finally {
            post.releaseConnection();
        }
    }

package com.ydl.test;

import com.itheima.sms.SmsManageApplication;
import com.itheima.sms.sms.dto.SmsParamsDTO;
import com.itheima.sms.sms.service.SmsSendService;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.HashMap;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = SmsManageApplication.class)
public class SdkTest {
    @Autowired
    private SmsSendService smsSendService;

    /**
     * 通过SDK方式调用短信接收服务
     */
    @Test
    public void testSend(){
         SmsParamsDTO dto = new SmsParamsDTO();
        dto.setMobile("15718888888");
        dto.setSignature("DXQM_000000001");
        dto.setTemplate("DXMB_000000001");
        Map<String, String> map = new HashMap<>();
        map.put("code","79857");
        dto.setParams(map);
        dto.setSendTime("2021-12-18 10:00");
        dto.setTimestamp(System.currentTimeMillis() +"");

        return smsSendService.sendSms(dto);
    }
}

@Autowired
   RedisLock redisLock;

   //使用redis分布式锁，无阻塞
   @GetMapping("/stock3")
   public String stock3() {
       String mylock = redisLock.tryLock("MYLOCK", 2000);
       if(mylock!=null){
           int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
           if (stock > 0) {
               stock--;
               //放回redis
               stringRedisTemplate.opsForValue().set("stock", stock + "");
               System.out.println("减库存成功，剩余库存是：" + stock);
           } else {
               System.out.println("库存不足了！");
           }

           redisLock.unlock("MYLOCK",mylock);
       }
       return "OK";
   }

   //使用redis分布式锁，有阻塞
   @GetMapping("/stock4")
   public String stock4() {
       String mylock = redisLock.lock("MYLOCK", 2000,1000);
       if(mylock!=null){
           int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
           if (stock > 0) {
               stock--;
               //放回redis
               stringRedisTemplate.opsForValue().set("stock", stock + "");
               System.out.println("减库存成功，剩余库存是：" + stock);
           } else {
               System.out.println("库存不足了！");
           }

           redisLock.unlock("MYLOCK",mylock);
       }
       return "OK";
   }

package com.itheima.job;

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Date;

@Component
public class DemoJob {


    @Scheduled(cron = "0/5 * * * * ?")
    public void testJob() {
        System.out.println("现在时间是" + new Date());
    }

}

package com.ydl.sms.factory;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * 服务注册器，将短信发送服务注册到Redis中，定时服务上报，定时服务检查
 */
@Component
@Slf4j
@Order(value = 100)
public class ServerRegister implements CommandLineRunner {
    //当前服务实例的唯一标识，可以使用UUID随机生成
    public static String SERVER_ID = null;

    @Autowired
    private RedisTemplate redisTemplate;


    /**
     * 项目启动时自动执行此方法，将当前服务实例注册到redis
     *
     * @param args
     */
    @Override
    public void run(String... args) {
        //TODO 服务注册器，项目启动时将当前服务id注册到Redis中，使用Redis的Hash结构，key为SERVER_ID_HASH，Hash结构的key为服务id，value为时间戳
        // SERVER_ID_HASH   key  value
        SERVER_ID = UUID.randomUUID().toString(); //当前实例key
        log.info("server实例启动，生成的id:{}", SERVER_ID);
        redisTemplate.opsForHash().put("SERVER_ID_HASH", SERVER_ID, System.currentTimeMillis());
    }

    /**
     * 定时服务报告
     * 报告服务信息证明服务存在 每三分钟报告一次，并传入当前时间戳
     */
    @Scheduled(cron = "1 1/3 * * * ?")
    public void serverReport() {
        //TODO 服务注册器，每三分钟报告一次，并传入当前时间戳

        log.info("服务定时上报。id:{}",SERVER_ID);
        redisTemplate.opsForHash().put("SERVER_ID_HASH", SERVER_ID, System.currentTimeMillis());
    }

    /**
     * 定时服务检查
     * 每十分钟检查一次服务列表，清空超过五分钟没有报告的服务
     */
    @Scheduled(cron = "30 1/10 * * * ?")
    public void checkServer() {
        //TODO 服务注册器，定时检查redis,每隔10分钟查看，超过5分钟还没上报自己信息的服务，清除掉
        log.info("定时服务检查。id:{}",SERVER_ID);
        Map map = redisTemplate.opsForHash().entries("SERVER_ID_HASH");
        log.info("当前服务有："+map);

        long now = System.currentTimeMillis();
        List removeKeys=new ArrayList(); //该要删除的key
        //jdk8 lambda表达式
        map.forEach((key,value)->{
            long registrTime = Long.parseLong(value.toString());
            if(now-registrTime>(5*60*1000)){
                removeKeys.add(key);
            }
        });

        log.info("该要删除的key有：{}",removeKeys);
        removeKeys.forEach(key->{
            redisTemplate.opsForHash().delete("SERVER_ID_HASH", key);
        });

    }
}

/**
     * 根据通道配置，初始化每个通道的bean对象
     */
    @SneakyThrows
    public void initConnect() {
        //TODO 根据通道配置，初始化每个通道的bean对象
        //1、查询数据库获得通道列表
        List<ConfigEntity> configEntitiesList = configService.listForConnect();
        List constructorList=new ArrayList();
        //2、遍历通道列表，通过反射创建每个通道的Bean对象（例如AliyunSmsService、MengWangSmsService等）
        configEntitiesList.forEach(configEntity->{
            try {
                SmsConfig config=new SmsConfig();
                config.setId(configEntity.getId());
                config.setDomain(configEntity.getDomain());
                config.setName(configEntity.getName());
                config.setPlatform(configEntity.getPlatform().trim());
                config.setAccessKeyId(configEntity.getAccessKeyId().trim());
                config.setAccessKeySecret(configEntity.getAccessKeySecret().trim());
                if(StringUtils.isNotBlank(configEntity.getOther())){
                    LinkedHashMap linkedHashMap = JSON.parseObject(configEntity.getOther(), LinkedHashMap.class);
                    config.setOtherConfig(linkedHashMap);
                }

                //反射 创建Service
                //全限定名 com.ydl.sms.sms.AliyunSmsService
                String className="com.ydl.sms.sms."+config.getPlatform()+"SmsService";
                System.out.println(className);

                Class<?> aClass = Class.forName(className); //加载类
                Constructor<?> constructor = aClass.getConstructor(SmsConfig.class);//拿到具体的构造方法
                Object obj = constructor.newInstance(config); //创建对象

                //从容器中获取签名和模板的service
                SignatureService signatureService = SpringUtils.getBean(SignatureService.class);
                TemplateService templateService = SpringUtils.getBean(TemplateService.class);
                //找到这两个service在父类中的属性
                Field signatureServiceField = aClass.getSuperclass().getDeclaredField("signatureService");
                Field templateServiceField = aClass.getSuperclass().getDeclaredField("templateService");
                //打开访问权限
                signatureServiceField.setAccessible(true);
                templateServiceField.setAccessible(true);
                //设置属性值了
                signatureServiceField.set(obj,signatureService);
                templateServiceField.set(obj,templateService);

                constructorList.add(obj);

                log.info("初始化通道成功：{}，{}",config.getName(),config.getPlatform());
            }catch (Exception e){
                log.warn("初始化通道异常：{}",e.getMessage());
            }

        });
        //3、将每个通道的Bean对象保存到CONNECT_LIST集合中
        if(!CONNECT_LIST.isEmpty()){
            CONNECT_LIST.clear();
        }
        CONNECT_LIST.addAll(constructorList);

        //解锁逻辑
        if(StringUtils.isNotBlank(BUILD_NEW_CONNECT_TOKEN)){
            redisLock.unlock("buildNewConnect", BUILD_NEW_CONNECT_TOKEN);
        }

        log.info("通道初始化完成了。{}",CONNECT_LIST);
    }

package com.ydl.sms.job;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.ydl.sms.entity.TimingPushEntity;
import com.ydl.sms.factory.SmsFactory;
import com.ydl.sms.mapper.TimingPushMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;
/**
 * 定时短信业务处理器
 */
@Component
@Slf4j
public class SendTimingSmsImpl implements SendTimingSms {

    @Autowired
    private TimingPushMapper timingPushMapper;

    @Autowired
    private SmsFactory smsFactory;

    /**
     * 发送定时短信
     * @param timing
     */
    @Override
    @Async
    public void execute(String timing) {//timing格式：yyyy-MM-dd HH:mm  2021-12-25 18:00
        //TODO 查询数据库获取本次需要发送的定时短信，调用短信工厂发送短信
        //1、查询数据库获取本次需要发送的定时短信
        LambdaQueryWrapper<TimingPushEntity> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(TimingPushEntity::getStatus,0);
        wrapper.eq(TimingPushEntity::getTiming,timing);
        wrapper.orderByAsc(TimingPushEntity::getCreateTime);
        List<TimingPushEntity> list = timingPushMapper.selectList(wrapper);

        log.info("这一批次要发的短信条数是：{}，{}",timing,list.size());

        list.forEach(x->{
            //2、调用短信工厂发送短信
            String request = x.getRequest();
            smsFactory.send(request);
            //3、更新短信发送状态为“已处理”
            x.setStatus(1);
            x.setUpdateTime(LocalDateTime.now());
            x.setUpdateUser("system");
            timingPushMapper.updateById(x);
        });

        log.info("任务执行完毕"+timing);

    }
}

package com.ydl.sms.job;

import com.ydl.sms.config.RedisLock;
import com.ydl.utils.DateUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.Date;

/**
 * 定时任务，用于发送定时短信
 */
@Component
@Slf4j
public class SendSmsJob {

    @Autowired
    private SendTimingSms sendTimingSms;

    @Autowired
    private RedisLock redisLock;

    /**
     * 每分钟检查一次是否有定时短信需要发送
     * @throws InterruptedException
     */
    //1、每分钟触发一次定时任务
    @Scheduled(cron = "10 0/1 * * * ?")     //每分钟的第10秒执行一次
    public void sendTimingSms() throws InterruptedException {
        //TODO 定时任务，每分钟检查一次是否有定时短信需要发送
        //1、每分钟触发一次定时任务
        //2、为了防止短信重复发送，需要使用分布式锁
        String lock = redisLock.tryLock("SEND_TIMING_SMS", 30 * 1000);
        if(StringUtils.isNotBlank(lock)){
            //3、调用SendTimingSmsImpl发送定时短信
            sendTimingSms.execute(DateUtils.format(new Date(),"yyyy-MM-dd HH:mm"));
        }
    }
}

package com.ydl.sms.redismq;

import com.ydl.sms.factory.SmsFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.TimeUnit;

/**
 * Redis队列-----消费者
 * 监听消息队列：TOPIC_HIGH_SMS，高优先级的短信，如验证码之类的短信
 */
@Component
@Slf4j
public class HighSmsListener extends Thread {
    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private SmsFactory smsFactory;

    private String queueKey = "TOPIC_HIGH_SMS";

    @Value("${spring.redis.queue.pop.timeout}")
    private Long popTimeout = 8000L;

    private ListOperations listOps;

    @PostConstruct
    private void init() {
        listOps = redisTemplate.opsForList();
        this.start();
    }

    @Override
    public void run() {
        //TODO 监听TOPIC_HIGH_SMS队列，如果有消息则调用短信发送工厂发送实时短信
        //监听    TOPIC_HIGH_SMS 发送
        while (true){
            log.debug("队列{}正在监听中",queueKey);
            //SmsSendDTO -->string
            String message = (String) listOps.rightPop(queueKey, popTimeout, TimeUnit.MILLISECONDS);
            if(StringUtils.isNotBlank(message)){
                log.info("{}收到消息了：{}",queueKey,message);
                //发送
                smsFactory.send(message);
            }
        }
    }
}

package com.ydl.sms.redismq;

import com.ydl.sms.factory.SmsFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.TimeUnit;

/**
 * Redis队列消费者，监听消息队列TOPIC_GENERAL_SMS，普通优先级的短信，如营销短信
 */
@Component
@Slf4j
public class GeneralSmsListener extends Thread {
    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private SmsFactory smsFactory;

    private String queueKey = "TOPIC_GENERAL_SMS";

    @Value("${spring.redis.queue.pop.timeout}")
    private Long popTimeout = 8000L;

    private ListOperations listOps;

    @PostConstruct
    private void init() {
        listOps = redisTemplate.opsForList();
        this.start();
    }

    @Override
    public void run() {
        //TODO 监听TOPIC_GENERAL_SMS队列，如果有消息则调用短信发送工厂发送实时短信
        //监听    TOPIC_GENERAL_SMS 发送
        while (true){
            log.debug("队列{}正在监听中",queueKey);
            //SmsSendDTO -->string
            String message = (String) listOps.rightPop(queueKey, popTimeout, TimeUnit.MILLISECONDS);
            if(StringUtils.isNotBlank(message)){
                log.info("{}收到消息了：{}",queueKey,message);
                //发送
                smsFactory.send(message);
            }
        }
    }
}

package com.ydl.sms.redismq;

import com.ydl.sms.factory.SmsConnectLoader;
import com.ydl.sms.model.ServerTopic;
import lombok.extern.slf4j.Slf4j;
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.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

/**
 * Redis发布订阅----订阅者，通过Redis的发布订阅模式监听TOPIC_HIGH_SERVER频道
 */
@Component
@Slf4j
public class HighServerReceiver implements MessageListener {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private SmsConnectLoader smsConnectLoader;

    /**
     * 消息监听
     * @param message
     * @param pattern
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        //TODO 消息监听，根据消息内容调用smsConnectLoader进行通道初始化或者通道更新

        RedisSerializer<?> valueSerializer = redisTemplate.getDefaultSerializer();
        String deserialize = valueSerializer.deserialize(message.getBody()).toString();

        ServerTopic serverTopic = ServerTopic.load(deserialize);

        switch (serverTopic.getOption()) {
            case ServerTopic.USE_NEW_CONNECT:
                log.info("服务：{} 发起【通道更新】", serverTopic.getValue());
                smsConnectLoader.changeNewConnect();
                break;
            case ServerTopic.INIT_CONNECT:
                log.info("服务：{} 发起【通道初始化】", serverTopic.getValue());
                smsConnectLoader.initConnect();
                break;
            default:
                break;
        }

    }
}