elasticsearch使用canal同步mysql

- 日理万妓 2022-02-28 09:54 389阅读 0赞

早期，阿里巴巴B2B公司因为存在杭州和美国双机房部署，存在跨机房同步的业务需求。不过早期的数据库同步业务，主要是基于trigger的方式获取增量变更，不过从2010年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务，从此开启了一段新纪元。

canal官网：[https://github.com/alibaba/canal/wiki/Introduction][https_github.com_alibaba_canal_wiki_Introduction]

之前使用logstash同步es，但是发现效率很低，对数据库有一定压力，而canal是模拟mysql主从复制，监听mysql的日志变化，效率比logstach快很多。

1、部署canal服务端    2、springboot集成客户端

修改mysql配置，开启主从复制

vi  /etc/my.cnf

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2thbmc1Nzg5_size_16_color_FFFFFF_t_70][]

\[mysqld\]

log-bin=mysql-bin \#添加这一行就ok

binlog-format=ROW \#选择row模式

server\_id=1 \#配置mysql replaction需要定义，不能和canal的slaveId重复

canal的原理是模拟自己为mysql slave，所以这里一定需要做为mysql slave的相关权限.

mysql -uroot -p

CREATE USER canal IDENTIFIED BY 'canal';    
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON \*.\* TO 'canal'@'%';  
FLUSH PRIVILEGES;

保存，重启，service mysql restart

下载canal  [https://github.com/alibaba/canal/releases/download/canal-1.1.2/canal.deployer-1.1.2.tar.gz][https_github.com_alibaba_canal_releases_download_canal-1.1.2_canal.deployer-1.1.2.tar.gz]

mkdir -p /usr/local/canal

tar -zxvf canal.deployer-1.1.2.tar.gz   -C   /usr/local/canal

vi canal/conf/example/instance.properties

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2thbmc1Nzg5_size_16_color_FFFFFF_t_70 1][]

修改数据库地址address、代理用户名dbUsername、密码dbPassword、数据库defaultDatabaseName

如果系统是1个cpu，需要将canal.instance.parser.parallel设置为false  
启动    sh bin/startup.sh  
查看日志    tail -fn300 logs/canal/canal.log

集成客户端到springboot，依赖包

<dependency>
    			<groupId>com.alibaba.otter</groupId>
    			<artifactId>canal.client</artifactId>
    			<version>1.1.0</version>
    			<exclusions>
    				<exclusion>
    					<groupId>org.springframework</groupId>
    					<artifactId>spring</artifactId>
    				</exclusion>
    				<exclusion>
    					<groupId>commons-io</groupId>
    					<artifactId>commons-io</artifactId>
    				</exclusion>
    				<exclusion>
    					<groupId>ch.qos.logback</groupId>
    					<artifactId>logback-core</artifactId>
    				</exclusion>
    				<exclusion>
    					<groupId>ch.qos.logback</groupId>
    					<artifactId>logback-classic</artifactId>
    				</exclusion>
    				<exclusion>
    					<groupId>org.slf4j</groupId>
    					<artifactId>jcl-over-slf4j</artifactId>
    				</exclusion>
    				<exclusion>
    					<groupId>org.slf4j</groupId>
    					<artifactId>slf4j-api</artifactId>
    				</exclusion>
    				<exclusion>
    					<groupId>org.apache.zookeeper</groupId>
    					<artifactId>zookeeper</artifactId>
    				</exclusion>
    			</exclusions>
    		</dependency>

项目启动监听

package com.sync;
    
    import java.net.InetSocketAddress;
    
    import javax.annotation.Resource;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.boot.ApplicationArguments;
    import org.springframework.boot.ApplicationRunner;
    import org.springframework.stereotype.Component;
    
    import com.alibaba.otter.canal.client.CanalConnector;
    import com.alibaba.otter.canal.client.CanalConnectors;
    
    /**
     * 
    启动监听，同步mysql数据到es
     *
     */
    @Component
    public class SyncMysql2Es implements ApplicationRunner {
    
    	private static final Logger LOGGER = LoggerFactory.getLogger(SyncMysql2Es.class);
    
    	@Resource
    	private MyCanalClient client;
    
    	private String ip = "192.168.33.13";
    
    	@Override
    	public void run(ApplicationArguments arguments) throws Exception {
    		// 根据ip，直接创建链接，无HA的功能
    		String destination = "example";
    		CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111), destination, "", "");
    		client.setConnector(connector);
    		client.setDestination(destination);
    		client.start();
    
    		Runtime.getRuntime().addShutdownHook(new Thread() {
    			public void run() {
    				try {
    					LOGGER.info("## stop the canal client");
    					client.stop();
    				} catch (Throwable e) {
    					LOGGER.warn("##something goes wrong when stopping canal:", e);
    				} finally {
    					LOGGER.info("## canal client is down.");
    				}
    			}
    		});
    	}
    
    }

业务逻辑

