CMQ改造为rabbitmq

admin

2024-10-16 bd885c7015446c6c0495d3299ef64068a4c9b30e

 service-push/src/main/java/com/ks/push/manager/PushManager.java

@@ -4,6 +4,7 @@
import com.ks.push.dao.BPushTaskDao;
import com.ks.push.dto.BPushDeviceDataSet;
import com.ks.push.exception.BPushTaskException;
import com.ks.push.manager.rabbitmq.RabbitmqManager;
import com.ks.push.pojo.DO.BPushDeviceToken;
import com.ks.push.pojo.DO.BPushPlatformAppInfo;
import com.ks.push.pojo.DO.BPushTask;
@@ -36,10 +37,13 @@
    private BPushPlatformAppInfoManager bPushPlatformAppInfoManager;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    private PushExcuteResultManager pushExcuteResultManager;

    @Resource
    private PushExcuteResultManager pushExcuteResultManager;
    private RedisTemplate redisTemplate;

    @Resource
    private RabbitmqManager rabbitmqManager;


    /**
@@ -81,13 +85,19 @@
        }

        String key = "task-" + taskId;
        if (!redisTemplate.opsForValue().setIfAbsent(key, "1", 5 * 60, TimeUnit.SECONDS)) {

        Boolean s = redisTemplate.opsForValue().setIfAbsent(key, "1", 5 * 60, TimeUnit.SECONDS);

        if (s == null || !s) {
            throw new BPushTaskException(BPushTaskException.CODE_BUSY, "服务器繁忙，请稍后再试");
        }
        logger.info("开始启动推送#taskId:{}", task.getId());
        updateState(taskId, BPushTask.STATE_PUSHING, null);
        boolean hasDevice = false;


        try {
            logger.info("开始启动推送#taskId:{}", task.getId());
            updateState(taskId, BPushTask.STATE_PUSHING, null);
            boolean hasDevice = false;

            // 查询可推送的平台
            List<BPushPlatformAppInfo> list = bPushPlatformAppInfoManager.listByAppCode(task.getAppCode());

@@ -132,19 +142,18 @@
                        //修改总数
                        pushExcuteResultManager.setDeviceCount(result.getId(), totalValidCount);
                    }
                    CMQManager.getInstance().addToPushQueue(result.getPushPlatform(), dataSet);
                    rabbitmqManager.addToPushQueue(result.getPushPlatform(), dataSet);
                }
                logger.info("加入推送队列#任务Id:{}#平台:{}#推送数量:{}", task.getId(), result.getPushPlatform().name(), count);
            }

            if (!hasDevice) {
                updateState(taskId, BPushTask.STATE_FINSIH, "没有满足条件的可推送设备");
            }
        } finally {
            redisTemplate.delete(key);
        }

        //没有可推送的设备
        if (!hasDevice) {
            updateState(taskId, BPushTask.STATE_FINSIH, "没有满足条件的可推送设备");
        }

        logger.info("启动推送结束#taskId:{}", task.getId());
    }