The canal-client's discuss from fanqinghui

canal-client's Issues

关于异步get的问题

贴一下你写的代码哈。讨教一个问题

@Override
            public void run() {
                    while (!threadPool.isShutdown() && !threadPool.isTerminated()) {
                        Message message = connector.getWithoutAck(canalConfig.getFetchSize(),10L,TimeUnit.SECONDS); // 获取指定数量的数据
                        long batchId = message.getId();
                        int size = message.getEntries().size();
                        if (batchId == -1 || size == 0) {
                            try {
                                Thread.sleep(1000);
                            } catch (InterruptedException e) {
                                logger.warn("CustomSimpleCanalClient invoke thread interrupted!");
                            }
                        } else {
                            List<Entry> entryList = message.getEntries();
                            for (Entry entry : entryList) {
                                try {
                                    customTableQueue.put(entry);
                                } catch (InterruptedException e) {
                                    //do nothing
                                    logger.warn("customTableQueue interrupt.", e);
                                }
                            }
                        }
                        connector.ack(batchId); // 提交确认
                    }
            }

如果你放到阻塞队列里面就ack了，那么 ack完后，突然宕机，是不是customTableQueue的entry都丢了，然后canal server的数据也被删除了？这是有问题的吧？

Recommend Projects