4000

 

管道-海量数据导入

由于做性能测试，需要往redis中导出千万级的数据。

得知redis-cli工具支持pipeline导入可以达到最佳性能。测试下500万条命令导入耗时43秒。

 

 

格式要求

官方文档：http://redis.io/topics/mass-insert

数据格式要求：

1. 以*开始
2. *n     n代表此条命令分成n个部分
3. 每个部分以\r\n结束

 

set name tony 表达为：

[code]*3\r\n

$3\r\n

set\r\n

$4\r\n

name\r\n

$4\r\n

tony\r\n

注意：此处的\r\n为换行符，不是输入的字符。

 

 

示例

[code]package redis;

import java.io.BufferedWriter;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.UnsupportedEncodingException;

import org.junit.Test;

public class TestRedisPipe {
/**
* 格式化成输入字符串
*/
private String getString(String... args) {
StringBuilder sb = new StringBuilder();
sb.append("*").append(args.length).append("\r\n");
for (String arg : args) {
sb.append("$").append(arg.length()).append("\r\n");
sb.append(arg).append("\r\n");
}
return sb.toString();
}

@Test
public void initFile2() {
Long startTime = System.currentTimeMillis();
String file = "d:\\d.txt";
BufferedWriter w = null;
StringBuilder sb = new StringBuilder();
try {
w = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(file), "utf-8"));
for(int i=100000000 ;i < 100100000;i++){
//for (int i = 1; i <= 100; i++) {
if (i / 30000 == 0) {
w.flush();
}
sb.setLength(0);
sb.append(this.getString("set", "u" + i, "name" + i));
//sb.append(this.getString("hmset", "usr" + i, "userid", "usr" + i, "username", "usrname" + i));
w.append(sb.toString());
}
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
w.flush();
w.close();
} catch (IOException e) {
e.printStackTrace();
}

}
long endTime = System.currentTimeMillis();
System.out.println("耗时: "+(endTime - startTime)/1000+" s。");
}
}

 

 

常见问题

[root@localhost redis]# cat d.txt |redis-cli --pipe

ERR Protocol error: too big mbulk count string

Error writing to the server: Connection reset by peer

文件太大，和所分配的内存大小密切相关，内存太少则会导致文件太大导入失败。

 

安装两个服务

 

 

打开6379端口

[code]/sbin/iptables -I INPUT -p tcp --dport 6379 -j ACCEPT

/etc/rc.d/init.d/iptables save     #修改生效

/etc/init.d/iptables status        #查看配置

 

 

复制改端口无需再次安装

只需要复制配置文件，启动时选择配置文件即可。

[code]cd /usr/local/src/redis/redis.2.8.17

cp redis.conf redis6380.conf

vi redis6380.conf    #修改端口为6380

redis-server redis6380.conf

注意：启动后，会残留些数据，不完全，必须flushall清除掉。

 

 

简洁开启实例

[code]redis-server --port 6300 --daemonize yes

 

 

开放远程访问

redis.conf中bind默认绑定127.0.0.1，只有本地可以访问。

[code]ps -ef |grep redis

root  2545  2532  005:51 pts/0   00:00:07 redis-server *:6379

root  2710  2674  006:14 pts/2   00:00:05 redis-server 127.0.0.1:6479

讲bind 127.0.0.1注释掉，前面加个#即可

root  2545  2532  005:51 pts/0   00:00:07 redis-server *:6379

root  2710  2674  006:14 pts/2   00:00:05 redis-server *:6479

    变成两个*即可远程访问，可以看出默认的redis.conf和复制后的文件还是有差异的。是个坑啊。

 

Redis分片

访问redis的驱动包。

使用最为广泛的是Jedis和Redisson（官方推荐），在企业中采用最多的是Jedis，我们重点学习Jedis。

Jedis官网地址：https://github.com/xetorthio/jedis

 

 

第一个jedis示例

