[Java知识库]处理Binlog相关类

一.Binlog策略模式接口

public interface BinlogKeeper {

    /**
     * 获取监听表名
     * @return
     */
    default String tableName(){
        return "default_table";
    }

    /**
     * 处理mq消息
     * @param binlogMessage
     * @return
     */
    default boolean dealMqMessage(BinlogMessage binlogMessage){
        return true;
    }
}

二.Binlog策略模式接口实现，其他的表Binlog都类似

@Slf4j
@Service(value = "defaultBinlogService")
public class DefaultBinlogServiceImpl implements BinlogKeeper {

    /**
     * 监听表名称
     */
    public static final String TABLE_NAME = "default_table";

    @Override
    public String tableName() {
        return TABLE_NAME;
    }

    @Override
    public boolean dealMqMessage(BinlogMessage binlogMessage) {
        String table = binlogMessage.getTable();
        log.warn("待处理的binlog未配置，方法名：defaultBinlogService，表名： " + table);
        return true;
    }
}

三.工厂+策略模式

@Component
public class BinlogServiceFactory {

    @Autowired
    ApplicationContext applicationContext;

    /**
     * binlog监听服务实现map
     */
    private static final Map<String, BinlogKeeper> serviceMap = new ConcurrentHashMap<>(8);

    @PostConstruct
    private void initServiceMap() {
        Map<String, BinlogKeeper> beans = applicationContext.getBeansOfType(BinlogKeeper.class);
        beans.forEach((beanName, service) -> serviceMap.put(service.tableName(), service));
    }
    public BinlogKeeper getInstance(String table) {
        Assert.notNull(table, "表名不能为null");
        if (serviceMap.containsKey(table)) {
            return serviceMap.get(table);
        } else {
            return serviceMap.get(DefaultBinlogServiceImpl.TABLE_NAME);
        }
    }

}

四.BinlogMessage 实体

@Data
public class BinlogMessage implements Serializable {

    private String database;

    private String table;

    private String type;

    private String ts;

    private String xid;

    private boolean commit;

    private String position;

    private String server_id;

    private Integer thread_id;

    private Integer schema_id;

    /**
     * 真正的行数据（json格式）
     */
    private String data;
    /**
     * 更新操作，更新前数据（json格式）
     */
    private JSONObject old;
}

五.监听binlog对应的MQ消息

  /**
     * 默认限流数量
     */
    @Value("${xxx:50}")
    private Integer limiter = 50;

    /**
     * 消费数量
     */
    @Value("${xxx:50}")
    private Integer consumerConcurrency;

    /**
     * 限流器
     */
    private RateLimiter rateLimiter;

   @PostConstruct
    public void init() {
        if (consumerConcurrency == null || consumerConcurrency <= 0) {
            rateLimiter = RateLimiter.create(limiter);
        }else {
            rateLimiter = RateLimiter.create(consumerConcurrency);
        }
    }

     public void receive(String payload) {
        // 限流，有可能会需要等待
        rateLimiter.acquire();
        log.info("接收到的信息:{}",payload);

        // 将消息转换成 binlog实体类型
        BinlogMessage binlogMessage = JSON.parseObject(payload, BinlogMessage.class);
        log.info("解析后的消息:{}",JSONObject.toJSONString(binlogMessage));
        String table = binlogMessage.getTable();
        if (StringUtils.isBlank(table) || StringUtils.isBlank(binlogMessage.getData())) {
            log.error("监听到的消息,表名或表数据为空,消息体:{}",JSONObject.toJSONString(binlogMessage));
        }
        // 找到对应表的binlog执行类
        BinlogKeeper binlogService = binlogServiceFactory.getInstance(table);
        // 执行解析MQ推送的binlog信息
        binlogService.dealMqMessage(binlogMessage);

    }

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