canal安装及使用-ag真人游戏

（3）更新依赖 mvn install

（4）canal-client.java  实例代码

[java] view plain copy  

1. /** 
2.  * created by hp on 14-7-17. 
3.  */  
4. import java.net.inetsocketaddress;  
5. import java.util.list;  
6.   
7. import com.alibaba.otter.canal.client.canalconnector;  
8. import com.alibaba.otter.canal.common.utils.addressutils;  
9. import com.alibaba.otter.canal.protocol.message;  
10. import com.alibaba.otter.canal.protocol.canalentry.column;  
11. import com.alibaba.otter.canal.protocol.canalentry.entry;  
12. import com.alibaba.otter.canal.protocol.canalentry.entrytype;  
13. import com.alibaba.otter.canal.protocol.canalentry.eventtype;  
14. import com.alibaba.otter.canal.protocol.canalentry.rowchange;  
15. import com.alibaba.otter.canal.protocol.canalentry.rowdata;  
16. import com.alibaba.otter.canal.client.*;  
17. import org.jetbrains.annotations.notnull;  
18.   
19. public class clientsample {  
20.   
21.     public static void main(string args[]) {  
22.         // 创建链接  
23.         canalconnector connector = canalconnectors.newsingleconnector(new inetsocketaddress(addressutils.gethostip(),  
24.                 11111), "example", "", "");  
25.         int batchsize = 1000;  
26.         int emptycount = 0;  
27.         try {  
28.             connector.connect();  
29.             connector.subscribe(".*\\..*");  
30.             connector.rollback();  
31.             int totalemtrycount = 1200;  
32.             while (emptycount < totalemtrycount) {  
33.                 message message = connector.getwithoutack(batchsize); // 获取指定数量的数据  
34.                 long batchid = message.getid();  
35.                 int size = message.getentries().size();  
36.                 if (batchid == -1 || size == 0) {  
37.                     emptycount ;  
38.                     system.out.println("empty count : "   emptycount);  
39.                     try {  
40.                         thread.sleep(1000);  
41.                     } catch (interruptedexception e) {  
42.                         e.printstacktrace();  
43.                     }  
44.                 } else {  
45.                     emptycount = 0;  
46.                     // system.out.printf("message[batchid=%s,size=%s] \n", batchid, size);  
47.                     printentry(message.getentries());  
48.                 }  
49.   
50.                 connector.ack(batchid); // 提交确认  
51.                 // connector.rollback(batchid); // 处理失败, 回滚数据  
52.             }  
53.   
54.             system.out.println("empty too many times, exit");  
55.         } finally {  
56.             connector.disconnect();  
57.         }  
58.     }  
59.   
60.     private static void printentry(@notnull list entrys) {  
61.         for (entry entry : entrys) {  
62.             if (entry.getentrytype() == entrytype.transactionbegin || entry.getentrytype() == entrytype.transactionend) {  
63.                 continue;  
64.             }  
65.   
66.             rowchange rowchage = null;  
67.             try {  
68.                 rowchage = rowchange.parsefrom(entry.getstorevalue());  
69.             } catch (exception e) {  
70.                 throw new runtimeexception("error ## parser of eromanga-event has an error , data:"   entry.tostring(),  
71.                         e);  
72.             }  
73.   
74.             eventtype eventtype = rowchage.geteventtype();  
75.             system.out.println(string.format("================> binlog[%s:%s] , name[%s,%s] , eventtype : %s",  
76.                     entry.getheader().getlogfilename(), entry.getheader().getlogfileoffset(),  
77.                     entry.getheader().getschemaname(), entry.getheader().gettablename(),  
78.                     eventtype));  
79.   
80.             for (rowdata rowdata : rowchage.getrowdataslist()) {  
81.                 if (eventtype == eventtype.delete) {  
82.                     printcolumn(rowdata.getbeforecolumnslist());  
83.                 } else if (eventtype == eventtype.insert) {  
84.                     printcolumn(rowdata.getaftercolumnslist());  
85.                 } else {  
86.                     system.out.println("-------> before");  
87.                     printcolumn(rowdata.getbeforecolumnslist());  
88.                     system.out.println("-------> after");  
89.                     printcolumn(rowdata.getaftercolumnslist());  
90.                 }  
91.             }  
92.         }  
93.     }  
94.   
95.     private static void printcolumn(@notnull list columns) {  
96.         for (column column : columns) {  
97.             system.out.println(column.getname()   " : "   column.getvalue()   "    update="   column.getupdated());  
98.         }  
99.     }  
100. }  

（5）运行java实例

启动后看到控制端信息：

[html] view plain copy  

1. empty count : 1  
2. empty count : 2  
3. empty count : 3  
4. empty count : 4  

（6）触发数据库变更

[html] view plain copy  

1. create table test (  
2. uid int (4) primary key not null auto_increment,  
3. name varchar(10) not null);  
4.   
5. insert into test (name) values('10');  

（7）client 抓取mysql信息:

[html] view plain copy  

1. ================> binlog[mysql-bin.000016:3281] , name[canal_test,test] , eventtype : insert  
2. uid : 7    update=false  
3. name : 10    update=false  
4. empty count : 1  
5. empty count : 2  

5、部署过程中产生问题：

（1）启动失败，log日志中地址正在使用

1、11111端口正在被占用 可以用 ls -i:11111 查看监听进程谁占用端口 或者 用 ps -ef | grep 11111 查看哪个进程占用端口号  然后 kill -9 进程号  杀掉占用进程

2、可以编辑 canal/conf/canal.properties 中的端口号 ，改为不占用的端口

（2）canal无法抓取mysql触发数据库改变的信息

1、检查mysql是否打开binlog写入功能  检查binlog 是否为行模式。

[html] view plain copy  

1. show variables like "binlog_format"  

2、检查my.cnf 和 instance.properties 等配置文件填写信息是否正确。

3、检查client 代码 调试实例代码

4、版本兼容问题，canal 1.8 换成 canal 1.7 继续测试

5、查看所有日志文件 分析日志