清涧县 柯坪县 华宁县 泰和县 务川 山阴县 阳原县 金坛市 昔阳县 潼南县 都昌县 青海省 沙洋县 搜索 黔江区 铁岭市
2018年最新新闻网最新发布:马云回应阿里300多亿美元投资质疑:只是刚刚开始 深网|2018年互联网趋势报告:中国互联网与美国差距缩小 河南:“中原粮仓”开镰收麦 正式拉开三夏序幕 苹果WWDC18真正主角:iOS 12愿望清单可能性指数分析 Cocoa包管理器之CocoaPods详解 EA旗下游戏《FIFA 18》预测世界杯法国将夺冠  

喜家德水饺_【Canal源码分析】Sink及Store工作过程

标签:波罗奢花 信宜市

一、序列图

image

二、源码分析

2.1 Sink

Sink阶段所做的事情,就是根据一定的规则,对binlog数据进行一定的过滤。我们之前跟踪过parser过程的代码,发现在parser完成后,会把数据放到一个环形队列TransactionBuffer中,也就是这个方法:

transactionBuffer.add(entry);

我们具体看下add这个方法。

public void add(CanalEntry.Entry entry) throws InterruptedException {
    switch (entry.getEntryType()) {
        case TRANSACTIONBEGIN:
            flush();// 刷新上一次的数据
            put(entry);
            break;
        case TRANSACTIONEND:
            put(entry);
            flush();
            break;
        case ROWDATA:
            put(entry);
            // 针对非DML的数据,直接输出,不进行buffer控制
            EventType eventType = entry.getHeader().getEventType();
            if (eventType != null && !isDml(eventType)) {
                flush();
            }
            break;
        default:
            break;
    }
}

判断一下事件的类型,如果是事务开头,那么直接刷新之前的数据,然后把当前事件加到队列中;如果是事务的结束,那鬼泣_2018年最新新闻网么先把当前事务放到队列后,刷新到下一个阶段;如果是普通的事件,直接放到队列中,如果事务头类型不为空,且不是DML类型,那么直接刷新队列中数据到下一个阶段。

我们需要理清楚这块的逻辑,什么时候flush,什么时候put,针对不同的事件,采取的策略不一样。

这里我们分析下flush和put两个步骤。

2.1.1 flush队列

这块其实还没有涉及到sink阶段,还在维护一个事件环形队列。这个环形队列,维护了两个指针,一个是flush的指针,一个是put的指针,flush的指针永远是滞后于put指针的。

private void flush() throws InterruptedException {
    long start = this.flushSequence.get() + 1;
    long end = this.putSequence.get();

    if (start <= end) {
        List<CanalEntry.Entry> transaction = new ArrayList<CanalEntry.Entry>();
        for (long next = start; next <= end; next++) {
            transaction.add(this.entries[getIndex(next)]);
        }

        flushCallback.flush(transaction);
        flushSequence.set(end);// flush成功后,更新flush位置
    }
}

start就是flush的指针,end就是put的指针,flush的动作就是把当前flush到put中间的数据,全部刷新到下一个阶段。具体传递到下一个阶段的代码在flushCallback.flush方法中。这块我们下文再分析。

2.1.2 put

private void put(CanalEntry.Entry data) throws InterruptedException {
    // 首先检查是否有空位
    if (checkFreeSlotAt(putSequence.get() + 1)) {
        long current = putSequence.get();
        long next = current + 1;

        // 先写数据,再更新对应的cursor,并发度高的情况,putSequence会被get请求可见,拿出了ringbuffer中的老的Entry值
        entries[getIndex(next)] = data;
        putSequence.set(next);
    } else {
        flush();// buffer区满了,刷新一下
        put(data);// 继续加一下新数据
    }
}

这块的注释都比较清晰了,就不赘述了。

2.1.3 flush到sink

具体的代码在AbstractEventParser中,定义transactionBuffer的地方。

public void flush(List<CanalEntry.Entry> transaction) throws InterruptedException {
    boolean successed = consumeTheEventAndProfilingIfNecessary(transaction);
    if (!running) {
        return;
    }

    if (!successed) {
        throw new CanalParseException("consume failed!");
    }

    LogPosition position = buildLastTransactionPosition(transaction);
    if (position != null) { // 可能position为空
        logPositionManager.persistLogPosition(AbstractEventParser.this.destination, position);
    }
}

主要的处理在consumeTheEventAndProfilingIfNecessary里面。这里面调用了eventSink.sink()方法。

2.1.4 sink

这里面进行了binlog数据的过滤。首先判断是否需要过滤事务头和尾,如果需要过滤的话,直接过滤掉,默认不过滤。

