一、说明
通过canal实时监听mysql binlog日志文件的变化，并将数据解析出来
二、环境准备
1、创建maven项目并修改pom.xml配置文件
com.alibaba.otter
canal.client
1.1.4
2、嗦代码
特别说明：在解析数据时，相当于程序是客户端，客户端在连接canal服务端时是不需要用户名和密码
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import java.net.InetSocketAddress;
import java.util.List;
public class CanalClient {
public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {
// 获取连接
CanalConnector canalConnector=CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.140.131",11111),
"example","","");
while(true)
{
// 连接
canalConnector.connect();
// 订阅数据库
canalConnector.subscribe("CanalDb.*");
// 获取数据
Message message = canalConnector.get(100);
// 获取Entry集合
List entries=message.getEntries();
// 判断集合是否为空，如果为空，则线程等待2秒再拉取数据
if (entries.size())
{
System.out.println("当次抓取没有数据，休息一会儿。。。");
Thread.sleep(2000);
}
else
{
// 遍历entries，单条解析
for (CanalEntry.Entry entry:entries)
{
// 1，获取表名
String tableName=entry.getHeader().getTableName();
// 2,获取类型
CanalEntry.EntryType entryType=entry.getEntryType();
// 3,获取序列化后的数据
ByteString storeValue=entry.getStoreValue();
// 4.判断当前entryType类型是否为ROWDATA
if (CanalEntry.EntryType.ROWDATA.equals(entryType))
{
//5.反序列化数据
CanalEntry.RowChange rowChange=CanalEntry.RowChange.parseFrom(storeValue);
//6.获取当前事件的操作类型
CanalEntry.EventType eventType=rowChange.getEventType();
//7.获取数据集
List rowDataList=rowChange.getRowDatasList();
//8.遍历rowDataList并打印数据集
for(CanalEntry.RowData rowData:rowDataList)
{
JSONObject beforData=new JSONObject();
List beforClountList=rowData.getBeforeColumnsList();
for (CanalEntry.Column column:beforClountList)
{
beforData.put(column.getName(),column.getValue());
}
JSONObject afterData=new JSONObject();
List afterClountList=rowData.getAfterColumnsList();
for (CanalEntry.Column column:afterClountList)
{
afterData.put(column.getName(),column.getValue());
}
// 打印数据
System.out.println(""+tableName+
",EventType:"+eventType+
",Before:"+beforData+
",After:"+afterData);
}
}
else
{
System.out.println("当前操作类型为"+entryType);
}
}
}
}
}
}
三、项目效果
Original: https://www.cnblogs.com/sportsky/p/16530806.html
Author: SportSky
Title: Canal实时解析mysql binlog数据实战
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/609208/
转载文章受原作者版权保护。转载请注明原作者出处！