CMQ转为Rabbitmq

admin

2024-10-17 b30fb8afd3cd6228bda9b182dc412bb3c8daf69c

 src/main/java/com/yeshi/buwan/service/imp/ClearService.java

@@ -1,30 +1,71 @@
package com.yeshi.buwan.service.imp;

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;

import javax.annotation.Resource;

import com.yeshi.buwan.dao.CategoryVideoDao;
import com.yeshi.buwan.dao.VideoInfoDao;
import com.yeshi.buwan.domain.VideoDetailInfo;
import com.yeshi.buwan.dto.mq.VideoDataChangeMQMsg;
import com.yeshi.buwan.service.imp.juhe.IqiyiService;
import com.yeshi.buwan.util.StringUtil;
import com.yeshi.buwan.util.TimeUtil;
import com.yeshi.buwan.util.mq.rabbit.RabbitmqManager;
import net.sf.json.JSONObject;
import org.hibernate.HibernateException;
import org.hibernate.Session;
import org.springframework.orm.hibernate4.HibernateCallback;
import org.springframework.stereotype.Service;

import com.fasterxml.jackson.databind.util.BeanUtil;
import com.yeshi.buwan.dao.VideoInfoDao;
import com.yeshi.buwan.domain.VideoDetailInfo;
import com.yeshi.buwan.service.imp.juhe.IqiyiService;
import com.yeshi.buwan.util.StringUtil;
import com.yeshi.buwan.util.TimeUtil;

import net.sf.json.JSONObject;
import javax.annotation.Resource;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

//清除数据库中的死数据
@Service
public class ClearService {


    private final String[] videoTables = new String[]{"wk_category_video", "wk_resource_video", "wk_video_attention",
            "wk_video_banquan_video", "wk_video_banquan_webvideo", "wk_video_collection", "wk_video_comment",
            "wk_video_comment2", "wk_video_getscore_collect", "wk_video_homead", "wk_video_homevideo",
            "wk_video_hotstar_video", "wk_video_iqiyi", "wk_video_letv", "wk_video_play_statistics",
            "wk_video_pptv", "wk_video_sohu", "wk_video_special_video", "wk_video_youku", "wk_video_youku_dj"};

    @Resource
    private VideoInfoDao videoInfoDao;

    @Resource
    private CategoryVideoDao categoryVideoDao;

    @Resource
    private RabbitmqManager rabbitmqManager;


    /**
     * @return void
     * @author hxh
     * @description 删除老旧数据
     * @date 16:49 2022/3/18
     * @param: typeId
     * @param: leftNumber 留下的数量
     **/
    public void clearOldVideo(long typeId, int leftNumber) {
        Long minCreateTime = categoryVideoDao.getHibernateTemplate().execute(new HibernateCallback<Long>() {
            @Override
            public Long doInHibernate(Session session) throws HibernateException {
                Object result = session.createSQLQuery("SELECT v.`createtime` FROM wk_category_video cv  LEFT JOIN wk_video_video v ON v.`id`=cv.`videoid` WHERE cv.`videotypeid`=? ORDER BY cv.`id` DESC LIMIT ?,1").setParameter(0, typeId).setParameter(1, leftNumber).uniqueResult();
                if (result == null) {
                    return null;
                }
                return Long.parseLong(result.toString());
            }
        });
        if (minCreateTime == null) {
            return;
        }
        clearVideos(new long[]{typeId}, TimeUtil.getGernalTime(minCreateTime, "yyyy-MM-dd"));
    }


