推送服务完善

admin

2021-09-17 a2c56bd6b79d2b8ca2c4c44a254ad2958fb72bca

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
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
package com.ks.push;
 
import com.ks.push.consumer.mq.PushTaskConsumer;
import com.ks.push.dto.mq.InvalidDeviceTokenInfo;
import com.ks.push.manager.CMQManager;
import com.ks.push.manager.PushDeviceTokenManager;
import com.ks.push.pojo.DO.BPushDeviceToken;
import com.ks.push.pojo.DO.PushPlatform;
import org.apache.dubbo.config.spring.context.annotation.EnableDubbo;
import org.mybatis.spring.annotation.MapperScan;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.yeshi.utils.mq.JobThreadExecutorServiceImpl;
 
import javax.annotation.Resource;
import java.util.List;
 
@SpringBootApplication
@EnableTransactionManagement
@MapperScan(basePackages = "com.ks.push.mapper")
@EnableDubbo(scanBasePackages = "com.ks.push.service.remote")
public class PushApplication implements ApplicationListener<ContextRefreshedEvent> {
    private final static Logger logger = LoggerFactory.getLogger(PushApplication.class);
 
    @Resource
    private PushTaskConsumer pushTaskConsumer;
 
    @Resource
    private PushDeviceTokenManager pushDeviceTokenManager;
 
    public static void main(String[] args) {
        SpringApplication.run(PushApplication.class, args);
        logger.info("应用启动成功");
    }
 
 
    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        logger.info("容器加载完毕");
        initMQMsgConsumer();
    }
 
    private void initMQMsgConsumer() {
        final int THREAD_NUM = 3;
        for (PushPlatform pushPlatform : PushPlatform.values()) {
            //创建三条队列处理
            for (int i = 0; i < THREAD_NUM; i++) {
                new JobThreadExecutorServiceImpl().run(new Runnable() {
                    @Override
                    public void run() {
                        pushTaskConsumer.consumeMsg(pushPlatform);
                    }
                });
            }
        }
        //清理无效token
        new JobThreadExecutorServiceImpl().run(new Runnable() {
            @Override
            public void run() {
                try {
                    List<CMQManager.MQMsgConsumeResult> list =  CMQManager.getInstance().consumeInvalidDeviceTokenQueue(16);
                    if (list != null) {
                        logger.info("清理无效token数量：" + list.size());
                        for (CMQManager.MQMsgConsumeResult result : list) {
                            InvalidDeviceTokenInfo tokenInfo = (InvalidDeviceTokenInfo) result.getData();
                            List<BPushDeviceToken> tokenList = pushDeviceTokenManager.list(tokenInfo.getAppCode(), tokenInfo.getPushPlatform(), tokenInfo.getToken(), 1, 20);
                            if (tokenList != null) {
                                for (BPushDeviceToken token : tokenList) {
                                    logger.info("清理无效token id-{}" + token.getId());
                                    //删除
                                    pushDeviceTokenManager.deleteByPrimaryKey(token.getId());
                                }
                            }
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
 
}