最近实习的时候被要求实现这一需求做个小demo，于是做完后记录一下做个小小的总结
相关链接：
相关知识介绍
Ubuntu用户
1. MySQL初始化
注：默认已安装MySQL
通过 sudo vim /etc/mysql/mysql.conf.d/mysqld.cnf 打开文件
在 [mysqld] 下加入
log-bin=mysql-bin
binlog-format=ROW
server_id=12. 安装部署canal
2.1. 下载文件
通过如下指令下载并解压
wget https://github.com/alibaba/canal/releases/download/canal-1.1.3/canal.deployer-1.1.3.tar.gz
mkdir /tmp/canal
tar zxvf canal.deployer-$version.tar.gz -C /tmp/canal解压完成后可看到该目录下有
- bin(运行文件)
- conf(配置文件)
- lib(使用的包)
- logs(日志文件)
2.2. 修改配置
输入 vim conf/example/instance.properties 打开配置文件，修改如下部分
# position info
canal.instance.master.address= # 填写自己的数据库IP和端口
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=
# username/password
canal.instance.dbUsername=canal # 填写数据库用户名
canal.instance.dbPassword=canal # 填写数据库密码
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.defaultDatabaseName = test # 设置监听数据库名注：如果系统是1个cpu，需要将canal.instance.parser.parallel设置为false
2.3. 准备启动
Ubuntu用户可通过 sh bin/start.sh 启动
可通过如下指令查看日志
vi logs/canal/canal.log
vi logs/example/example.log2.4. 关闭canal服务
输入 sh bin/stop.sh 即可关闭
3. 编写canal client获取日志推送
3.1.下载示例
可下载此文件，解压后使用 IDEA 打开，等下载完依赖文件即可使用
3.2. 自定义client
首先创建链接，然后获取指定数量的数据，若数据为空，则等待一秒后发送确认信息，否则打印信息详情，连续120次获取数据为空时退出程序。
完整代码：
package com.alibaba.otter.canal.example;
import java.net.InetSocketAddress;
import java.util.List;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.Message;
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.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.google.protobuf.InvalidProtocolBufferException;
public class MyCanalClient {
public static void main(String[] args){
// 创建链接 CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(), 11111), "example", "canal", "canal");
int batchSize = 1000;
int emptyCount = 0;
try {
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();
int totalEmptyCount = 120;
while (emptyCount < totalEmptyCount) {
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据 long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
emptyCount++;
System.out.println("empty count : " + emptyCount);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
} else {
emptyCount = 0;
// System.out.printf("message[batchId=%s,size=%s] \n", batchId, size); printEntry(message.getEntries());
}
connector.ack(batchId); // 提交确认 // connector.rollback(batchId); // 处理失败, 回滚数据 }
System.out.println("empty too many times, exit");
} finally {
connector.disconnect();
}
}
private static void printEntry(List<Entry> entries){
for(Entry entry : entries){
if(entry.getEntryType()==EntryType.TRANSACTIONBEGIN || entry.getEntryType()==EntryType.TRANSACTIONEND){
continue;
}
RowChange rowChange = null;
try {
rowChange = RowChange.parseFrom(entry.getStoreValue());
} catch (InvalidProtocolBufferException e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
}
EventType eventType = rowChange.getEventType();
System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s", entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(), entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));
for(RowData rowData : rowChange.getRowDatasList()){
if (eventType == EventType.DELETE) {
printColumn(rowData.getBeforeColumnsList());
} else if (eventType == EventType.INSERT) {
printColumn(rowData.getAfterColumnsList());
} else {
System.out.println("-------> before");
printColumn(rowData.getBeforeColumnsList());
System.out.println("-------> after");
printColumn(rowData.getAfterColumnsList());
}
}
}
}
private static void printColumn(List<Column> columns){
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}打开 canal 服务后执行上述代码，往数据库中插入/修改数据便可在控制台看到日志信息。
4. 安装配置kafka
4.1. 下载文件
通过此链接可下载最新版本的kafka安装包
4.2. 修改配置文件
解压后可看到该目录下有
- bin(执行文件)
- config(配置文件)
- libs(包文件)
- site-docs
- LICENCE
- NOTICE
在 config 文件夹下有很多配置文件，其中
- consumer.properties 消费者配置文件
- producer.properties 生产者配置文件
- server.properties kafka服务器配置文件
注：高版本kafka自带zookeeper，使用默认配置即可
4.3. 运行
运行zookeepr 进入 kafka 解压目录
bin/zookeeper-server-start.sh config/zookeeper.properties
运行 kafka
bin/kafka-server-start.sh config/server.properties
其他相关指令：
创建一个名为 test 的 topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
查看已创建的 topic
bin/kafka-topics.sh --list --zookeeper localhost:2181
创建用于消费 topic 为 test 的消费者
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
创建消息生产者
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
5. 在Java程序中使用kafka
5.1. 创建 topic
public void CreateTopic(String topicName, int numPartitions, short replicationFactor){
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
AdminClient adminClient = AdminClient.create(props);
ArrayList<NewTopic> topics = new ArrayList<NewTopic>();
NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
topics.add(newTopic);
CreateTopicsResult res = adminClient.createTopics(topics);
try{
res.all().get();
}catch (InterruptedException | ExecutionException e){
e.printStackTrace();
}
}5.2. 生产者发送消息
public void publishMessage(String text, String topicName){
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<String, String>(props);
producer.send(new ProducerRecord<String, String>(topicName, text));
producer.close();
}5.3. 消费者消费消息
public void ConsumeMessage(String topicName, String groupId){
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", groupId);
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
final KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);
consumer.subscribe(Arrays.asList(topicName),new ConsumerRebalanceListener() {
public void onPartitionsRevoked(Collection<TopicPartition> collection) {
}
public void onPartitionsAssigned(Collection<TopicPartition> collection) {
//将偏移设置到最开始 consumer.seekToBeginning(collection);
}
});
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
}注：查看最终运行效果需打开canal服务、zookeeper服务、kafka服务，同时注意加入相应的jar包
Windows用户
相关操作与Ubuntu类似，仅指出不同的地方
MySQL配置文件路径为 C:\ProgramData\MySQL\MySQL Server 8.0\my.ini (文件路径可能随MySQL版本有所不同)
canal 可通过此链接下载并解压
canal 相关启动脚本在 Windows 中有相应的.bat脚本
kafka 相关启动脚本在 bin\windows\ 下有相应的.bat脚本
最后附上相应代码下载地址