    @SuppressWarnings("rawtypes")
    public void clearVideos(long[] ls, String date) {
@@ -40,31 +81,57 @@
        videoInfoDao.excute(new HibernateCallback() {
            public Object doInHibernate(Session session) throws HibernateException {
                for (JSONObject item : list) {
                    List list = session
                    Object result = session
                            .createSQLQuery(String.format(
                                    "SELECT cv.`videoid` FROM wk_category_video cv LEFT JOIN wk_video_video v ON v.`id`=cv.`videoid` WHERE cv.`videotypeid`=%s  AND v.`createtime`<%s",
                                    "SELECT count(*) FROM wk_category_video cv LEFT JOIN wk_video_video v ON v.`id`=cv.`videoid` WHERE cv.`videotypeid`=%s  AND v.`createtime`<%s",
                                    item.getLong("videotype"), item.getLong("time")))
                            .setFirstResult(0).setMaxResults(1000).list();
                    session.getTransaction().begin();
                    if (list != null) {
                        int pagecount = 50;
                        int pages = list.size() % pagecount == 0 ? list.size() / pagecount
                                : list.size() / pagecount + 1;
                        for (int i = 0; i < pages; i++) {
                            String where = "";
                            for (int p = 0; p < pagecount; p++) {
                                int pos = i * pagecount + p;
                                pos = pos >= list.size() ? list.size() - 1 : pos;
                                where += String.format("id=%s or ", list.get(pos) + "");
                            }
                            if (where.endsWith("or "))
                                where = where.substring(0, where.length() - 3);

                            session.createSQLQuery("delete from wk_video_video where " + where).executeUpdate();
                        }
                            .uniqueResult();
                    int totalCount = Integer.parseInt(result.toString());
                    if (totalCount <= 0) {
                        return null;
                    }
                    session.flush();
                    session.getTransaction().commit();
                    int BETCH_SIZE = 1000;
                    int page = totalCount / BETCH_SIZE + 1;
                    for (int n = 0; n < page; n++) {
                        List list = session
                                .createSQLQuery(String.format(
                                        "SELECT cv.id,cv.videoid FROM wk_category_video cv LEFT JOIN wk_video_video v ON v.`id`=cv.`videoid` WHERE cv.`videotypeid`=%s  AND v.`createtime`<%s",
                                        item.getLong("videotype"), item.getLong("time")))
                                .setFirstResult(0).setMaxResults(BETCH_SIZE).list();
                        session.getTransaction().begin();
                        if (list != null) {
                            int pagecount = 50;
                            int pages = list.size() % pagecount == 0 ? list.size() / pagecount
                                    : list.size() / pagecount + 1;
                            for (int i = 0; i < pages; i++) {
                                String ors = "";
                                String cvors = "";

                                List<String> videoIds = new ArrayList<>();
                                for (int p = 0; p < pagecount; p++) {
                                    int pos = i * pagecount + p;
                                    pos = pos >= list.size() ? list.size() - 1 : pos;
                                    ors += String.format("id=%s or ", ((Object[]) list.get(pos))[1] + "");
                                    cvors += String.format("id=%s or ", ((Object[]) list.get(pos))[0] + "");
                                    videoIds.add(((Object[]) list.get(pos))[1] + "");
                                }
                                if (ors.endsWith("or "))
                                    ors = ors.substring(0, ors.length() - 3);

                                if (cvors.endsWith("or "))
                                    cvors = cvors.substring(0, cvors.length() - 3);

                                session.createSQLQuery("delete from wk_video_video where " + ors).executeUpdate();
                                session.createSQLQuery("delete from wk_category_video where " + cvors).executeUpdate();
                                if (videoIds.size() > 0) {
                                    rabbitmqManager.addVideoDataChanged(new VideoDataChangeMQMsg(VideoDataChangeMQMsg.TYPE_VIDEO_INFO, org.yeshi.utils.StringUtil.concat(videoIds, ","), VideoDataChangeMQMsg.ACTION_DELETE));
                                }

                            }
                        }
                        session.flush();
                        session.getTransaction().commit();
                    }
                }
                return null;
            }
@@ -75,19 +142,33 @@
    // 清楚依赖VideoInfo的数据
    @SuppressWarnings("rawtypes")
    public void clearDependVideo() {
        String[] tables = new String[]{"wk_category_video", "wk_resource_video", "wk_video_attention",
                "wk_video_banquan_video", "wk_video_banquan_webvideo", "wk_video_collection", "wk_video_comment",
                "wk_video_comment2", "wk_video_getscore_collect", "wk_video_homead", "wk_video_homevideo",
                "wk_video_hotstar_video", "wk_video_iqiyi", "wk_video_letv", "wk_video_play_statistics",
                "wk_video_pptv", "wk_video_sohu", "wk_video_special_video", "wk_video_youku", "wk_video_youku_dj"};
        for (String table : videoTables) {
            clearDependVideoByTable(table);
        }
    }

        for (String table : tables) {
            final String sql = String.format(
                    "SELECT vv.id FROM `%s` vv LEFT JOIN wk_video_video v ON v.`id`=vv.`videoid` WHERE v.`id` IS NULL",
                    table);
            Object obj = videoInfoDao.excute(new HibernateCallback() {

    public void clearDependVideoByTable(String table) {
        final String sql = String.format(
                "SELECT vv.id FROM `%s` vv LEFT JOIN wk_video_video v ON v.`id`=vv.`videoid` WHERE v.`id` IS NULL",
                table);
        final String countSql = String.format(
                "SELECT count(*) FROM `%s` vv LEFT JOIN wk_video_video v ON v.`id`=vv.`videoid` WHERE v.`id` IS NULL",
                table);

        Object obj = videoInfoDao.excute(new HibernateCallback() {
            public Object doInHibernate(Session session) throws HibernateException {
                return session.createSQLQuery(countSql).uniqueResult();
            }
        });
        int count = Integer.parseInt(obj.toString());
        int pageSize = 1000;
        int page = count / pageSize + 1;

        for (int p = 0; p < page; p++) {
            obj = videoInfoDao.excute(new HibernateCallback() {
                public Object doInHibernate(Session session) throws HibernateException {
                    return session.createSQLQuery(sql).setFirstResult(0).setMaxResults(1000).list();
                    return session.createSQLQuery(sql).setFirstResult(0).setMaxResults(pageSize).list();
                }
            });

@@ -121,11 +202,40 @@
                }

            }

        }
    }

    /**
     * 清理SQL数据库依赖视频
     *
     * @param videoId
     */
    public void clearDependVideo(final String videoId) {
        clearDependVideo(Arrays.asList(new String[]{videoId}));
    }


    public void clearDependVideo(final List<String> videoIds) {
        videoInfoDao.excute(new HibernateCallback() {
            public Object doInHibernate(Session session) throws HibernateException {
                session.getTransaction().begin();
                List<String> ors = new ArrayList<>();
                for (String videoId : videoIds) {
                    ors.add("videoid=" + videoId);
                }

                for (String table : videoTables)
                    session.createSQLQuery(
                            String.format("delete from %s where %s", table, org.yeshi.utils.StringUtil.concat(ors, " or ")))
                            .executeUpdate();
                session.flush();
                session.getTransaction().commit();
                return null;
            }
        });
    }

    /**
     * 清除无效的爱奇艺视频
     */
    @SuppressWarnings("unchecked")