1.利用Redis-Replicator实现Redis-Cluster集群同步,源端备端都是三主三从,分布在三台CentOS7.6,4核8G的linux服务器上。在单机同步成功的基础上,开了三个线程,分别对源备的三个master进行点对点同步,结果源端1000w数据,备端只同步了300w不到,但是每个线程启动后立刻sleep 30s左右,1000w数据一个不少
public class MigrationExample {
public static void main(String[] args) throws IOException, URISyntaxException {
sync("redis://127.0.0.1:6379", "redis://127.0.0.1:6380");
}
/*
* Precondition:
* 1. Make sure the two redis version is same.
* 2. Make sure the single key-value is not very big.(highly recommend less then 1 MB)
*
* We running following steps to sync two redis.
* 1. Get rdb stream from source redis.
* 2. Convert source rdb stream to redis dump format.
* 3. Use Jedis RESTORE command to restore that dump format to target redis.
* 4. Get aof stream from source redis and sync to target redis.
*/
public static void sync(String sourceUri, String targetUri) throws IOException, URISyntaxException {
RedisURI suri = new RedisURI(sourceUri);
RedisURI turi = new RedisURI(targetUri);
final ExampleClient target = new ExampleClient(turi.getHost(), turi.getPort());
Configuration tconfig = Configuration.valueOf(turi);
if (tconfig.getAuthPassword() != null) {
Object auth = target.send(AUTH, tconfig.getAuthPassword().getBytes());
System.out.println("AUTH:" + auth);
}
final AtomicInteger dbnum = new AtomicInteger(-1);
Replicator r = dress(new RedisReplicator(suri));
r.addEventListener(new EventListener() {
@Override
public void onEvent(Replicator replicator, Event event) {
if (event instanceof DumpKeyValuePair) {
DumpKeyValuePair dkv = (DumpKeyValuePair) event;
// Step1: select db
DB db = dkv.getDb();
int index;
if (db != null && (index = (int) db.getDbNumber()) != dbnum.get()) {
target.send(SELECT, toByteArray(index));
dbnum.set(index);
System.out.println("SELECT:" + index);
}
// Step2: restore dump data
if (dkv.getExpiredMs() == null) {
Object r = target.restore(dkv.getKey(), 0L, dkv.getValue(), true);
System.out.println(r);
} else {
long ms = dkv.getExpiredMs() - System.currentTimeMillis();
if (ms <= 0) return;
Object r = target.restore(dkv.getKey(), ms, dkv.getValue(), true);
System.out.println(r);
}
}
if (event instanceof DefaultCommand) {
// Step3: sync aof command
DefaultCommand dc = (DefaultCommand) event;
Object r = target.send(dc.getCommand(), dc.getArgs());
System.out.println(r);
}
}
});
r.addCloseListener(new CloseListener() {
@Override
public void handle(Replicator replicator) {
target.close();
}
});
r.open();
}
public static class ExampleClient implements Closeable {
private Jedis jedis;
public ExampleClient(final String host, final int port) {
DefaultJedisClientConfig.Builder config = DefaultJedisClientConfig.builder();
config.timeoutMillis(10000);
this.jedis = new Jedis(new HostAndPort(host, port), config.build());
}
public Object send(Protocol.Command cmd, final byte[]... args) {
Object r = jedis.sendCommand(cmd, args);
if (r instanceof byte[]) {
return Strings.toString(r);
} else {
return r;
}
}
public Object send(final byte[] cmd, final byte[]... args) {
return send(Protocol.Command.valueOf(Strings.toString(cmd).toUpperCase()), args);
}
public Object restore(byte[] key, long expired, byte[] dumped, boolean replace) {
if (!replace) {
return send(RESTORE, key, toByteArray(expired), dumped);
} else {
return send(RESTORE, key, toByteArray(expired), dumped, "REPLACE".getBytes());
}
}
@Override
public void close() {
if (jedis != null) {
jedis.close();
}
}
}
这段代码实现了单机源备同步,现在把它单独写在一个Thread类里的run方法中,有三台源备开三个线程同时同步,CPU百分之两百多,备端同步数据不全。想定位到哪里资源耗尽了导致程序异常或者让这三个线程能顺序执行,完成任务