mysql+canal+kafkaデータリアルタイム同期

文書ディレクトリ

ビジネスシーン

zkはcanal serverの高可用性を保証し、同じ時間にcanal-serverが本当に働いているのは1つだけです.

クラスタ環境

zookeeper導入&構成

ダウンロード:

解凍

変更構成

はそれぞれ3台の機械で検証zk

を起動する.

kafka導入&構成

canal導入&構成

ダウンロード(https://github.com/alibaba/canal/releases)

転載先:https://www.jianshu.com/p/c0b19ff9f140
ビジネスシーン
会社はソーシャル電子商取引をしていて、いくつかの業務シーンはユーザーにいろいろなタイプの収益を送って、それからユーザーはappの上で自分のいろいろな収益と収益の明細をまとめることができます.プロジェクト運営初期にはユーザ数が少なく,収益明細が少なかったため,ユーザがappで収益要約を取得する際に性能に関する問題はなかった.その後、ユーザー数の増加に伴い、ビジネスの拡張により、ユーザー収益データが増加し、ユーザーが収益要約情報にアクセスするたびに、リアルタイムでsql要約クエリを介してパフォーマンスに問題が発生します.ソリューションA:オフライン要約+キャッシュB:要約テーブル+ビジネスロジックリアルタイム【メッセージキュー等】要約テーブルC:mysql+canal+kafkaリアルタイム計算に書き込む
Aシナリオの問題:リアルタイム性が不足し、ユーザーが収益を得た後、明細は存在するが、要約情報に追加されていない.また,業務に利用可能な残高で支払うシーンがあり,ユーザや業務のニーズを完全に満たすことができない.
B方案問題:業務結合、各業務ラインはデータのリアルタイム問題に注目する.業務ラインが多くなるとシーンの一部が抜けてしまう可能性があります.
C方案:業務のデカップリング、業務の開発時に過度に収益の総括問題に関心を持つ必要がない.リアルタイム性は比較的に強くて、収益データは正常にユーザーに発生して基本的にミリ秒レベルmysql+canal+kafkaのリアルタイム計算の紹介を見て、ユーザーの収益データがmysqlに書き込むことを生成して、canalはmysqlの書く操作を監視して、binlogログを解析してkafkaに書きます.消費者(python)はkafkaメッセージを消費し、具体的な業務要約データから要約表に、appは直接要約表の記録を読み取ることができる.
zkはcanal serverの高可用性を保証し、同じ時間にcanal-serverが本当に働いているのは1つだけです.
クラスタ環境
デフォルトjdk環境はすべてOKに設定されています

192.168.0.81 (zk/kafka/canal)
192.168.0.82 (zk/kafka/canal)
192.168.0.83 (zk/kafka/kafka-consumer)

zookeeperの導入&構成
クラスタ・モードの導入
ダウンロード:

wget https://www.apache.org/dist/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz

解凍

tar -zxvf zookeeper-3.4.14.tar.gz
mv zookeeper-3.4.14 /usr/local/zk

構成の変更

vi /usr/local/zk/conf/zoo.cfg
  tickTime=2000
  initLimit=10
  syncLimit=5
  dataLogDir=/opt/zookeeper/logs
  dataDir=/opt/zookeeper/data
  clientPort=2181
  autopurge.snapRetainCount=500
  autopurge.purgeInterval=24
  server.1= 192.168.0.81:2888:3888
  server.2= 192.168.0.82:2888:3888
  server.3= 192.168.0.83:2888:3888 

ディレクトリ/opt/zookeeper/dataで実行

192.168.0.81  echo "1" >> myid
192.168.0.82  echo "2" >> myid
192.168.0.83  echo "3" >> myid

それぞれ3台の機械で検証zkを起動する

cd /usr/local/zk/bin
# 1、  
./zkServer.sh start
# 2、  
./zkServer.sh stop
# 3、    
./zkServer.sh status
# 4、  
./zkServer.sh restart

kafka配置&構成
対応するバージョンを/usr/local/kafkaディレクトリに解凍する:構成変更プロファイル/usr/local/kafka/config/serverを構成する.properties主な構成項目

192.168.0.81 
 broker.id=1
 advertised.listeners=PLAINTEXT://192.168.0.81:9092
 zookeeper.connect=192.168.0.81:2181,192.168.0.82:2181,192.168.0.83:2181
192.168.0.82 
 broker.id=2
 advertised.listeners=PLAINTEXT://192.168.0.82:9092
 zookeeper.connect=192.168.0.81:2181,192.168.0.82:2181,192.168.0.83:2181
192.168.0.83 
 broker.id=3
 advertised.listeners=PLAINTEXT://192.168.0.83:9092
 zookeeper.connect=192.168.0.81:2181,192.168.0.82:2181,192.168.0.83:2181
  ：
 cd /usr/local/kafka/bin; ./kafka-server-start.sh -daemon config/server.properties &

kafkaに関する操作は他の記事を参照してください
canal導入&構成
ダウンロード(https://github.com/alibaba/canal/releases)
インストール、サービス側のパッケージのみ必要

tar zxvf canal.deployer-1.1.3-SNAPSHOT.tar.gz  -C /opt/canal

構成/usr/local/canal/conf/canal.propertiesの主な構成項目:

canal.id = 1 （192.168.0.81  1， 192.168.0.82  2，canal HA  ）
canal.zkServers =192.168.0.81:2181,192.168.0.82:2181,192.168.0.83:2181
canal.serverMode = kafka
canal.destinations = test #canal.properties      test             。         
canal.mq.servers = 192.168.0.81:2181,192.168.0.82:2181,192.168.0.83:2181
  /usr/local/canal/conf/test/instance.properties
canal.instance.master.address=192.168.0.30:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
canal.instance.defaultDatabaseName = test #        
canal.instance.filter.regex=test\\.commission
canal.mq.topic=commission
canal.mq.partitionsNum=3
canal.mq.partitionHash=commission:id
  
cd /usr/local/canal/bin
sh ./start.sh