ApsaraMQ for Kafka:Sinkronisasi data MySQL ke ApsaraMQ for Kafka menggunakan Canal

Referensi parameter instance.properties

Akses Internet (SASL_SSL)

Saat terhubung melalui Internet, Canal melakukan otentikasi dengan ApsaraMQ for Kafka melalui SASL_SSL. Anda perlu mengubah tiga file.

File 1: conf/canal.properties

# Atur Canal untuk menggunakan Kafka sebagai output
canal.serverMode = kafka

# Titik akhir SSL instans ApsaraMQ for Kafka Anda
# Dapatkan ini dari bagian Informasi Titik Akhir di halaman Detail Instans
kafka.bootstrap.servers = alikafka-pre-cn-zv**********-1.alikafka.aliyuncs.com:9093,alikafka-pre-cn-zv**********-2.alikafka.aliyuncs.com:9093,alikafka-pre-cn-zv**********-3.alikafka.aliyuncs.com:9093

# Pengaturan produsen Kafka (sesuaikan jika diperlukan)
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0

# Otentikasi SASL_SSL
kafka.ssl.truststore.location = ../conf/kafka_client_truststore_jks
kafka.ssl.truststore.password = KafkaOnsClient
kafka.security.protocol = SASL_SSL
kafka.sasl.mechanism = PLAIN
kafka.ssl.endpoint.identification.algorithm =

File 2: bin/startup.sh

Tambahkan path konfigurasi JAAS ke variabel JAVA_OPTS:

JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8 -Djava.security.auth.login.config=/home/doc/tools/canal.deployer-1.1.5/conf/kafka_client_jaas.conf"

File 3: conf/kafka_client_producer_jaas.conf

Tentukan kredensial SASL untuk instans ApsaraMQ for Kafka Anda:

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="<your-sasl-username>"
  password="<your-sasl-password>";
};

Ganti placeholder dengan nilai aktual Anda:

Unduh file truststore SSL kafka.client.truststore.jks dan simpan ke direktori conf/.

Akses VPC (PLAINTEXT)

Saat terhubung dari dalam VPC, tidak diperlukan otentikasi atau enkripsi. Atur canal.serverMode dan kafka.bootstrap.servers di conf/canal.properties:

# Atur Canal untuk menggunakan Kafka sebagai output
canal.serverMode = kafka

# Titik akhir default (VPC) instans ApsaraMQ for Kafka Anda
# Dapatkan ini dari bagian Informasi Titik Akhir di halaman Detail Instans
kafka.bootstrap.servers = alikafka-pre-cn-zv**********-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-3-vpc.alikafka.aliyuncs.com:9092

# Pengaturan produsen Kafka (sesuaikan jika diperlukan)
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0

Referensi parameter canal.properties

Verifikasi bahwa Canal telah berjalan

Periksa log server Canal:

tail -f logs/canal/canal.log

Output yang diharapkan:

2013-02-05 22:45:27.967 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
2013-02-05 22:45:28.113 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.XX.XX:11111]
2013-02-05 22:45:28.210 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......

Periksa log instans Canal:

tail -f logs/example/example.log

Output yang diharapkan:

2013-02-05 22:50:45.636 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
2013-02-05 22:50:45.641 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
2013-02-05 22:50:45.803 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example
2013-02-05 22:50:45.810 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....

Uji dengan tabel sampel

1. Di database MySQL bernama mysql, buat tabel uji dan masukkan data sampel:

       mysql> SELECT * FROM T_Student;
       +--------+---------+------+------+
       | stuNum | stuName | age  | sex  |
       +--------+---------+------+------+
       |      1 | Wang    |   18 | girl |
       |      2 | Zhang   |   17 | boy  |
       +--------+---------+------+------+
       2 rows in set (0.00 sec)

2. Konfirmasi bahwa Canal menangkap perubahan dengan memeriksa log meta. Setiap INSERT, UPDATE, atau DELETE menghasilkan entri baru: Output yang diharapkan: Setiap baris menunjukkan file log biner (log.000001), offset byte (misalnya, 29723), dan alamat sumber MySQL—mengonfirmasi bahwa Canal membaca perubahan.

       tail -f logs/example/meta.log

       2020-07-29 09:21:05.110 - clientId:1001 cursor:[log.000001,29723,1591190230000,1,] address[/192.168.XX.XX:3306]
       2020-07-29 09:23:46.109 - clientId:1001 cursor:[log.000001,30047,1595985825000,1,] address[localhost/192.168.XX.XX:3306]

3. Verifikasi bahwa pesan telah tiba di ApsaraMQ for Kafka. Buka Konsol ApsaraMQ for Kafka dan kueri pesan untuk topik Anda. Untuk informasi lebih lanjut, lihat Kueri pesan.

   

4. Setelah selesai menguji, hentikan Canal:

       sh bin/stop.sh