Canal作为消息中间件实时消费MySQL中新增数据

#主服务器的id
server-id=1
#启用二进制日志
log-bin=mysql-bin
#设置不复制的数据库(选配)
binlog-ignore-db=mysql
#设置要复制的数据库(选配)
binlog-do-db=需要复制的主数据库名字（设置一个之前没有的数据库）
#设置logbin的格式
binlog_format=row

service mysql restart

mysql> show variables like ‘log_%‘;

#//这个id不能跟mysql中配置的id相同！！！
canal.instance.mysql.slaveId=100
//mysql地址
canal.instance.master.address=hadoop102:3306
#连接数据库所需要的用户名和密码
canal.instance.dbUsername=root
canal.instance.dbPassword=123
canal.instance.connectionCharset = UTF-8
canal.instance.defaultDatabaseName =
# table regex
//要监听的数据库，用正则表达式，这里表示gmall数据库中的所有表
canal.instance.filter.regex=gmall\\..*
# table black regex
//黑名单
canal.instance.filter.black.regex=

bin/startup.sh

bin/stop.sh

<dependencies>

<dependency>
<groupId>com.alibaba.ottergroupId>
<artifactId>canal.clientartifactId>
<version>1.1.2version>
dependency>


<dependency>
<groupId>org.apache.kafkagroupId>
<artifactId>kafka-clientsartifactId>
<version>0.11.0.0version>
dependency>
dependencies>

import java.net.InetSocketAddress
import java.util
import com.alibaba.fastjson.JSONObject
import com.alibaba.otter.canal.client.{CanalConnector, CanalConnectors}
import com.alibaba.otter.canal.protocol.CanalEntry.{EntryType, EventType, RowChange}
import com.alibaba.otter.canal.protocol.{CanalEntry, Message}
import com.atguigu.gmall.common.Constant
import com.google.protobuf.ByteString
/**
* Author atguigu
* Date 2020/5/30 15:29
*/
object CanalClient {
// 真正的处理数据
def parseData(rowDataList: util.List[CanalEntry.RowData],
tableName: String,
eventType: CanalEntry.EventType): Unit = {
// 计算订单总额 ,每在order_info表中插入一条数据就发送给kafka
if(tableName == "order_info" && eventType == EventType.INSERT && rowDataList != null && rowDataList.size() > 0){
import scala.collection.JavaConversions._
for(rowData <- rowDataList){
val result: JSONObject = new JSONObject()
// 一个rowData表示一行数据, 所有列组成一个json对象, 写入到Kafka中
val columnList: util.List[CanalEntry.Column] = rowData.getAfterColumnsList
for(column <- columnList){ // column 列
val key: String = column.getName // 列名
val value: String = column.getValue // 列值
result.put(key, value)
}
// 把数据写入到kafka中. 用一个生产者
MykafkaUtil.send(Constant.ORDER_INFO_TOPIC, result.toJSONString)
}
}

}

def main(args: Array[String]): Unit = {
// 1. 连接到canal服务器
// 1.1 canal服务器的地址 canal服务器的端口号
val address: InetSocketAddress = new InetSocketAddress("hadoop102", 11111)
val connector: CanalConnector = {
CanalConnectors.newSingleConnector(address, destination="example", username="", password="")
}
// 1.2 连接到canal
connector.connect()
// 2. 订阅你要处理的具体表 gmall1128下所有的表
connector.subscribe("gmall.*")

// 3. 读取数据, 解析
while (true) {
// 一致监听mysql数据变化, 所以这个地方不挺
// 100表示最多一次拉取由于100条sql导致的数据的变化
val msg: Message = connector.get(100)
val entries: util.List[CanalEntry.Entry] = msg.getEntries
if (entries != null && entries.size() > 0) {
// 遍历拿到每个entry
import scala.collection.JavaConversions._
for (entry <- entries) {
// 处理的EntryType应该时刻RowData
if (entry != null && entry.hasEntryType && entry.getEntryType == EntryType.ROWDATA) {
// 获取storeValue. 每个entry一个
val storeValue: ByteString = entry.getStoreValue
// 每个storeVales一个RowChange
val rowChange: RowChange = RowChange.parseFrom(storeValue)
// 每个rowChange中多个RowData. 一个RowData就表示一行数据
val rowDataList: util.List[CanalEntry.RowData] = rowChange.getRowDatasList
//调用处理数据的方法，在这里对每行的数据进行真正的处理
parseData(rowDataList, entry.getHeader.getTableName, rowChange.getEventType)
}
}
} else {
println("没有拉倒数据, 2s之后继续拉....")
Thread.sleep(2000)
}
}
}
}