package com.sync;
    
    import java.util.List;
    import java.util.Set;
    
    import javax.annotation.Resource;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.slf4j.MDC;
    import org.springframework.stereotype.Component;
    import org.springframework.util.Assert;
    
    import com.alibaba.otter.canal.client.CanalConnector;
    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.alibaba.otter.canal.protocol.Message;
    import com.google.common.collect.Sets;
    import com.jk.cloud.service.HcsSearchService;
    
    
    @Component
    public class MyCanalClient {
    
    	protected final static Logger logger = LoggerFactory.getLogger(MyCanalClient.class);
    
    	@Resource
    	private HcsSearchService hcsSearchService;
    
    	protected volatile boolean running = false;
    	protected Thread thread = null;
    	protected CanalConnector connector;
    	protected String destination;
    
    	public void setConnector(CanalConnector connector) {
    		this.connector = connector;
    	}
    
    	public void setDestination(String destination) {
    		this.destination = destination;
    	}
    
    	protected void start() {
    		Assert.notNull(connector, "connector is null");
    		thread = new Thread(new Runnable() {
    			public void run() {
    				process();
    			}
    		});
    
    		running = true;
    		thread.start();
    	}
    
    	protected void stop() {
    		if (!running) {
    			return;
    		}
    		running = false;
    		if (thread != null) {
    			try {
    				thread.join();
    			} catch (InterruptedException e) {
    				// ignore
    			}
    		}
    		MDC.remove("destination");
    	}
    
    	protected void process() {
    		int batchSize = 1024;
    		while (running) {
    			try {
    				MDC.put("destination", destination);
    				connector.connect();
    				connector.subscribe();
    				while (running) {
    					Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
    					long batchId = message.getId();
    					int size = message.getEntries().size();
    					if (batchId == -1 || size == 0) {
    						// try {
    						// Thread.sleep(1000);
    						// } catch (InterruptedException e) {
    						// }
    					} else {
    						syncData(message.getEntries());
    					}
    
    					connector.ack(batchId); // 提交确认
    					// connector.rollback(batchId); // 处理失败, 回滚数据
    				}
    			} catch (Exception e) {
    				logger.error("process error!", e);
    			} finally {
    				connector.disconnect();
    				MDC.remove("destination");
    			}
    		}
    	}
    
    	protected void syncData(List<Entry> entrys) {
    		for (Entry entry : entrys) {
    			if (entry.getEntryType() == EntryType.ROWDATA) {
    				RowChange rowChage = null;
    				try {
    					rowChage = RowChange.parseFrom(entry.getStoreValue());
    				} catch (Exception e) {
    					throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
    				}
    
    				EventType eventType = rowChage.getEventType();
    				String tableName = entry.getHeader().getTableName();
    
    				if (eventType == EventType.DELETE || eventType == EventType.INSERT || eventType == EventType.UPDATE) {
    					// 更新商品
    					if ("t_hcs_sku".equals(tableName)) {
    						Set<Integer> ids = Sets.newHashSet();
    						List<RowData> rowDatasList = rowChage.getRowDatasList();
    						for (RowData rowData : rowDatasList) {
    							List<Column> columnsList = rowData.getBeforeColumnsList();
    							for (Column column : columnsList) {
    								if ("sku_id".equals(column.getName())) {
    									String id = column.getValue();
    									ids.add(Integer.valueOf(id));
    								}
    								break;
    							}
    							columnsList = rowData.getAfterColumnsList();
    							for (Column column : columnsList) {
    								if ("sku_id".equals(column.getName())) {
    									String id = column.getValue();
    									ids.add(Integer.valueOf(id));
    								}
    								break;
    							}
    						}
    						hcsSearchService.syncSku(ids);
    					} else if ("t_hcs_seller".equals(tableName)) {
    						// 更新商家
    						Set<Integer> ids = Sets.newHashSet();
    						for (RowData rowData : rowChage.getRowDatasList()) {
    							List<Column> columnsList = rowData.getBeforeColumnsList();
    							for (Column column : columnsList) {
    								if ("seller_id".equals(column.getName())) {
    									String id = column.getValue();
    									ids.add(Integer.valueOf(id));
    								}
    								break;
    							}
    							columnsList = rowData.getAfterColumnsList();
    							for (Column column : columnsList) {
    								if ("seller_id".equals(column.getName())) {
    									String id = column.getValue();
    									ids.add(Integer.valueOf(id));
    								}
    								break;
    							}
    						}
    						hcsSearchService.syncSeller(ids);
    					}
    				}
    			}
    		}
    	}
    
    }

修改数据，看下效果

[https_github.com_alibaba_canal_wiki_Introduction]: https://github.com/alibaba/canal/wiki/Introduction
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2thbmc1Nzg5_size_16_color_FFFFFF_t_70]: /images/20220228/02b6464364564971af2316d2ce54b069.png
[https_github.com_alibaba_canal_releases_download_canal-1.1.2_canal.deployer-1.1.2.tar.gz]: https://github.com/alibaba/canal/releases/download/canal-1.1.2/canal.deployer-1.1.2.tar.gz
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2thbmc1Nzg5_size_16_color_FFFFFF_t_70 1]: /images/20220228/78454ca7f6fc45ada062342dee42ec08.png