CentOS 6+Reids3.2.11多实例部署
单实例单线程的redis进程不足以高效率使用cpu和内存资源，所以一般来讲redis在同一台机器上要启动多个进程完成多实例部署；默认占用6379的情况下无法完成直接的3个实例启动，我们需要了解如何通过指定配置文件，将多实例部署在linux上
启动redis服务的命令redis-server 没有加载任何配置文件指定各种各样的配置信息（端口指定，ip绑定，后台运行，指定持久化文件等）
例如在根目录存在一个配置文件的模板（大部分与默认启动的配置相同）redis.conf
#redis-server 配置文件的名称
配置文件（在启动时指定的配置文件，核心的配置文件）
/redis根目录/redis.conf
一个redis实例默认占用所有物理内存（上限是物理内存大小）,在实际使用中需要限制大小
配置文件的修改内容
p61 bind 用#注释掉
如果需要绑定监听的ip（客户端只有通过被绑定的ip才可以利用redis-cli -h ip地址链接服务器）
bing 127.0.0.1 192.168.65.128(外网可访问当前服务器的ip)
一旦用#注释bind，没有任何限制，只要可以链接服务器，都允许使用redis
p80 保护模式不启动
保护模式开启，需要登录密码，改成no
p84 6379是默认端口(要启动其他的redis实例需要修改端口)
p105 当客户端空闲时间达到一小时,就会自动断开连接,0秒表示
不启用超时配置
p128 daemonize 设置成yes让redis服务器启动有守护进程管理
(后台执行)
p150 对应不同的redis实例,pid的文件名称需要和端口同名
每个进程在linux或者其他操作系统中都会占用pid号，当系统中的进程过多时，需要查找redis进程号可能比较麻烦，直接打开pid文件查看即可
P163logfile 需要指定，利用端口号命名，放到redis根目录
save 900(秒) 1(变动的数据条数)
当900以内,至少有1条数据变动,看是flush保存数据到文件
save 300 10
300秒以内至少10条数据变动,保存文件
save 60 10000
P237左右，指定dump的持久化文件，每个服务单独指向一个文件，
重启时，数据不会错乱
启动第二和第三个redis实例
redis-server redis.conf(指定启动文件)
需要第二个实例的配置文件
需要第三个实例的配置文件
拷贝redis.conf为redis_6379.conf、redis_6380.conf、redis_6381.conf
将拷贝的文件中只修改与端口有关内容
启动另外两个节点
redis-server redis_6379.conf
redis-server redis_6380.conf
redis-server redis_6381.conf
指定端口登录客户端redis-cli -p [端口号] -h [ip]
redis-cli -p 6379
redis-cli -p 6380
redis-cli -p 6381
ps -el | grep redis
此时就已经准备好了三个服务实例。
附上测试代码：
/*
java语言提供多种可以链接redis的包；
其中比较常用，更新比较快速；jedis
依赖jar包
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-redis</artifactId>
<version>1.4.5.RELEASE</version>
</dependency>
*/
/*
* 1 使用jedis链接redis服务端
*/
@Test
public void test01(){
//从代码链接任何技术，必须的参数
//ip port，新建一个jedis对象，获取ip和端口信息
//即可完成链接服务的操作
Jedis jedis=new Jedis("192.168.65.128", 6379);
//利用着一个链接对象，操作redis服务
//jedis.set("name", "王湛鲲");
System.out.println(jedis.get("city"));
jedis.close();
}
/*
* 2 模拟用户访问某个商品时使用缓存的逻辑
*
* * 接收到用户的一次商品访问的请求 打印输出
* * 根据商品携带的参数，定义全局唯一的key值
* key值的计算逻辑=“item_”+itemId 商品id
* * 判断缓存是否有数据
* 有：直接返回，做响应，不再访问数据库（第一次访问肯定访问数据库）
* 没有（说明是第一次访问该商品） 从数据库获取数据，存到缓存内存，将数据响应出去
*/
@Test
public void test02(){
String id="1";
String product="联想电脑 拯救者";
System.out.println("用户开始访问商品，商品id"+id);
//根据商品访问的业务逻辑，生成全局唯一的key对应商品的访问
String key="ITEM_"+id;//item_123
//链接redis
Jedis jedis=new Jedis("192.168.65.128", 6380);
//利用jedis客户端判断key是否存在
if(jedis.exists(key)){//表示存在缓存数据
String info=jedis.get(key);
System.out.println("数据从缓存获取，商品信息是："+info);
jedis.close();
}else{//缓存不存在数据，从数据库获取
System.out.println("select * from product where id="+id);
System.out.println("从数据库获取数据，并且存储到缓存");
jedis.set(key, product);
}
}
假设1000条商品信息都需要存储到缓存中，均衡的分配到每个节点中存储；
脑海中计算逻辑，对应实现代码里，这个计算步骤--数据分片的计算；
数据分片：数据层分布式集群中，数据被切分存储/读取，从多个服务器集群中的过程，必须对应正确严谨的计算逻辑，被切分的每一份数据都称为一个数据分片
@Test
public void test03(){
//准备jedis链接6379，6380，6381的客户端
Jedis jedis1=new Jedis("192.168.65.128", 6379);
Jedis jedis2=new Jedis("192.168.65.128", 6380);
Jedis jedis3=new Jedis("192.168.65.128", 6381);
//海量数据的存储需求
for(int i=0;i<1000;i++){
//每次循环，相当于需要存储一个商品信息，生成对应的key-value
String key="key_"+i;
String value="value_"+i;
if(i<333){
jedis1.set(key, value);//333 以下的数据存储到6379
}else if(i<666){
jedis2.set(key, value);//6380
}else{
jedis3.set(key, value);
}
}
//海量数据的获取需求
for(int i=0;i<1000;i++){
String key="key_"+i;
if(i<333){
System.out.println(jedis1.get(key));//333 以下的数据存储到6379
}else if(i<666){
System.out.println(jedis2.get(key));
}else{
System.out.println(jedis3.get(key));
}
}
}以上内容是根据数据情况（1000条商品）完成的自定义数据分片的计算逻辑实现；读和取必须保持同一个计算逻辑，才能正确的操作缓存分布式集群；
自定义的前提条件：必须对整体数据有所了解；
自定义的分片逻辑，是在早期数据增长数量不多，企业数据稳定的情况下，工程师根据企业具体情况定义的最高效的计算逻辑
随着数据的不断增长，和数据内容的不可控；
缺点：
1. 数据倾斜严重（修改自定义的计算逻辑）
2. key值的取值范围发生变动（改代码，重新定义计算逻辑）
需要一种能够随着数据发生变动，一劳永逸的计算逻辑
hash取余
计算公式：object表示任何一种类型的java对象,N表示分布式集群中的数据分片数量（3）
(object.hashCode()&Integer.MAX_VALUE)%N
保真计算 有符号数字最大值的二进制：首位0其余位是1 因此做与运算可以把负数转化为正数
@Test
public void test04() {
for (int i = 1; i <= 100; i++) {
String key = "";
if (i > 1)
{
key = "wzk又收到了苹果:" + i + "个!目前拥有" + (1 + i) * i / 2 + "个！";// 1000次循环，生成0100个string类型的对象
}
else {
key = "wzk收到了苹果:1个!目前拥有1个！";
}
/*
* hashCode方法是Object的方法，可以将任何类型的数据
* 计算映射到一个整数区间；
* hashCode&Integer的最大数，位的与运算，保真运算，31位2进制的值
* 将负的值，转化成了正数
* 对n取余，n取值是根据数据分片个数决定的，3
*/
System.out.println(key+"key.hashCode():"+key.hashCode()+"(key.hashCode() & Integer.MAX_VALUE):"+(key.hashCode() & Integer.MAX_VALUE)+"分片:"+(key.hashCode() & Integer.MAX_VALUE) % 3);
}
}没有办法预测获取key值的取值结果范围时，可以利用hash取余计算，将任何key的值映射到一个循环的整数区间内[0,n-1]
整合到redis
/*
* 5利用hash取余的公式，完成数据分片的计算逻辑
*/
@Test
public void test05(){
//准备jedis链接6379，6380，6381的客户端
Jedis jedis1=new Jedis("192.168.65.128", 6379);
Jedis jedis2=new Jedis("192.168.65.128", 6380);
Jedis jedis3=new Jedis("192.168.65.128", 6381);
for(int i=1;i<=1000;i++){
String value = "";
if (i > 1)
{
value = "wzk又收到了苹果:" + i + "个!目前拥有" + (1 + i) * i / 2 + "个！";// 100次循环，生成100个string类型的对象
}
else {
value = "wzk收到了苹果:1个!目前拥有1个！";
}
String key=value;
//获取取余结果
int result=(key.hashCode()&Integer.MAX_VALUE)%3;
if(result==0){//存储到6379
jedis1.set(key, value);
}
if(result==1){//6380
jedis2.set(key, value);
}
if(result==2){//6381
jedis3.set(key, value);
}
}
for(int i=1;i<=1000;i++){
String value = "";
if (i > 1)
{
value = "wzk又收到了苹果:" + i + "个!目前拥有" + (1 + i) * i / 2 + "个！";// 1000次循环，生成1000个string类型的对象
}
else {
value = "wzk收到了苹果:1个!目前拥有1个！";
}
String key=value;
//获取取余结果
int result=(key.hashCode()&Integer.MAX_VALUE)%3;
if(result==0){//存储到6379
System.out.println(jedis1.get(key));
}
if(result==1){//6380
System.out.println(jedis2.get(key));
}
if(result==2){//6381
System.out.println(jedis3.get(key));
}
}
jedis1.flushAll();
jedis2.flushAll();
jedis3.flushAll();
jedis1.close();
jedis2.close();
jedis3.close();
}
jedis的数据分片
利用hash取余完成了数据分片的计算，jedis的底层有自己的计算逻辑（hash一致性）
/*
* jedis底层的数据分片计算逻辑是使用的hash一致性
*/
@Test
public void test06(){
//不能使用单独的链接对象jedis来操作集群，需要创建分片的对象
//收集节点信息，告诉jedis集群的所有节点都是谁
List<JedisShardInfo> infoList=new ArrayList<JedisShardInfo>();
//封装3个节点的链接信息
JedisShardInfo info1=new
JedisShardInfo("192.168.65.128", 6379);
JedisShardInfo info2=new
JedisShardInfo("192.168.65.128", 6380);
JedisShardInfo info3=new
JedisShardInfo("192.168.65.128", 6381);
infoList.add(info1);
infoList.add(info2);
infoList.add(info3);
//利用收集到的信息，创建一个分片对象，分片对象的操作和jedis对象一值
//但是在存储，读取数据时，已经对key值做了数据分片的计算，从而可以正确的到
//某个节点获取，存储数据
ShardedJedis sJedis=new ShardedJedis(infoList);
//sJedis.set("name", "haha");
for(int i=1;i<=1000;i++){
String value = "";
if (i > 1)
{
value = "wzk又收到了苹果:" + i + "个!目前拥有" + (1 + i) * i / 2 + "个！";// 1000次循环，生成1000个string类型的对象
}
else {
value = "wzk收到了苹果:1个!目前拥有1个！";
}
String key=value;
sJedis.set(key, value);
}
for(int i=1;i<=1000;i++){
String value = "";
if (i > 1)
{
value = "wzk又收到了苹果:" + i + "个!目前拥有" + (1 + i) * i / 2 + "个！";// 1000次循环，生成1000个string类型的对象
}
else {
value = "wzk收到了苹果:1个!目前拥有1个！";
}
String key=value;
System.out.println(sJedis.get(key));
}
sJedis.close();
}分片的连接池
/*
* 7 利用jedis完成分片连接池的配置，每次调用是从池中获取链接，返回资源
*/
@SuppressWarnings("deprecation")
@Test
public void test07(){
//链接池中的每个链接对象，都需要链接3个节点的集群
//收集节点信息，告诉jedis集群的所有节点都是谁
List<JedisShardInfo> infoList=new ArrayList<JedisShardInfo>();
//封装3个节点的链接信息
JedisShardInfo info1=new
JedisShardInfo("192.168.65.128", 6379);
JedisShardInfo info2=new
JedisShardInfo("192.168.65.128", 6380);
JedisShardInfo info3=new
JedisShardInfo("192.168.65.128", 6381);
infoList.add(info1);
infoList.add(info2);
infoList.add(info3);
//连接池，具有自己的一些配置信息，最大连接数，最大空闲链接
//准备一个配置对象
JedisPoolConfig config=new JedisPoolConfig();
//设置一些参数
config.setMaxIdle(8);
config.setMaxTotal(200);//最大连接数
//创建分片连接池对象
ShardedJedisPool pool=new
ShardedJedisPool(config, infoList);
//获取链接对象
ShardedJedis sJedis = pool.getResource();
for(int i=1;i<=1000;i++){
String value = "";
if (i > 1)
{
value = "wzk又收到了苹果:" + i + "个!目前拥有" + (1 + i) * i / 2 + "个！";// 1000次循环，生成1000个string类型的对象
}
else {
value = "wzk收到了苹果:1个!目前拥有1个！";
}
String key=value;
System.out.println(sJedis.get(key));
}
pool.returnResource(sJedis);
}下一篇会介绍一下hash一致性。。。