本文整理汇总了Java中com.alibaba.otter.canal.protocol.CanalEntry类的典型用法代码示例。如果您正苦于以下问题：Java CanalEntry类的具体用法？Java CanalEntry怎么用？Java CanalEntry使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
CanalEntry类属于com.alibaba.otter.canal.protocol包，在下文中一共展示了CanalEntry类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。
示例1: insert
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
public void insert(List<CanalEntry.Column> data, String schemaName, String tableName) {
DBObject obj = DBConvertUtil.columnToJson(data);
logger.debug("insert ：{}", obj.toString());
//订单库单独处理
if (schemaName.equals("order")) {
//保存原始数据
if (tableName.startsWith("order_base_info")) {
tableName = "order_base_info";
} else if (tableName.startsWith("order_detail_info")) {
tableName = "order_detail_info";
} else {
logger.info("unknown data ：{}.{}:{}", schemaName, tableName, obj);
return;
}
insertData(schemaName, tableName, obj, obj);
} else {
DBObject newObj = (DBObject) ObjectUtils.clone(obj);
if (newObj.containsField("id")) {
newObj.put("_id", newObj.get("id"));
newObj.removeField("id");
}
insertData(schemaName, tableName, newObj, obj);
}
}
开发者ID:zhangtr，项目名称:canal-mongo，代码行数:25，代码来源:DataService.java
示例2: update
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
public void update(List<CanalEntry.Column> data, String schemaName, String tableName) {
DBObject obj = DBConvertUtil.columnToJson(data);
logger.debug("update：{}", obj.toString());
//订单库单独处理
if (schemaName.equals("order")) {
if (tableName.startsWith("order_base_info")) {
tableName = "order_base_info";
} else if (tableName.startsWith("order_detail_info")) {
tableName = "order_detail_info";
} else {
logger.info("unknown data：{}.{}:{}", schemaName, tableName, obj);
}
updateData(schemaName, tableName, new BasicDBObject("orderId", obj.get("orderId")), obj);
} else {
if (obj.containsField("id")) {
updateData(schemaName, tableName, new BasicDBObject("_id", obj.get("id")), obj);
} else {
logger.info("unknown data structure");
}
}
}
开发者ID:zhangtr，项目名称:canal-mongo，代码行数:22，代码来源:DataService.java
示例3: insertData
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
public void insertData(String schemaName, String tableName, DBObject naive, DBObject complete) {
int i = 0;
DBObject logObj = (DBObject) ObjectUtils.clone(complete);
//保存原始数据
try {
String path = "/" + schemaName + "/" + tableName + "/" + CanalEntry.EventType.INSERT.getNumber();
i++;
naiveMongoTemplate.getCollection(tableName).insert(naive);
i++;
SpringUtil.doEvent(path, complete);
i++;
} catch (MongoClientException | MongoSocketException clientException) {
//客户端连接异常抛出，阻塞同步，防止mongodb宕机
throw clientException;
} catch (Exception e) {
logError(schemaName, tableName, 1, i, logObj, e);
}
}
开发者ID:zhangtr，项目名称:canal-mongo，代码行数:19，代码来源:DataService.java
示例4: updateData
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
public void updateData(String schemaName, String tableName, DBObject query, DBObject obj) {
String path = "/" + schemaName + "/" + tableName + "/" + CanalEntry.EventType.UPDATE.getNumber();
int i = 0;
DBObject newObj = (DBObject) ObjectUtils.clone(obj);
DBObject logObj = (DBObject) ObjectUtils.clone(obj);
//保存原始数据
try {
obj.removeField("id");
i++;
naiveMongoTemplate.getCollection(tableName).update(query, obj);
i++;
SpringUtil.doEvent(path, newObj);
i++;
} catch (MongoClientException | MongoSocketException clientException) {
//客户端连接异常抛出，阻塞同步，防止mongodb宕机
throw clientException;
} catch (Exception e) {
logError(schemaName, tableName, 2, i, logObj, e);
}
}
开发者ID:zhangtr，项目名称:canal-mongo，代码行数:21，代码来源:DataService.java
示例5: deleteData
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
public void deleteData(String schemaName, String tableName, DBObject obj) {
int i = 0;
String path = "/" + schemaName + "/" + tableName + "/" + CanalEntry.EventType.DELETE.getNumber();
DBObject newObj = (DBObject) ObjectUtils.clone(obj);
DBObject logObj = (DBObject) ObjectUtils.clone(obj);
//保存原始数据
try {
i++;
if (obj.containsField("id")) {
naiveMongoTemplate.getCollection(tableName).remove(new BasicDBObject("_id", obj.get("id")));
}
i++;
SpringUtil.doEvent(path, newObj);
} catch (MongoClientException | MongoSocketException clientException) {
//客户端连接异常抛出，阻塞同步，防止mongodb宕机
throw clientException;
} catch (Exception e) {
logError(schemaName, tableName, 3, i, logObj, e);
}
}
开发者ID:zhangtr，项目名称:canal-mongo，代码行数:21，代码来源:DataService.java
示例6: filter
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
@Override
public boolean filter(CanalEntry.Entry entry) {
//过滤掉事物头尾等 非 row data 的 entry
boolean rowData = entry.getEntryType() == CanalEntry.EntryType.ROWDATA;
if (!rowData) {
return false;
}
//只保存 insert update delete 类型的 时间
boolean eventType = filterEventType(entry.getHeader().getEventType());
if (!eventType) {
return false;
}
return true;
}
开发者ID:zhongchengxcr，项目名称:canal-elasticsearch，代码行数:20，代码来源:SimpleMessageFilter.java
示例7: printSummary
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
private void printSummary(Message message, long batchId, int size) {
long memsize = 0;
for (CanalEntry.Entry entry : message.getEntries()) {
memsize += entry.getHeader().getEventLength();
}
String startPosition = null;
String endPosition = null;
if (!CollectionUtils.isEmpty(message.getEntries())) {
startPosition = buildPositionForDump(message.getEntries().get(0));
endPosition = buildPositionForDump(message.getEntries().get(message.getEntries().size() - 1));
}
logger.info(context_format, new Object[]{batchId, size, memsize, format.format(new Date()), startPosition,
endPosition});
}
开发者ID:zhongchengxcr，项目名称:canal-elasticsearch，代码行数:17，代码来源:SelectorTask.java
示例8: changedDataParse
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
/**
* {@link Schema.Table#columns}有值, 则对于UPDATE操作过滤更改的字段是否包含在{@link Schema.Table#columns}
* DELETE, INSERT事件执行条件过滤, 对于UPDATE的过滤不在这做, 比较复杂, 由子类自己实现过滤
*
* @param rowChange 更改的数据
* @return 解析结果
*/
@SuppressWarnings({"rawtypes", "unchecked"})
@Override
protected final List<RowChangedData> changedDataParse(CanalEntry.RowChange rowChange) {
List<RowChangedData> dataList = RowChangedData.build(rowChange, currentTable.getRowDataColumns());
if (CommonsUtils.isEmpty(dataList)) return null;
ConditionContainer columnCondition;
if (currentEventType != CanalEntry.EventType.UPDATE
&& (columnCondition = currentTable.getColumnCondition()) != null) {
//对于INSERT类型的记录更新, 如果条件判断没有通过, 可以认为该更新事件没有发生~~~~
//对于DELETE类型的记录更新, 如果条件判断没有通过, 可以认为该数据删除之前就不关心, 那这次删除我们更不关心了~~~
Iterator<RowChangedData> it = dataList.iterator();
while (it.hasNext()) {
if (!columnCondition.verify(it.next())) {
it.remove();
}
}
}
return dataList;
}
开发者ID:wxingyl，项目名称:search-commons，代码行数:27，代码来源:ActionableInstanceHandle.java
示例9: startHandle
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
@SuppressWarnings({"rawtypes", "unchecked"})
@Override
public boolean startHandle(CanalEntry.Header header) {
currentEventType = header.getEventType();
Map.Entry<String, String> virtualName = schemaTableNameAdapter.getVirtualName(header.getSchemaName(), header.getTableName());
currentTable = actionFactory.getTable(virtualName.getKey(), virtualName.getValue());
if (currentTable == null) return false;
//排除事件类型过滤, 对于UPDATE类型, 如果存在条件判断, 在这儿没有办法执行排除
if ((RowChangedData.getEventTypeFlag(currentEventType) & currentTable.getForbidEventType()) != 0
&& (currentEventType != CanalEntry.EventType.UPDATE || currentTable.getColumnCondition() == null)) {
return false;
}
T action = currentTable.getAction();
if (action instanceof CurrentHandleTable) {
((CurrentHandleTable<T>) action).setCurrentTable(currentTable);
}
return true;
}
开发者ID:wxingyl，项目名称:search-commons，代码行数:19，代码来源:ActionableInstanceHandle.java
示例10: runLastEventTypeOfAction
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
@SuppressWarnings({"rawtypes", "unchecked"})
private void runLastEventTypeOfAction(int eventType, List<? extends RowChangedData> dataList) {
if (log.isDebugEnabled()) {
log.debug("canal instance: " + instanceName + " need handle data size: " + dataList.size() + ", eventType: " + eventType
+ " table: " + lastTable);
}
if (eventType == CanalEntry.EventType.UPDATE_VALUE) {
lastTable.getAction().onUpdateAction(Collections.unmodifiableList((List<RowChangedData.Update>) dataList));
} else if (eventType == CanalEntry.EventType.INSERT_VALUE) {
lastTable.getAction().onInsertAction(Collections.unmodifiableList((List<RowChangedData.Insert>) dataList));
} else {
lastTable.getAction().onDeleteAction(Collections.unmodifiableList((List<RowChangedData.Delete>) dataList));
}
//这儿主动调用clear, 数据无效掉, 避免调用Action时保留引用导致无法回收
for (RowChangedData data : dataList) {
data.close();
}
//这儿清楚掉
dataList.clear();
}
开发者ID:wxingyl，项目名称:search-commons，代码行数:21，代码来源:EventTypeSectionHandle.java
示例11: onAction
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
@Override
public void onAction(final List<? extends RowChangedData> changedData) {
Iterator<? extends RowChangedData> it = changedData.iterator();
CanalEntry.EventType lastType = null;
int startIndex = 0, endIndex = 0;
while (it.hasNext()) {
RowChangedData curData = it.next();
CanalEntry.EventType curType = RowChangedData.getEventType(curData);
endIndex++;
if (lastType == null) {
lastType = curType;
} else if (lastType != curType) {
doAction(changedData, lastType, startIndex, endIndex);
startIndex = endIndex;
lastType = null;
}
}
if (lastType != null) {
doAction(changedData, lastType, startIndex, endIndex);
}
}
开发者ID:wxingyl，项目名称:search-commons，代码行数:22，代码来源:Actions.java
示例12: consumerMessage
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
/**
* 消费当前消息
*/
private void consumerMessage(Message message) {
log.debug("canal instance: " + handle.instanceName() + " get message entry size " + message.getEntries().size());
try {
for (CanalEntry.Entry e : message.getEntries()) {
if (e.getEntryType() != CanalEntry.EntryType.ROWDATA || !e.hasStoreValue()) continue;
CanalEntry.Header header = e.getHeader();
if (header.getExecuteTime() < startRtTime
|| header.getEventType().getNumber() > CanalEntry.EventType.DELETE_VALUE
|| !handle.startHandle(header)) continue;
try {
CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(e.getStoreValue());
if (rowChange.getIsDdl()) continue;
handle.rowChangeHandle(rowChange);
} catch (InvalidProtocolBufferException e1) {
log.error("canal instance: " + handle.instanceName() + " parse store value have exception: ", e1);
}
}
handle.ack(message.getId());
} finally {
handle.finishMessageHandle();
}
}
开发者ID:wxingyl，项目名称:search-commons，代码行数:26，代码来源:CanalExecutor.java
示例13: put
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
@Override
public void put(List<Event> list) throws InterruptedException, CanalStoreException {
if (CollectionUtils.isNotEmpty(list)) {
for (Event event : list) {
CanalEntry.Entry entry = event.getEntry();
try {
CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
CanalEntry.EventType eventType = rowChange.getEventType();
if (eventType == CanalEntry.EventType.INSERT || eventType == CanalEntry.EventType.UPDATE || eventType == CanalEntry.EventType.DELETE) {
List<CanalEntry.RowData> rowDatas = rowChange.getRowDatasList();
for (CanalEntry.RowData rowData : rowDatas) {
sendKafkaMsg(rowData, entry, eventType);
}
}
} catch (InvalidProtocolBufferException e) {
LOGGER.error(e.getMessage(), e);
}
}
}
}
开发者ID:kevinKaiF，项目名称:cango，代码行数:22，代码来源:KafkaMysqlStore.java
示例14: encapsulationColumn
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
private void encapsulationColumn(CangoMsgDto msgDto, List<CanalEntry.Column> columns, List<ColumnDto> columnDtos, boolean isNewData) {
for (CanalEntry.Column column : columns) {
if (column.getIsKey()) {
msgDto.setPkName(column.getName()).setPkValue(convertValue(column.getValue(), column.getSqlType()));
}
ColumnDto columnDto = getColumnDto(columnDtos, column.getName());
columnDto.setJdbcType(column.getSqlType());
if (isNewData) {
columnDto.setStrValue(column.getValue());
columnDto.setValue(convertValue(column.getValue(), column.getSqlType()));
} else {
columnDto.setOldStrValue(column.getValue());
columnDto.setOldValue(convertValue(column.getValue(), column.getSqlType()));
}
}
}
开发者ID:kevinKaiF，项目名称:cango，代码行数:19，代码来源:KafkaMysqlStore.java
示例15: add
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
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;
}
}
开发者ID:alibaba，项目名称:canal，代码行数:23，代码来源:EventTransactionBuffer.java
示例16: AbstractEventParser
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
public AbstractEventParser(){
// 初始化一下
transactionBuffer = new EventTransactionBuffer(new TransactionFlushCallback() {
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);
}
}
});
}
开发者ID:alibaba，项目名称:canal，代码行数:22，代码来源:AbstractEventParser.java
示例17: consumeTheEventAndProfilingIfNecessary
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
protected boolean consumeTheEventAndProfilingIfNecessary(List<CanalEntry.Entry> entrys) throws CanalSinkException,
InterruptedException {
long startTs = -1;
boolean enabled = getProfilingEnabled();
if (enabled) {
startTs = System.currentTimeMillis();
}
boolean result = eventSink.sink(entrys, (runningInfo == null) ? null : runningInfo.getAddress(), destination);
if (enabled) {
this.processingInterval = System.currentTimeMillis() - startTs;
}
if (consumedEventCount.incrementAndGet() < 0) {
consumedEventCount.set(0);
}
return result;
}
开发者ID:alibaba，项目名称:canal，代码行数:21，代码来源:AbstractEventParser.java
示例18: parseAndProfilingIfNecessary
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
protected CanalEntry.Entry parseAndProfilingIfNecessary(EVENT bod, boolean isSeek) throws Exception {
long startTs = -1;
boolean enabled = getProfilingEnabled();
if (enabled) {
startTs = System.currentTimeMillis();
}
CanalEntry.Entry event = binlogParser.parse(bod, isSeek);
if (enabled) {
this.parsingInterval = System.currentTimeMillis() - startTs;
}
if (parsedEventCount.incrementAndGet() < 0) {
parsedEventCount.set(0);
}
return event;
}
开发者ID:alibaba，项目名称:canal，代码行数:17，代码来源:AbstractEventParser.java
示例19: put
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
public void put(List<Event> datas) throws InterruptedException, CanalStoreException {
for (Event data : datas) {
CanalEntry.Header header = data.getEntry().getHeader();
Date date = new Date(header.getExecuteTime());
SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
if (data.getEntry().getEntryType() == EntryType.TRANSACTIONBEGIN
|| data.getEntry().getEntryType() == EntryType.TRANSACTIONEND) {
// System.out.println(MessageFormat.format(messgae, new Object[]
// { Thread.currentThread().getName(),
// header.getLogfilename(), header.getLogfileoffset(),
// format.format(date),
// data.getEntry().getEntryType(), "" }));
System.out.println(data.getEntry().getEntryType());
} else {
System.out.println(MessageFormat.format(messgae,
new Object[] { Thread.currentThread().getName(), header.getLogfileName(),
String.valueOf(header.getLogfileOffset()), format.format(date), header.getEventType(),
header.getSchemaName(), header.getTableName() }));
}
}
}
开发者ID:alibaba，项目名称:canal，代码行数:23，代码来源:DummyEventStore.java
示例20: delete
import com.alibaba.otter.canal.protocol.CanalEntry; //导入依赖的package包/类
public void delete(List<CanalEntry.Column> data, String schemaName, String tableName) {
DBObject obj = DBConvertUtil.columnToJson(data);
logger.debug("delete：{}", obj.toString());
if (schemaName.equals("order")) {
logger.info("not support delete：{}.{}:{}", schemaName, tableName, obj);
} else {
deleteData(schemaName, tableName, obj);
}
}
开发者ID:zhangtr，项目名称:canal-mongo，代码行数:10，代码来源:DataService.java
注：本文中的com.alibaba.otter.canal.protocol.CanalEntry类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论