学成在线项目开发技巧整理---第二部分

#访问动态资源时,将请求负载均衡到多个服务器实例或者多个网关实例
upstream webservice{
   server 192.168.200.146:8080;
}

server {
        listen       80;
        server_name  localhost;

        #动态资源
        location /demo {
                proxy_pass http://webservice;
        }
        #静态资源
        location ~/.*\.(png|jpg|gif|js){
                root html/web;
                gzip on;
        }

        location / {
            root   html/web;
            index  index.html index.htm;
        }
}

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;

DROP TABLE IF EXISTS `transaction_message`;

CREATE TABLE `transaction_message` (
    `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键',
    `message_type` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '消息类型',
    `business_key1` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci COMMENT '关联业务信息,如: 用户ID',
    `business_key2` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci COMMENT '关联业务信息,如: 订单信息',
    `business_key3` varchar(256) CHARACTER SET utf8 COLLATE utf8_general_ci COMMENT '关联业务信息,如: 更加复杂的信息',
    `retry_num` int NOT NULL DEFAULT '0' COMMENT '重试次数',
    `state` smallint NOT NULL DEFAULT '0' COMMENT '处理状态,0:初始,1:成功,2:失败',
    `stage_state` smallint NOT NULL DEFAULT '0' COMMENT '各个小阶段的处理状态',
    `return_failure_msg` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci COMMENT '失败原因',
    `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    `latest_execute_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '最近一次被消费的时间',
    PRIMARY KEY (id)
) ENGINE = InnoDB AUTO_INCREMENT = 0 CHARACTER SET = 'utf8' COLLATE = 'utf8_general_ci' ROW_FORMAT = compact COMMENT '事务型消息表';

SET FOREIGN_KEY_CHECKS = 1;

package dhy.com;

import java.util.List;
import java.util.Map;

/**
 * 事务型消息服务类
 * @author 大忽悠
 * @create 2023/1/27 19:48
 */
public interface ITMService {

    /**
     * @param messageType 要获取的消息类型,传入null表示获取所有消息类型
     * @param count  扫描记录数
     * @param examples 额外的查询条件
     * @return 返回获取到的事务消息列表
     */
    List<TransactionMessage> getTransactionMessageList(String messageType, int count,Map<Object,Object> examples);

    /**
     * @param tmId 事务消息ID
     * @return 事务消息
     */
    TransactionMessage getTransactionMessage(long tmId);

    /**
     * 事务型消息是否被成功消费
     * @param tmId 事务消息ID
     * @return 消费是否成功
     */
    boolean completed(long tmId);

    /**
     * 获取小阶段完成状态
     * @param tmId 事务消息ID
     * @param stageSeq 小阶段任务序号
     * @return 当前小阶段状态是否为完成
     */
    boolean getStage(long tmId,byte stageSeq);
}

package dhy.com;

import dhy.com.util.BitUtil;
import dhy.com.util.TMUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author 大忽悠
 * @create 2023/1/27 20:33
 */
@Slf4j
public abstract class AbstractTmMessageImpl implements ITMService{
    /**
     * 线程池大小默认为CPU核心数
     */
    private final ExecutorService THREAD_POOL= Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    /**
     * 默认超时等待时间
     */
    private final long TIME_OUT=1800;

    /**
     * 根据事务消息执行对应的任务
     * @param transactionMessage 事务消息
     * @return 此次事务消息是否处理成功
     */
    public abstract boolean execute(TransactionMessage transactionMessage);

    public void process(String messageType, int count, Map<Object,Object> examples){
        try{
            List<TransactionMessage> tmList = getTransactionMessageList(messageType, count, examples);
            log.info("扫描得到的事务消息个数为: {}",tmList.size());
            CountDownLatch countDownLatch = new CountDownLatch(tmList.size());
            THREAD_POOL.execute(()->{
                for (TransactionMessage tm : tmList) {
                    log.info("开始执行任务: {}",tm);
                    try {
                        boolean res = execute(tm);
                        clearAfterExecute(tm, res);
                    }catch (Throwable t){
                        log.error("任务执行出现异常: {},任务信息: {}",t.getMessage(),tm);
                        t.printStackTrace();
                    }
                    countDownLatch.countDown();
                }
            });
            //等待指定时间,防止无限等待
            countDownLatch.await(TIME_OUT, TimeUnit.SECONDS);
        } catch (Exception e){
           e.printStackTrace();
        }finally {
            THREAD_POOL.shutdown();
        }
    }

    private void clearAfterExecute(TransactionMessage tm, boolean res) {
        if(res &&completed(tm.getId())){
            doRemoveTm(tm.getId());
            //TODO: 将此次处理的事务消息写入事务消息处理历史表中,方便回溯操作
            log.info("任务执行成功: {}", tm);
        }else {
            //设置当前事务消息状态为失败
            tm.setState((byte) 2);
            doUpdateTmAfterFailure(tm);
            log.info("任务执行失败: {}", tm);
        }
    }

    /**
     * @param tm 将事务消息失败状态更新到数据库中
     */
    protected abstract void doUpdateTmAfterFailure(TransactionMessage tm);

    /**
     * @param id 删除某个被成功执行的tm
     */
    protected abstract void doRemoveTm(Long id);

    /**
     * 设置某个小阶段任务状态为完成
     *
     * @param tmId     事务消息ID
     * @param stageSeq 小阶段任务序号
     * @return 设置是否成功
     */
    protected boolean completedStage(long tmId, byte stageSeq) {
        TransactionMessage transactionMessage = getTransactionMessage(tmId);
        TMUtil.completedStage(transactionMessage,stageSeq);
        return doCompletedStage(transactionMessage);
    }

    protected abstract boolean doCompletedStage(TransactionMessage transactionMessage);


    /**
     * 事务型消息是否被成功消费
     *
     * @param tmId 事务消息ID
     * @return 消费是否成功
     */
    @Override
    public boolean completed(long tmId) {
        return getTransactionMessage(tmId).getState()==1;
    }

    /**
     * 获取小阶段完成状态
     *
     * @param tmId     事务消息ID
     * @param stageSeq 小阶段任务序号
     * @return 当前小阶段状态是否为完成
     */
    @Override
    public boolean getStage(long tmId, byte stageSeq) {
        return BitUtil.is1(getTransactionMessage(tmId).getStageState(),stageSeq);
    }
}

package dhy.com;

import lombok.Data;

import java.time.LocalDateTime;

/**
 * 事务消息实体类
 * @author zdh
 */
@Data
public class TransactionMessage {
    private Long id;
    private String messageType;
    private String businessKey1;
    private String businessKey2;
    private String businessKey3;
    private Integer retryNum;
    private Byte state;
    private Byte stageState;
    private String returnFailureMsg;
    private LocalDateTime createTime;
    private LocalDateTime latestExecuteTime;
}

package dhy.com.util;

/**
 * @author 大忽悠
 * @create 2023/1/27 20:17
 */
public class BitUtil {
    /**
     * 将x的第y位设置为1
     */
    public static byte setBit1(byte x,byte y){
        return x|=(1<<y);
    }

    /**
     * 将x的第y位设置为0
     */
    public static byte setBit0(byte x,byte y){
        return x&=~(1<<y);
    }

    /**
     * x的第index位是否为1
     */
    public static boolean is1(byte x,byte index){
        return (x>>(index-1) & 1)==1;
    }
}

package dhy.com.util;

import dhy.com.TransactionMessage;

/**
 * @author 大忽悠
 * @create 2023/1/27 20:02
 */
public class TMUtil {
    /**
     * 设置某个小阶段任务状态为完成
     * @param transactionMessage 事务消息
     * @param stage 小阶段任务序号
     */
    public static TransactionMessage completedStage(TransactionMessage transactionMessage, byte stage){
        transactionMessage.setStageState(BitUtil.setBit1(transactionMessage.getStageState(),stage));
        return transactionMessage;
    }

    /**
     * 判断某个小阶段任务是否完成
     * @param transactionMessage 事务消息
     * @param stage 小阶段任务序号
     * @return 小阶段任务是否完成
     */
    public static boolean getStage(TransactionMessage transactionMessage,byte stage){
        return BitUtil.is1(transactionMessage.getStageState(),stage);
    }
}