Mysql通过binlog同步redis

实现过程

通过canal解析binlog，canal主要模拟了mysql的Slave向Master传送请求，当mysql有增删改查时则会出发请求将资料传送到canal服务中，canal将资料存放到内存，直到客户端程式(canal服务端和客户端程式都是由java编写,且客户端逻辑由我们借助com.alibaba.otter.canal工具包下的类完成开发)通过释出-订阅这种模式消费canal服务中的资料。

配置mysql

1.在my.ini档案里的【mysqld】下新增如下：

重启mysql，注意如果是档名是my-default.ini要改名成my.ini

2.建立mysql使用者，并配置canal许可权，进入到mysql程式中，输入下列命令，建立使用者，(使用者名称和密码都是：canal)：

CREATE USER canal IDENTIFIED BY 'canal';

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON数据库名.表名 TO 'canal'@'%';

-- GRANT ALL PRIVILEGES ON数据库名.表名 TO 'canal'@'%' ;

FLUSH PRIVILEGES;

下载并配置启动canal：

1.]官网地址：[https://github.com/alibaba/canal/releases，下载后解压

2.配置canal：主要配置的档案有两处，canal/conf/example/instance.properties和 canal/conf/canal.properties .下面对这两处做修改：

cd canal/conf/example/目录下对 instance.properties进行修改：

image.png

cd canal/conf/目录下对 canal.properties 进行修改：

3.启动canal：双击/bin/startup.bat档案启动

Java连线canal执行同步操作:

1.在maven专案中中载入canal和redis依赖包:

com.alibaba.otter

canal.client

1.0.12

redis.clients

jedis

2.4.2

2.建立canal客户端，从canal中获取资料，并将资料更新至Redis:

package canal;

import java.net.InetSocketAddress;

import java.util.List;

import com.alibaba.fastjson.JSONObject;

import com.alibaba.otter.canal.client.CanalConnector;

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.alibaba.otter.canal.client.*;

/**

* @Author: daijunchen

* @Date: 2019/3/12 17:30

*/

public class ClientSample {

public static void main(String args[]) {

// 建立连结，hostname位canal服务器ip port位canal服务器埠，username，password可不填

CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",

11111), "example", "", "");

int batchSize = 1000;

try {

connector.connect();

connector.subscribe(".*\..*");

connector.rollback();

while (true) {

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) {

e.printStackTrace();

}

} else {

printEntry(message.getEntries());

}

connector.ack(batchId); // 提交确认

// connector.rollback(batchId); // 处理失败, 回滚资料

}

} finally {

connector.disconnect();

}

}

private static void printEntry(List entrys) {

for (Entry entry : entrys) {

if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {

continue;

}

RowChange rowChage = null;

try {

rowChage = RowChange.parseFrom(entry.getStoreValue());

} catch (Exception e) {

throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),

e);

}

EventType eventType = rowChage.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 : rowChage.getRowDatasList()) {

if (eventType == EventType.DELETE) {

redisDelete(rowData.getBeforeColumnsList());

} else if (eventType == EventType.INSERT) {

redisInsert(rowData.getAfterColumnsList());

} else {

System.out.println("-------> before");

printColumn(rowData.getBeforeColumnsList());

System.out.println("-------> after");

redisUpdate(rowData.getAfterColumnsList());

printColumn(rowData.getAfterColumnsList());

}

}

}

}

/**

* 打印变化的资料

*

* @param columns

*/

private static void printColumn(List columns) {

for (Column column : columns) {

System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());

}

}

/**

* 资料插入同步redis

*

* @param columns

*/

private static void redisInsert(List columns) {

JSONObject json = new JSONObject();

for (Column column : columns) {

json.put(column.getName(), column.getValue());

}

if (columns.size() > 0) {

RedisUtil.stringSet("user:" + columns.get(0).getValue(), json.toJSONString());

}

}

/**

* 更新同步redis

*

* @param columns

*/

private static void redisUpdate(List columns) {

JSONObject json = new JSONObject();

for (Column column : columns) {

json.put(column.getName(), column.getValue());

}

if (columns.size() > 0) {

RedisUtil.stringSet("user:" + columns.get(0).getValue(), json.toJSONString());

}

}

/**

* 资料删除同步redis

*

* @param columns

*/

private static void redisDelete(List columns) {

JSONObject json = new JSONObject();

for (Column column : columns) {

json.put(column.getName(), column.getValue());

}

if (columns.size() > 0) {

RedisUtil.delKey("user:" + columns.get(0).getValue());

}

}

}

