chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed PR format

Browse Source
This commit is contained in:
jianghang.loujh
2023-07-28 13:20:56 +08:00
parent b8c2a757a0
commit 97aa441bca
11 changed files with 62 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
example/src/main/java/com/alibaba/otter/canal/example/BaseCanalClientTest.java
View File

@@ -13,26 +13,19 @@ import org.springframework.util.CollectionUtils;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.Pair;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.CanalEntry.TransactionBegin;
import com.alibaba.otter.canal.protocol.CanalEntry.TransactionEnd;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.google.protobuf.InvalidProtocolBufferException;
public class BaseCanalClientTest {
protected final static Logger logger = LoggerFactory.getLogger(AbstractCanalClientTest.class);
protected final static Logger logger = LoggerFactory
.getLogger(AbstractCanalClientTest.class);
protected static final String SEP = SystemUtils.LINE_SEPARATOR;
protected static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
protected volatile boolean running = false;
protected Thread.UncaughtExceptionHandler handler = (t, e) -> logger.error("parse events has an error",
e);
e);
protected Thread thread = null;
protected CanalConnector connector;
protected static String context_format = null;
@@ -51,8 +44,7 @@ public class BaseCanalClientTest {
+ "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {}({}) , gtid : ({}) , delay : {} ms"
+ SEP;
transaction_format = SEP
+ "================> binlog[{}:{}] , executeTime : {}({}) , gtid : ({}) , delay : {}ms"
transaction_format = SEP + "================> binlog[{}:{}] , executeTime : {}({}) , gtid : ({}) , delay : {}ms"
+ SEP;
}
@@ -71,8 +63,8 @@ public class BaseCanalClientTest {
}
SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
logger.info(context_format, new Object[] { batchId, size, memsize, format.format(new Date()), startPosition,
endPosition });
logger.info(context_format,
new Object[] { batchId, size, memsize, format.format(new Date()), startPosition, endPosition });
}
protected String buildPositionForDump(Entry entry) {
@@ -94,7 +86,8 @@ public class BaseCanalClientTest {
Date date = new Date(entry.getHeader().getExecuteTime());
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
|| entry.getEntryType() == EntryType.TRANSACTIONEND) {
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN) {
TransactionBegin begin = null;
try {
@@ -105,9 +98,10 @@ public class BaseCanalClientTest {
// 打印事务头信息，执行的线程id，事务耗时
logger.info(transaction_format,
new Object[] { entry.getHeader().getLogfileName(),
String.valueOf(entry.getHeader().getLogfileOffset()),
String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
entry.getHeader().getGtid(), String.valueOf(delayTime) });
String.valueOf(entry.getHeader().getLogfileOffset()),
String.valueOf(entry.getHeader().getExecuteTime()),
simpleDateFormat.format(date), entry.getHeader().getGtid(),
String.valueOf(delayTime) });
logger.info(" BEGIN ----> Thread id: {}", begin.getThreadId());
printXAInfo(begin.getPropsList());
} else if (entry.getEntryType() == EntryType.TRANSACTIONEND) {
@@ -123,9 +117,10 @@ public class BaseCanalClientTest {
printXAInfo(end.getPropsList());
logger.info(transaction_format,
new Object[] { entry.getHeader().getLogfileName(),
String.valueOf(entry.getHeader().getLogfileOffset()),
String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
entry.getHeader().getGtid(), String.valueOf(delayTime) });
String.valueOf(entry.getHeader().getLogfileOffset()),
String.valueOf(entry.getHeader().getExecuteTime()),
simpleDateFormat.format(date), entry.getHeader().getGtid(),
String.valueOf(delayTime) });
}
continue;
@@ -143,10 +138,10 @@ public class BaseCanalClientTest {
logger.info(row_format,
new Object[] { entry.getHeader().getLogfileName(),
String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
entry.getHeader().getTableName(), eventType,
String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
entry.getHeader().getGtid(), String.valueOf(delayTime) });
String.valueOf(entry.getHeader().getLogfileOffset()),
entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType,
String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
entry.getHeader().getGtid(), String.valueOf(delayTime) });
if (eventType == EventType.QUERY || rowChange.getIsDdl()) {
logger.info("ddl : " + rowChange.getIsDdl() + " , sql ----> " + rowChange.getSql() + SEP);
@@ -174,8 +169,8 @@ public class BaseCanalClientTest {
if (StringUtils.containsIgnoreCase(column.getMysqlType(), "BLOB")
|| StringUtils.containsIgnoreCase(column.getMysqlType(), "BINARY")) {
// get value bytes
builder.append(column.getName() + " : "
+ new String(column.getValue().getBytes("ISO-8859-1"), "UTF-8"));
builder.append(
column.getName() + " : " + new String(column.getValue().getBytes("ISO-8859-1"), "UTF-8"));
} else {
builder.append(column.getName() + " : " + column.getValue());
}

6
example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientPermanceTest.java
View File

@@ -22,10 +22,8 @@ public class SimpleCanalClientPermanceTest {
long end = 0;
final ArrayBlockingQueue<Long> queue = new ArrayBlockingQueue<>(100);
try {
final CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111),
destination,
"canal",
"canal");
final CanalConnector connector = CanalConnectors
.newSingleConnector(new InetSocketAddress(ip, 11111), destination, "canal", "canal");
Thread ackThread = new Thread(() -> {
while (true) {

6
example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java
View File

@@ -22,10 +22,8 @@ public class SimpleCanalClientTest extends AbstractCanalClientTest {
// 根据ip，直接创建链接，无HA的功能
String destination = "example";
String ip = AddressUtils.getHostIp();
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111),
destination,
"canal",
"canal");
CanalConnector connector = CanalConnectors
.newSingleConnector(new InetSocketAddress(ip, 11111), destination, "canal", "canal");
final SimpleCanalClientTest clientTest = new SimpleCanalClientTest(destination);
clientTest.setConnector(connector);

3
example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaClientExample.java
View File

@@ -34,7 +34,8 @@ public class CanalKafkaClientExample {
public static void main(String[] args) {
try {
final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample(AbstractKafkaTest.zkServers,
final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample(
AbstractKafkaTest.zkServers,
AbstractKafkaTest.servers,
AbstractKafkaTest.topic,
AbstractKafkaTest.partition,

3
example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaClientFlatMessageExample.java
View File

@@ -35,7 +35,8 @@ public class CanalKafkaClientFlatMessageExample {
public static void main(String[] args) {
try {
final CanalKafkaClientFlatMessageExample kafkaCanalClientExample = new CanalKafkaClientFlatMessageExample(AbstractKafkaTest.zkServers,
final CanalKafkaClientFlatMessageExample kafkaCanalClientExample = new CanalKafkaClientFlatMessageExample(
AbstractKafkaTest.zkServers,
AbstractKafkaTest.servers,
AbstractKafkaTest.topic,
AbstractKafkaTest.partition,

3
example/src/main/java/com/alibaba/otter/canal/example/kafka/CanalKafkaOffsetClientExample.java
View File

@@ -43,7 +43,8 @@ public class CanalKafkaOffsetClientExample {
public static void main(String[] args) {
try {
final CanalKafkaOffsetClientExample kafkaCanalClientExample = new CanalKafkaOffsetClientExample(AbstractKafkaTest.servers,
final CanalKafkaOffsetClientExample kafkaCanalClientExample = new CanalKafkaOffsetClientExample(
AbstractKafkaTest.servers,
AbstractKafkaTest.topic,
AbstractKafkaTest.partition,
AbstractKafkaTest.groupId);

24
example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientExample.java
View File

@@ -1,13 +1,15 @@
package com.alibaba.otter.canal.example.rocketmq;
import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector;
import com.alibaba.otter.canal.protocol.Message;
/**
* RocketMQ client example
*
@@ -26,15 +28,23 @@ public class CanalRocketMQClientExample extends AbstractRocektMQTest {
private Thread.UncaughtExceptionHandler handler = (t, e) -> logger.error("parse events has an error", e);
public CanalRocketMQClientExample(String nameServers, String topic, String groupId) {
public CanalRocketMQClientExample(String nameServers, String topic, String groupId){
connector = new RocketMQCanalConnector(nameServers, topic, groupId, 500, false);
}
public CanalRocketMQClientExample(String nameServers, String topic, String groupId, boolean enableMessageTrace,
String accessKey, String secretKey, String accessChannel, String namespace) {
connector = new RocketMQCanalConnector(nameServers, topic, groupId, accessKey,
secretKey, -1, false, enableMessageTrace,
null, accessChannel, namespace);
String accessKey, String secretKey, String accessChannel, String namespace){
connector = new RocketMQCanalConnector(nameServers,
topic,
groupId,
accessKey,
secretKey,
-1,
false,
enableMessageTrace,
null,
accessChannel,
namespace);
}
public static void main(String[] args) {

6
example/src/main/java/com/alibaba/otter/canal/example/rocketmq/CanalRocketMQClientFlatMessageExample.java
View File

@@ -18,7 +18,8 @@ import com.alibaba.otter.canal.protocol.FlatMessage;
*/
public class CanalRocketMQClientFlatMessageExample extends AbstractRocektMQTest {
protected final static Logger logger = LoggerFactory.getLogger(CanalRocketMQClientFlatMessageExample.class);
protected final static Logger logger = LoggerFactory
.getLogger(CanalRocketMQClientFlatMessageExample.class);
private RocketMQCanalConnector connector;
@@ -34,7 +35,8 @@ public class CanalRocketMQClientFlatMessageExample extends AbstractRocektMQTest
public static void main(String[] args) {
try {
final CanalRocketMQClientFlatMessageExample rocketMQClientExample = new CanalRocketMQClientFlatMessageExample(nameServers,
final CanalRocketMQClientFlatMessageExample rocketMQClientExample = new CanalRocketMQClientFlatMessageExample(
nameServers,
topic,
groupId);
logger.info("## Start the rocketmq consumer: {}-{}", topic, groupId);