[code]package redis;

import java.util.List;

import redis.clients.jedis.Jedis;

public class TestRedis {
public static void main(String[] args) {
//设置连接服务器IP地址和访问端口
Jedis jedis = new Jedis("192.168.115.115",6379);

//单个值
//jedis.set("test", "456789");				//设置值
//System.out.println(jedis.get("test"));		//获取值

//多个值
//jedis.mset("test1","1","test2","2");
List<String> oList = jedis.mget("test1","test2");
for(String s : oList){
System.out.println(s);
}

jedis.close();	//关闭
}
}

命令窗口：

[code]127.0.0.1:6379> keys *
1) "bomb"
127.0.0.1:6379> get bomb
"tnt"
127.0.0.1:6379>

 

 

连接池JedisPool创建jedis连接

[code]package cn.redis;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class JedisPoolDemo {

public static void main(String[] args) {
// 构建连接池配置信息
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
// 设置最大连接数
jedisPoolConfig.setMaxTotal(200);

// 构建连接池
JedisPool jedisPool = new JedisPool(jedisPoolConfig, "127.0.0.1", 6379);

// 从连接池中获取连接
Jedis jedis = jedisPool.getResource();

// 读取数据
System.out.println(jedis.get("bomb"));

// 将连接还回到连接池中
jedisPool.returnResource(jedis);

// 释放连接池
jedisPool.close();
}
}

 

 

分片ShardedJedisPool

实现分布式缓存，Redis多个节点的透明访问

[code]@Test	//分片
public void shard(){

//构造各个节点链接信息，host和port
List<JedisShardInfo> infoList = new ArrayList<JedisShardInfo>();
JedisShardInfo info1 = new JedisShardInfo("192.168.163.200",6379);
//info1.setPassword("123456");
infoList.add(info1);
JedisShardInfo info2 = new JedisShardInfo("192.168.163.200",6380);
infoList.add(info2);
JedisShardInfo info3 = new JedisShardInfo("192.168.163.200",6381);
infoList.add(info3);

//分片jedis

JedisPoolConfig config = new JedisPoolConfig();
config.setMaxTotal(500);	//最大链接数

ShardedJedisPool pool = new ShardedJedisPool(config, infoList);
//ShardedJedis jedis = new ShardedJedis(infoList);
ShardedJedis jedis = pool.getResource();	//从pool中获取
for(int i=0;i<10;i++){
jedis.set("n"+i, "t"+i);
}
System.out.println(jedis.get("n9"));
jedis.close();
}

 

 

数据倾斜

3个节点，可以看到n为key时会发生数据倾斜，而换成text就缓解很多。

[code]	redis			CRC16
name+I		43/29/27		38/26/35
text+I		29/34/36		28/35/36

CRC16hash测试

[code]package redis;

import java.util.ArrayList;
import java.util.List;

import org.junit.Test;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;
public class Crc16Mod {

@Test
public void runCrc() {
for (int i = 1; i < 100; i++) {
System.out.println(this.getCrc(("name" + i).getBytes()) % 3);
}
}

private static Integer getCrc(byte[] data) {
int high;
int flag;

// 16位寄存器，所有数位均为1
int wcrc = 0xffff;
for (int i = 0; i < data.length; i++) {
// 16 位寄存器的高位字节
high = wcrc >> 8;
// 取被校验串的一个字节与 16 位寄存器的高位字节进行“异或”运算
wcrc = high ^ data[i];

for (int j = 0; j < 8; j++) {
flag = wcrc & 0x0001;
// 把这个 16 寄存器向右移一位
wcrc = wcrc >> 1;
// 若向右(标记位)移出的数位是 1,则生成多项式 1010 0000 0000 0001 和这个寄存器进行“异或”运算
if (flag == 1)
wcrc ^= 0xa001;
}
}

// return Integer.toHexString(wcrc);
return wcrc;
}
}

 

阅读更多