【死磕Sharding-jdbc】---异步送达JOB

CONTAINER_MAIN=com.dangdang.ddframe.rdb.transaction.soft.bed.BestEffortsDeliveryJobMainnohup java -classpath $CONF_DIR:$LIB_DIR:. $CONTAINER_MAIN >/dev/null 2>&1 &

public final class BestEffortsDeliveryJobMain {    public static void main(final String[] args) throws Exception {        try (InputStreamReader inputStreamReader = new InputStreamReader(BestEffortsDeliveryJobMain.class.getResourceAsStream("/conf/config.yaml"), "UTF-8")) {            BestEffortsDeliveryConfiguration config = new Yaml(new Constructor(BestEffortsDeliveryConfiguration.class)).loadAs(inputStreamReader, BestEffortsDeliveryConfiguration.class);            new BestEffortsDeliveryJobFactory(config).init();        }    }}

jobConfig:  #作业名称  name: bestEffortsDeliveryJob  #触发作业的cron表达式--每5s重试一次  cron: 0/5 * * * * ?  #每次作业获取的事务日志最大数量  transactionLogFetchDataCount: 100  #事务送达的最大尝试次数.  maxDeliveryTryTimes: 3  #执行送达事务的延迟毫秒数,早于此间隔时间的入库事务才会被作业执行，其SQL为 where *** AND `creation_time`< (now() - maxDeliveryTryDelayMillis)，即至少60000ms，即一分钟前入库的事务日志才会被拉取出来；  maxDeliveryTryDelayMillis: 60000

@RequiredArgsConstructorpublic final class BestEffortsDeliveryJobFactory {    // 这个属性赋值通过有参构造方法进行赋值--new BestEffortsDeliveryJobFactory(config)，就是通过`config.yaml`配置的属性    private final BestEffortsDeliveryConfiguration bedConfig;    /**     * BestEffortsDeliveryJobMain中调用该init()方法，初始化最大努力尝试型异步JOB，该JOB基于elastic-job；     * Initialize best efforts delivery job.     */    public void init() {        // 根据config.yaml中配置的zkConfig节点，得到协调调度中心CoordinatorRegistryCenter        CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(createZookeeperConfiguration(bedConfig));        // 调度中心初始化        regCenter.init();        // 构造elastic-job调度任务        JobScheduler jobScheduler = new JobScheduler(regCenter, createBedJobConfiguration(bedConfig));        jobScheduler.setField("bedConfig", bedConfig);        jobScheduler.setField("transactionLogStorage", TransactionLogStorageFactory.createTransactionLogStorage(new RdbTransactionLogDataSource(bedConfig.getDefaultTransactionLogDataSource())));        jobScheduler.init();    }    // 根据该方法可知，创建的是BestEffortsDeliveryJob    private JobConfiguration createBedJobConfiguration(final BestEffortsDeliveryConfiguration bedJobConfig) {        // 根据config.yaml中配置的jobConfig节点得到job配置信息，且指定job类型为BestEffortsDeliveryJob        JobConfiguration result = new JobConfiguration(bedJobConfig.getJobConfig().getName(), BestEffortsDeliveryJob.class, 1, bedJobConfig.getJobConfig().getCron());        result.setFetchDataCount(bedJobConfig.getJobConfig().getTransactionLogFetchDataCount());        result.setOverwrite(true);        return result;    }

@Slf4jpublic class BestEffortsDeliveryJob extends AbstractIndividualThroughputDataFlowElasticJob<TransactionLog> {    @Setter    private BestEffortsDeliveryConfiguration bedConfig;    @Setter    private TransactionLogStorage transactionLogStorage;    @Override    public List<TransactionLog> fetchData(final JobExecutionMultipleShardingContext context) {        // 从transaction_log表中抓取最多100条事务日志（相关参数都在config.yaml中jobConfig节点下）        return transactionLogStorage.findEligibleTransactionLogs(context.getFetchDataCount(),             bedConfig.getJobConfig().getMaxDeliveryTryTimes(), bedConfig.getJobConfig().getMaxDeliveryTryDelayMillis());    }    @Override    public boolean processData(final JobExecutionMultipleShardingContext context, final TransactionLog data) {        try (            Connection conn = bedConfig.getTargetDataSource(data.getDataSource()).getConnection()) {            // 调用事务日志存储器的processData()进行处理            transactionLogStorage.processData(conn, data, bedConfig.getJobConfig().getMaxDeliveryTryTimes());        } catch (final SQLException | TransactionCompensationException ex) {            log.error(String.format("Async delivery times %s error, max try times is %s, exception is %s", data.getAsyncDeliveryTryTimes() + 1,                 bedConfig.getJobConfig().getMaxDeliveryTryTimes(), ex.getMessage()));            return false;        }        return true;    }}