add 分发 master_author 添加字段 subscribe increment， 添加7个表 master_author...

kimi

2020-04-22 5268a2b7dfa556bd6f5a2d5e446cea3ea9940c10

 src/main/java/com/highdatas/mdm/util/pool/MqMessage.java

@@ -1,14 +1,18 @@
package com.highdatas.mdm.util.pool;



import com.alibaba.fastjson.JSONObject;

import com.highdatas.mdm.pojo.CodeMsg;

import com.highdatas.mdm.entity.*;

import com.highdatas.mdm.pojo.Page;

import com.highdatas.mdm.pojo.Result;

import com.highdatas.mdm.util.Constant;

import com.highdatas.mdm.util.DbUtils;

import lombok.extern.slf4j.Slf4j;

import org.springframework.http.MediaType;



import javax.servlet.http.HttpServletResponse;

import java.io.IOException;

import java.io.PrintWriter;

import java.util.Date;

import java.util.List;

import java.util.Map;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.stream.Collectors;



/**

 * @author kimi

@@ -18,23 +22,161 @@


@Slf4j

public class MqMessage extends PriorityTask{

    public MqMessage(String code, HttpServletResponse response) {

        super(code,response);





    public MqMessage(MqEntity mqEntity) {

        super(mqEntity);

    }



    @Override

    public void run() {

        try {

            log.info("code {} ---------- run", code);

            Thread.sleep(1000 * 30 * 1);

            response.setCharacterEncoding("UTF-8");

            response.setContentType(MediaType.APPLICATION_JSON_VALUE);

            PrintWriter writer = response.getWriter();

            writer.write(JSONObject.toJSONString(Result.success(CodeMsg.SUCCESS)));

        } catch (InterruptedException e) {

            e.printStackTrace();

        } catch (IOException e) {

            e.printStackTrace();

        log.info("mq: {} ---------- run", mqEntity.toString());

        String type = mqEntity.getType();

        try{

            if (type.equalsIgnoreCase(Constant.Master)) {

                sendMasterData();

            } else if (type.equalsIgnoreCase(Constant.View)){

                sendViewData();

            }

        }

        catch (Exception e) {

            e.printStackTrace();

            //log

        }



    }



    private void sendMasterData() throws Exception {

        AtomicInteger pageNo = mqEntity.getPageNo();

        AtomicInteger pageSize = mqEntity.getPageSize();

        AtomicInteger pages = mqEntity.getPages();

        String maintainId = mqEntity.getMaintainId();

        String userId = mqEntity.getUserId();

        TUser user = new TUser().setUserId(userId);

        Maintain maintain = DbUtils.maintainService.selectById(maintainId);

        long totalCnt = 0;



        if (pageSize == null) {

            MasterAuthor masterAuthor = new MasterAuthor().setTableName(maintain.getTableName());

            Page initPageInfo = DbUtils.masterAuthorService.getInitPageInfo(masterAuthor, maintain, user, mqEntity.getIncrement());

            pages = new AtomicInteger(initPageInfo.getPages());

            pageSize = new AtomicInteger(initPageInfo.getPageSize());

            mqEntity.setTotalSize(new AtomicInteger(Long.valueOf(initPageInfo.getRecordCount()).intValue()));

            mqEntity.setPages(pages);

            mqEntity.setPageSize(pageSize);

        }



        if (pageNo == null) {

            pageNo = new AtomicInteger(1);

            mqEntity.setPageNo(pageNo);

            for (pageNo.get(); pageNo.get() <= pages.get(); pageNo.addAndGet(1)){

                sendOneMasterPackage(pageNo, pageSize,totalCnt,  maintain, user);

            }

        } else {

            sendOneMasterPackage(pageNo, pageSize, totalCnt, maintain, user);

        }

    }



    private void sendOneMasterPackage(AtomicInteger pageNo, AtomicInteger pageSize, long totalCnt, Maintain maintain, TUser user) throws Exception {

        Boolean increment = mqEntity.getIncrement();

        List<SysField> fieldByMaintain = DbUtils.fieldService.getFieldByMaintain(maintain.getId());

        List<String> fieldList = fieldByMaintain.stream().map(sysField -> sysField.getField()).collect(Collectors.toList());

        String fields = fieldByMaintain.stream().map(sysField -> sysField.getField()).collect(Collectors.joining(Constant.COMMA));



        String filter = DbUtils.masterAuthorService.getFilter(user, maintain.getId());

        String content;

        if (increment) {

            String maintainTableName = maintain.getTableName();

            String tempTableName = maintainTableName + Constant.RECORD;

            Page page = new Page(totalCnt);

            page.setPageSize(mqEntity.getPageSize().get());

            page.setPageNo(mqEntity.getPageNo().get());

            List<Map<String, Object>> maps = DbUtils.maintainDetailMapper.selectMaintainDetail(fields, tempTableName, maintain.getId(), filter, page.getLimitSQL());

            JSONObject object = new JSONObject();

            object.fluentPut("total", page.getRecordCount());

            object.fluentPut("size", page.getPageSize());

            object.fluentPut("pages", page.getPageCount());

            object.fluentPut("current", page.getPageNo());

            object.fluentPut("records", maps);

            content = JSONObject.toJSONString(object);

        } else {

            Result result = DbUtils.masterDataService.selectListByPageByVersion(user, maintain.getTableName(), fieldList, filter, pageNo.get(), pageSize.get(), maintain.getVersion(), false);

            JSONObject data = (JSONObject) result.getData();

            JSONObject grid = data.getJSONObject("grid");

            JSONObject object = new JSONObject();

            object.fluentPut("total", grid.get("total"));

            object.fluentPut("size", grid.get("size"));

            object.fluentPut("pages", grid.get("pages"));

            object.fluentPut("current", grid.get("current"));

            object.fluentPut("records", grid.get("record"));



            content = JSONObject.toJSONString(object);



        }

        sendMqBody(pageSize.get(), content);

    }



    private void sendViewData() throws Exception {

        //logs

        String dataId = mqEntity.getDataId();

        AtomicInteger pageNo = mqEntity.getPageNo();

        AtomicInteger pageSize = mqEntity.getPageSize();

        AtomicInteger pages = mqEntity.getPages();

        SysView sysView = DbUtils.viewService.selectById(dataId);

        ViewStatus status = sysView.getStatus();

        if (!ViewStatus.working.equals(status)) {

            return;

        }

        if (pageSize == null) {

            Page initPageInfo = DbUtils.viewService.getInitPageInfo(sysView.getId());

            pages = new AtomicInteger(initPageInfo.getPages());

            pageSize = new AtomicInteger(initPageInfo.getPageSize());

            mqEntity.setPages(pages);

            mqEntity.setPageSize(pageSize);

            mqEntity.setTotalSize(new AtomicInteger(Long.valueOf(initPageInfo.getRecordCount()).intValue()));

        }

        if (pageNo == null) {

            //主动

            pageNo =new AtomicInteger(1);

            mqEntity.setPageNo(pageNo);

            for (pageNo.get(); pageNo.get() <= pages.get(); pageNo.addAndGet(1)) {

                sendOneViewPackage(pageNo, pageSize, sysView);

            }



        } else {

            //被动

            sendOneViewPackage(pageNo, pageSize, sysView);

        }



    }



    private void sendOneViewPackage(AtomicInteger pageNo, AtomicInteger pageSize, SysView sysView) throws Exception {

        Result viewData = DbUtils.viewService.getViewData(sysView, pageNo.get(), pageSize.get());

        JSONObject object = (JSONObject) viewData.getData();

        String content = JSONObject.toJSONString(object);

        sendMqBody(pageSize.get(), content);

    }



    private void sendMqBody(Integer pageSize, String content) throws Exception {

        byte[] bytes = content.getBytes("UTF-8");

        String size = (bytes.length / 1024) + "k";

        log.info(size);

        Date now = new Date();



//        String mqBody = com.datacvg.common.AESUtil.getInstance().encoder(content, mqEntity.getUserId(),"http://180.169.94.250:9107");

        String mqBody = content;

        bytes = mqBody.getBytes("UTF-8");

        size = (bytes.length / 1024) + "k";

        log.info(size);

        Date after = new Date();

        long l = after.getTime() - now.getTime();

        log.info("time:"+l);

//        String mqBody = content;

        MqEntity.MsgBodyBean msgBody = mqEntity.getMsgBody();

        msgBody.setData(mqBody);

        msgBody.setTotal(pageSize);

        mqEntity.setMsgBody(msgBody);

        Thread.sleep(1000 * 60 * 3);

        mqEntity.send2Mq();

    }

}