3.RedisUtil工具类，用来连线redis：

package canal;

import redis.clients.jedis.Jedis;

import redis.clients.jedis.JedisPool;

import redis.clients.jedis.JedisPoolConfig;

/**

* @Author: daijunchen

* @Date: 2019/3/12 17:29

*/

public class RedisUtil {

// Redis服务器IP

private static String ADDR = "127.0.0.1";

// Redis的埠号

private static int PORT = 6379;

// 访问密码

private static String AUTH = "";

// 可用连线例项的最大数目，预设值为8；

// 如果赋值为-1，则表示不限制；如果pool已经分配了maxActive个jedis例项，则此时pool的状态为exhausted(耗尽)。

private static int MAX_ACTIVE = 1024;

// 控制一个pool最多有多少个状态为idle(空闲的)的jedis例项，预设值也是8。

private static int MAX_IDLE = 200;

// 等待可用连线的最大时间，单位毫秒，预设值为-1，表示永不超时。如果超过等待时间，则直接丢掷JedisConnectionException；

private static int MAX_WAIT = 10000;

// 过期时间

protected static int expireTime = 660 * 660 * 24;

// 连线池

protected static JedisPool pool;

/**

* 静态程式码，只在初次呼叫一次

*/

static {

JedisPoolConfig config = new JedisPoolConfig();

//最大连线数

config.setMaxTotal(MAX_ACTIVE);

//最多空闲例项

config.setMaxIdle(MAX_IDLE);

//超时时间

config.setMaxWaitMillis(MAX_WAIT);

//

config.setTestOnBorrow(false);

pool = new JedisPool(config, ADDR, PORT, 1000);

}

/**

* 获取jedis例项

*/

protected static synchronized Jedis getJedis() {

Jedis jedis = null;

try {

jedis = pool.getResource();

} catch (Exception e) {

e.printStackTrace();

if (jedis != null) {

pool.returnBrokenResource(jedis);

}

}

return jedis;

}

/**

* 释放jedis资源

*

* @param jedis

* @param isBroken

*/

protected static void closeResource(Jedis jedis, boolean isBroken) {

try {

if (isBroken) {

pool.returnBrokenResource(jedis);

} else {

pool.returnResource(jedis);

}

} catch (Exception e) {

}

}

/**

* 是否存在key

*

* @param key

*/

public static boolean existKey(String key) {

Jedis jedis = null;

boolean isBroken = false;

try {

jedis = getJedis();

jedis.select(0);

return jedis.exists(key);

} catch (Exception e) {

isBroken = true;

} finally {

closeResource(jedis, isBroken);

}

return false;

}

/**

* 删除key

*

* @param key

*/

public static void delKey(String key) {

Jedis jedis = null;

boolean isBroken = false;

try {

jedis = getJedis();

jedis.select(0);

jedis.del(key);

} catch (Exception e) {

isBroken = true;

} finally {

closeResource(jedis, isBroken);

}

}

/**

* 取得key的值

*

* @param key

*/

public static String stringGet(String key) {

Jedis jedis = null;

boolean isBroken = false;

String lastVal = null;

try {

jedis = getJedis();

jedis.select(0);

lastVal = jedis.get(key);

jedis.expire(key, expireTime);

} catch (Exception e) {

isBroken = true;

} finally {

closeResource(jedis, isBroken);

}

return lastVal;

}

/**

* 新增string资料

*

* @param key

* @param value

*/

public static String stringSet(String key, String value) {

Jedis jedis = null;

boolean isBroken = false;

String lastVal = null;

try {

jedis = getJedis();

jedis.select(0);

lastVal = jedis.set(key, value);

jedis.expire(key, expireTime);

} catch (Exception e) {

e.printStackTrace();

isBroken = true;

} finally {

closeResource(jedis, isBroken);

}

return lastVal;

}

/**

* 新增hash资料

*

* @param key

* @param field

* @param value

*/

public static void hashSet(String key, String field, String value) {

boolean isBroken = false;

Jedis jedis = null;

try {

jedis = getJedis();

if (jedis != null) {

jedis.select(0);

jedis.hset(key, field, value);

jedis.expire(key, expireTime);

}

} catch (Exception e) {

isBroken = true;

} finally {

closeResource(jedis, isBroken);

}

}

}

验证修改mysql数据库内容后发现redis也实时更新了：