遍历传到这个阶段的binlog列表,根据正则表达式判断,是否需要进行过滤,一般来说是根据表名、库名等进行过滤。这边的过滤类主要是AviaterRegexFilter,根据库名.表名和表达式进行过滤。如果需要进行过滤,那么直接把这个事件过滤。否则,加到binlog列表中,进行二次过滤。第二次过滤的主要内容是HEARTBEAT类型的事件,主要的代码在这里:

protected boolean doSink(List<Event> events) {
    for (CanalEventDownStreamHandler<List<Event>> handler : getHandlers()) {
        events = handler.before(events);//处理heartbeat事件
    }

    int fullTimes = 0;
    do {
        if (ev奸臣_2018年最新新闻网entStore.tryPut(events)) {
            for (CanalEventDownStreamHandler<List<Event>> handler : getHandlers()) {
                events = handler.after(events);
            }
            return true;
        } else {
            applyWait(++fullTimes);
        }

        for (Ca雅诗兰黛_2018年最新新闻网nalEventDownStreamHandler<List<Event>> handler : getHandlers()) {
            events = handler.retry(events);
        }

    } while (running && !Thread.interrupted());
    return false;
}

这里的CanalEventDownStreamHandler其实只有HeartBeatEntryEventHandler,也就是在before方法中把heartbeat事件从events去掉。这个心跳事件其实是parser过程生成的,我们之前有提到过。after目前是空的方法。

去掉之后,剩余的事件列表就会被调用tryPut()方法,送到下一步骤store中。

这里还有个applyWait方法,防止无限等待。

private void applyWait(int fullTimes) {
    int newFullTimes = fullTimes > maxFullTimes ? maxFullTimes : fullTimes;
    if (fullTimes <= 3) { // 3次以内
        Thread.yield();
    } else { // 超过3次,最多只sleep 10ms
        LockSupport.parkNanos(1000 * 1000L * newFullTimes);
    }

}

2.2 Store

目前只有基于内存模式的Store,这个阶段是真正Server中的落盘阶段。数据经历了mysql master到parser,再到sink,最后终于到了这里。

public boolean tryPut(List<Event> data) throws CanalStoreException {
    if (data == null || data.isEmpty()) {
        return true;
    }

    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        if (!checkFreeSlotAt(putSequence.get() + data.size())) {
            return false;
        } else {
            doPut(data);
            return true;
        }
    } finally {
        lock.unlock();
    }
}

在进行数据put的时候,加了一把锁。首先计算下是否还有剩余的空间进行数据处理,这里的计算,不光是计算了队列的剩余长度,还计算了剩余空间。队列的长度默认是16*1024,如果空间不足,直接拒绝,返回false,等待空间空日产_2018年最新新闻网余出来后,再进行put操作。否则,直接doPut()。

/**
 * 执行具体的put操作
 */
private void doPut(List<Event> data) {
    long current = putSequence.get();
    long end = current + data.size();

    // 先写数据,再更新对应的cursor,并发度高的情况,putSequence会被get请求可见,拿出了ringbuffer中的老的Entry值
    for (long next = current + 1; next <= end; next++) {
        entries[getIndex(next)] = data.get((int) (next - current - 1));
    }

    putSequence.set(end);

 北京交通大学_2018年最新新闻网   // 记录一下gets memsize信息,方便快速检索
    if (batchMode.isMemSize()) {
        long size = 0;
        for (Event event : data) {
            size += calculateSize(event);
        }

        putMemSize.getAndAdd(size);
    }

    // tell other threads that store is not empty
    notEmpty.signal();
}

这里主要对put一些指针,还outlook_2018年最新新闻网有空间做了重新的计算。放到队列中之后,通知其他等待notEmpty的线程,来进行数据的获取,这时候,client可以进行数据获取了。

当前文章:http://zcvo1w-tongguoqi-com.yangfish.cn/pqw/65170_151991.html

发布时间:2019-05-21 01:39:02

澳门银河娱乐yh163am.com网曝王菲真人秀酬劳近亿 经纪人回应:洽谈中  银河娱乐在线官方网址  银河娱乐在线官方网址  澳门银河官网真人娱乐  澳门银河www66356com  曾经令人感动的“爱心妈妈”李利娟被刑拘  澳门银河娱乐官网下载小猪佩奇不差钱是什么梗 佩奇为何不差钱  澳门银河官网yh163am.com  银河娱乐手机官网  澳门银河送彩金  

上一篇:3年来,微软市值首次大幅超过谷歌 下一篇:美国国家科学院院士张首晟:区块链结合人工智能,必产生重大影响

2018年最新新闻网相关阅读