推送服务完善

admin

2021-06-26 6b2670dfa68af9ce2e36a5f9580125f4fc6da570

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
89
90
91
92
package com.ks.push.consumer.mq;
 
import com.ks.push.dto.BPushDeviceDataSet;
import com.ks.push.manager.BPushPlatformAppInfoManager;
import com.ks.push.manager.CMQManager;
import com.ks.push.manager.PushExcuteResultManager;
import com.ks.push.pojo.DO.BPushPlatformAppInfo;
import com.ks.push.pojo.DO.BPushTask;
import com.ks.push.pojo.DO.PushPlatform;
import com.ks.push.service.BPushTaskService;
import com.ks.push.utils.PushUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
 
import javax.annotation.Resource;
import java.util.List;
 
@Component
public class PushTaskConsumer {
 
    Logger logger = LoggerFactory.getLogger(PushTaskConsumer.class);
 
 
    @Resource
    private BPushTaskService bPushTaskService;
 
    @Resource
    private PushExcuteResultManager pushExcuteResultManager;
 
    @Resource
    private BPushPlatformAppInfoManager bPushPlatformAppInfoManager;
 
 
    /**
     * 消费任务消息
     *
     * @param platform
     */
    public void consumeMsg(PushPlatform platform) {
 
        if (platform == null) {
            return;
        }
        logger.info("消费消息开始:{}  Thread:{}", platform.name(),Thread.currentThread().getId());
        List<CMQManager.MQMsgConsumeResult> resultList = null;
        try {
            resultList = CMQManager.getInstance().consumePushQueue(platform, 1);
        } catch (Exception e) {
        }
        logger.info("消费消息结束:{} Thread:{}", platform.name(),Thread.currentThread().getId());
        if (resultList != null) {
            for (CMQManager.MQMsgConsumeResult result : resultList) {
                try {
                    BPushDeviceDataSet dataSet = (BPushDeviceDataSet) result.getData();
                    BPushTask task = bPushTaskService.getTask(dataSet.getTaskId());
                    if (task != null) {
                        if (task.getState() == BPushTask.STATE_PUSHING) {
                            BPushPlatformAppInfo platformAppInfo = bPushPlatformAppInfoManager.selectByAppCodeAndPlatform(task.getAppCode(), platform);
                            if (platformAppInfo != null) {
                                try {
                                    PushUtil.pushNotifyCation(platform, platformAppInfo.getPushAppInfo(), task.getMessage(), dataSet.getDeviceTokenList());
                                    logger.info("推送任务执行成功,taskId-{},batchId-{}", dataSet.getTaskId(), dataSet.getBatchId());
                                } catch (Exception e) {
                                    logger.error("推送出错：ttaskId-{},batchId-{}，错误原因：{}", task.getId(), dataSet.getBatchId(), e.getMessage());
                                }
                            }
                        } else {
                            if (task.getState() == BPushTask.STATE_PAUSED) {
                                //任务暂停，不删除消息等待下次消费
                                continue;
                            } else if (task.getState() == BPushTask.STATE_CANCELED) {
                                //任务取消，删除消息
                            }
                        }
                    } else {
                        logger.error("任务为空，taskId-{},batchId-{}", dataSet.getTaskId(), dataSet.getBatchId());
                    }
                    //删除消息
                    logger.info("推送任务执行完成,taskId-{},batchId-{}", dataSet.getTaskId(), dataSet.getBatchId());
                    CMQManager.getInstance().deleteMsg(result.getQueueName(), result.getReceiptHandle());
                    pushExcuteResultManager.batchPushFinish(dataSet, platform);
                } catch (Exception e) {
//                    e.printStackTrace();
                }
            }
 
        }
 
    }
 
}