CMQ改造为rabbitmq

admin

2024-10-16 bd885c7015446c6c0495d3299ef64068a4c9b30e

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

@@ -6,6 +6,8 @@
import com.ks.push.pojo.DO.BPushTask;
import com.ks.push.pojo.DO.BPushTaskExcuteResult;
import com.ks.push.pojo.DO.PushPlatform;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
@@ -16,11 +18,15 @@

@Component
public class PushExcuteResultManager {

    private Logger logger = LoggerFactory.getLogger(PushExcuteResultManager.class);

    @Resource
    private BPushTaskExcuteResultMapper bPushTaskExcuteResultMapper;


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

    @Resource
    private BPushTaskDao bPushTaskDao;
@@ -37,7 +43,6 @@
        result.setId(BPushTaskExcuteResult.createId(taskId, pushPlatform));
        //初始化
        bPushTaskExcuteResultMapper.insertSelective(result);

        String key = getRedisKey(taskId, pushPlatform);
        redisTemplate.delete(key);
        return result;
@@ -70,10 +75,12 @@
     */
    public void batchPushFinish(BPushDeviceDataSet dataSet, PushPlatform pushPlatform) {
        String key = getRedisKey(dataSet.getTaskId(), pushPlatform);
        redisTemplate.opsForSet().remove(key, dataSet.getBatchId());
        Long size = redisTemplate.opsForSet().size(key);
        redisTemplate.opsForSet().remove(key, dataSet.getBatchId());
        Long afterSize = redisTemplate.opsForSet().size(key);
        logger.info("{}推送剩余数量 taskId-{} size-{} afterSize-{}", pushPlatform.name(), dataSet.getTaskId(), size + "", afterSize + "");
        //判断是否推送完成
        if (size == null || size == 0L) {
        if (afterSize == null || afterSize == 0L) {
            //已经推送完了
            String id = BPushTaskExcuteResult.createId(dataSet.getTaskId(), pushPlatform);
            BPushTaskExcuteResult update = new BPushTaskExcuteResult();
@@ -92,6 +99,7 @@
                updateTask.setStateDesc("推送完成");
                updateTask.setUpdateTime(new Date());
                bPushTaskDao.updateSelective(updateTask);
                logger.info("{}推送完成 taskId-{}", pushPlatform.name(), dataSet.getTaskId());
            }
        }
    }