WRONGTYPE Operation against a key holding the wrong kind of value

redis报错:

2022/07/20 18:12:32.277 ERROR [0a2cd65e1bef3f0c][task-304318095] o.s.a.i.SimpleAsyncUncaughtExceptionHandler : Unexpected exception occurred invoking async method: public void com.test.app.system.bus.listener.DeviceUpwardListener.onApplicationEvent(com.test.app.system.bus.e.DeviceEvent)
redis.clients.jedis.exceptions.JedisDataException: WRONGTYPE Operation against a key holding the wrong kind of value
at redis.clients.jedis.Protocol.processError(Protocol.java:132) ~[jedis-3.1.0.jar!/:na]
at redis.clients.jedis.Protocol.process(Protocol.java:166) ~[jedis-3.1.0.jar!/:na]
at redis.clients.jedis.Protocol.read(Protocol.java:220) ~[jedis-3.1.0.jar!/:na]
at redis.clients.jedis.Connection.readProtocolWithCheckingBroken(Connection.java:318) ~[jedis-3.1.0.jar!/:na]
at redis.clients.jedis.Connection.getIntegerReply(Connection.java:260) ~[jedis-3.1.0.jar!/:na]
at redis.clients.jedis.BinaryJedis.hexists(BinaryJedis.java:1034) ~[jedis-3.1.0.jar!/:na]

 

原因是写入redis的key存在两个(多个),但是类型不一样;

如:

jedis.set(“test”,1);

jedis.hset(“test”,”key”1);

此时testkey就存在两种类型且两个,则有可能会报这个错误。

解决方法就是删除重复的键值,建议每一种set方法都加入一个redis方法名作为前缀,这样子就不可能会出现重复的键。

如:

jedis.set(“set:test”,1);

jedis.hset(“hset:test”,”key”1);

分布式锁的三种实现方式

分布式锁三种实现方式:

  1. 基于数据库实现分布式锁;
  2. 基于缓存(Redis等)实现分布式锁;
  3. 基于Zookeeper实现分布式锁;

一, 基于数据库实现分布式锁

1. 悲观锁

利用select … where … for update 排他锁

*其他附加功能与实现一基本一致,这里需要注意的是“where name=lock ”,name字段必须要走索引,否则会锁表。有些情况下,比如表不大,mysql优化器会不走这个索引,导致锁表问题。

2. 乐观锁

所谓乐观锁与前边最大区别在于基于CAS思想,是不具有互斥性,不会产生锁等待而消耗资源,操作过程中认为不存在并发冲突,只有update version失败后才能觉察到。

我们的抢购、秒杀就是用了这种实现以防止超卖。通过增加递增的版本号字段实现乐观锁

1637136916-5676-e3b761944edd6611defde7ee28-0

二, 基于缓存(Redis等)实现分布式锁

1. 使用命令介绍:

(1)SETNX
SETNX key val:当且仅当key不存在时,set一个key为val的字符串,返回1;若key存在,则什么都不做,返回0。
(2)expire
expire key timeout:为key设置一个超时时间,单位为second,超过这个时间锁会自动释放,避免死锁。
(3)delete
delete key:删除key

在使用Redis实现分布式锁的时候,主要就会使用到这三个命令。

2. 实现思想:

(1)获取锁的时候,使用setnx加锁,并使用expire命令为锁添加一个超时时间,超过该时间则自动释放锁,锁的value值为一个随机生成的UUID,通过此在释放锁的时候进行判断。
(2)获取锁的时候还设置一个获取的超时时间,若超过这个时间则放弃获取锁。
(3)释放锁的时候,通过UUID判断是不是该锁,若是该锁,则执行delete进行锁释放。

3. 分布式锁的简单实现代码:

package com.demo.app;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.exceptions.JedisException;

import java.util.List;
import java.util.UUID;

/**
 * 分布式锁的简单实现代码
 */
class DistributedLock {

    private final JedisPool jedisPool;

    public DistributedLock(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
    }

    /**
     * 加锁
     *
     * @param lockName       锁的key
     * @param acquireTimeout 获取超时时间
     * @param timeout        锁的超时时间
     * @return 锁标识
     */

    public String lockWithTimeout(String lockName, long acquireTimeout, long timeout) {

        Jedis conn = null;
        String retIdentifier = null;
        try {
            // 获取连接
            conn = jedisPool.getResource();
            // 随机生成一个value
            String identifier = UUID.randomUUID().toString();
            // 锁名,即key值
            String lockKey = "lock:" + lockName;
            // 超时时间,上锁后超过此时间则自动释放锁
            int lockExpire = (int) (timeout / 1000);
            // 获取锁的超时时间,超过这个时间则放弃获取锁
            long end = System.currentTimeMillis() + acquireTimeout;

            while (System.currentTimeMillis() < end) {
                if (conn.setnx(lockKey, identifier) == 1) {
                    conn.expire(lockKey, lockExpire);
                    // 返回value值,用于释放锁时间确认
                    retIdentifier = identifier;
                    return retIdentifier;
                }

                // 返回-1代表key没有设置超时时间,为key设置一个超时时间
                if (conn.ttl(lockKey) == -1) {
                    conn.expire(lockKey, lockExpire);
                }
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        } catch (JedisException e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                conn.close();
            }
        }
        return retIdentifier;
    }

    /**
     * 释放锁
     *
     * @param lockName   锁的key
     * @param identifier 释放锁的标识
     * @return
     */
    public boolean releaseLock(String lockName, String identifier) {
        Jedis conn = null;
        String lockKey = "lock:" + lockName;
        boolean retFlag = false;

        try {
            conn = jedisPool.getResource();
            while (true) {
                // 监视lock,准备开始事务
                conn.watch(lockKey);
                // 通过前面返回的value值判断是不是该锁,若是该锁,则删除,释放锁
                if (identifier.equals(conn.get(lockKey))) {
                    Transaction transaction = conn.multi();
                    transaction.del(lockKey);
                    List<Object> results = transaction.exec();
                    if (results == null) {
                        continue;
                    }
                    retFlag = true;
                }
                conn.unwatch();
                break;
            }
        } catch (JedisException e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                conn.close();
            }
        }
        return retFlag;
    }
}

4. 测试刚才实现的分布式锁

例子中使用50个线程模拟秒杀一个商品,使用–运算符来实现商品减少,从结果有序性就可以看出是否为加锁状态。
模拟秒杀服务,在其中配置了jedis线程池,在初始化的时候传给分布式锁,供其使用。

public class Service {
    private static JedisPool pool = null;
    private DistributedLock lock = new DistributedLock(pool);
    int n = 500;
    
    static {
        JedisPoolConfig config = new JedisPoolConfig();
        // 设置最大连接数
        config.setMaxTotal(200);
        // 设置最大空闲数
        config.setMaxIdle(8);
        // 设置最大等待时间
        config.setMaxWaitMillis(1000 * 100);
        // 在borrow一个jedis实例时,是否需要验证,若为true,则所有jedis实例均是可用的
        config.setTestOnBorrow(true);
        pool = new JedisPool(config, "127.0.0.1", 6379, 3000);

    }

    public void secKill() {
        // 返回锁的value值,供释放锁时候进行判断
        String identifier = lock.lockWithTimeout("resource", 5000, 1000);
        System.out.println(Thread.currentThread().getName() + "获得了锁");
        System.out.println(--n);
        lock.releaseLock("resource", identifier);

    }
}

模拟线程进行秒杀服务;

public class ThreadA extends Thread {
    private Service service;
    public ThreadA(Service service) {
        this.service = service;
    }
    
    @Override
    public void run() {
        service.secKill();
    }
}

public class Test {
    public static void main(String[] args) {
        Service service = new Service();
        for (int i = 0; i < 50; i++) {
            ThreadA threadA = new ThreadA(service);
            threadA.start();
        }
    }
}

结果如下,结果为有序的:

1637136917-5764-ceea4c8cbfa184228a02822f99-1

若注释掉使用锁的部分:

public void secKill() {
    // 返回锁的value值,供释放锁时候进行判断
    //String indentifier = lock.lockWithTimeout("resource", 5000, 1000);
    System.out.println(Thread.currentThread().getName() + "获得了锁");
    System.out.println(--n);
    //lock.releaseLock("resource", indentifier);
}

从结果可以看出,有一些是异步进行的:

1637136917-6356-ceea4c8cbfa184228a02822f99-2

三, 基于Zookeeper实现分布式锁

ZooKeeper是一个为分布式应用提供一致性服务的开源组件,它内部是一个分层的文件系统目录树结构,规定同一个目录下只能有一个唯一文件名。基于ZooKeeper实现分布式锁的步骤如下:

  • (1)创建一个目录mylock;
  • (2)线程A想获取锁就在mylock目录下创建临时顺序节点;
  • (3)获取mylock目录下所有的子节点,然后获取比自己小的兄弟节点,如果不存在,则说明当前线程顺序号最小,获得锁;
  • (4)线程B获取所有节点,判断自己不是最小节点,设置监听比自己次小的节点;
  • (5)线程A处理完,删除自己的节点,线程B监听到变更事件,判断自己是不是最小的节点,如果是则获得锁。

这里推荐一个Apache的开源库Curator,它是一个ZooKeeper客户端,Curator提供的InterProcessMutex是分布式锁的实现,acquire方法用于获取锁,release方法用于释放锁。

实现源码如下:

public class ZkLock implements DistributionLock {

    private String zkAddress = "zk_adress";
    private static final String root = "package root";
    private CuratorFramework zkClient;
    private final String LOCK_PREFIX = "/lock_";
    
    @Bean
    public DistributionLock initZkLock() {
        if (StringUtils.isBlank(root)) {
            throw new RuntimeException("zookeeper 'root' can't be null");
        }
        zkClient = CuratorFrameworkFactory
                .builder()
                .connectString(zkAddress)
                .retryPolicy(new RetryNTimes(2000, 20000))
                .namespace(root)
                .build();
        zkClient.start();
        return this;
    }

    public boolean tryLock(String lockName) {
        lockName = LOCK_PREFIX + lockName;
        boolean locked = true;

        try {
            Stat stat = zkClient.checkExists().forPath(lockName);
            if (stat == null) {
                log.info("tryLock:{}", lockName);
                stat = zkClient.checkExists().forPath(lockName);
                if (stat == null) {
                    zkClient
                            .create()
                            .creatingParentsIfNeeded()
                            .withMode(CreateMode.EPHEMERAL)
                            .forPath(lockName, "1".getBytes());
                } else {
                    log.warn("double-check stat.version:{}", stat.getAversion());
                    locked = false;
                }
            } else {
                log.warn("check stat.version:{}", stat.getAversion());
                locked = false;
            }

        } catch (Exception e) {
            locked = false;
        }
        return locked;
    }

    public void release(String lockName) {
        lockName = LOCK_PREFIX + lockName;
        try {
            zkClient
                    .delete()
                    .guaranteed()
                    .deletingChildrenIfNeeded()
                    .forPath(lockName);
            log.info("release:{}", lockName);
        } catch (Exception e) {
            log.error("删除", e);
        }

    }

    public void setZkAddress(String zkAddress) {
        this.zkAddress = zkAddress;
    }

}

优点:具备高可用、可重入、阻塞锁特性,可解决失效死锁问题。

缺点:因为需要频繁的创建和删除节点,性能上不如Redis方式。

四,对比

数据库分布式锁实现

缺点:

1.db操作性能较差,并且有锁表的风险

2.非阻塞操作失败后,需要轮询,占用cpu资源;

3.长时间不commit或者长时间轮询,可能会占用较多连接资源

Redis(缓存)分布式锁实现

缺点:

1.锁删除失败 过期时间不好控制

2.非阻塞,操作失败后,需要轮询,占用cpu资源;

ZK分布式锁实现

缺点:性能不如redis实现,主要原因是写操作(获取锁释放锁)都需要在Leader上执行,然后同步到follower。

ZooKeeper有较好的性能和可靠性。

从理解的难易程度角度(从低到高)数据库 > 缓存 > Zookeeper

从实现的复杂性角度(从低到高)Zookeeper >= 缓存 > 数据库

从性能角度(从高到低)缓存 > Zookeeper >= 数据库

从可靠性角度(从高到低)Zookeeper > 缓存 > 数据库

MISCONF Redis is configured to save RDB snapshots, but is currently not able to persist on disk. Com

之前没有遇到过MISCONF Redis is configured to save RDB snapshots, but is currently not able to persist on disk. 错误信息,但是通过信息应该是磁盘的问题。

确认磁盘空间充足、内存也充足的情况下,网上找了一下解决方案:

有建议设置参数“config set stop-writes-on-bgsave-error no”。但这样子只是暂时性的,问题依旧没有解决。

Linux内核参数之 overcommit_memory

/etc/sysctl.conf
vm.overcommit_memory=1
或者
sysctl vm.overcommit_memory=1
或者
echo 1 > /proc/sys/vm/overcommit_memory

内核参数说明如下:

overcommit_memory文件指定了内核针对内存分配的策略,其值可以是0、1、2。

  •   0, 表示内核将检查是否有足够的可用内存供应用进程使用;如果有足够的可用内存,内存申请允许;否则,内存申请失败,并把错误返回给应用进程。
  • 1, 表示内核允许分配所有的物理内存,而不管当前的内存状态如何。
  • 2, 表示内核允许分配超过所有物理内存和交换空间总和的内存

 

为什么系统明明还剩2GB的内存,Redis会说内存不够呢?

这里有一个帖子 ,分析很到位:http://www.linuxidc.com/Linux/2012-07/66079.htm

 

碰到一个悲催的事情:一台Redis服务器,4核,16G内存且没有任何硬件上的问题。持续高压运行了大约3个月,保存了大约14G的数据,设置了比较完备的Save参数。而就是这台主机,在一次重起之后,丢失了大量的数据,14G的数据最终只恢复了几百兆而已。

正常情况下,像Redis这样定期回写磁盘的内存数据库,丢失几个数据也是在情理之中,可超过80%数据丢失率实在太离谱。排除了误操作的可能性之后,开始寻找原因。

重启动时的日志:

[26641] 21 Dec 09:46:34 * Slave ask for synchronization

[26641] 21 Dec 09:46:34 * Starting BGSAVE for SYNC

[26641] 21 Dec 09:46:34 # Can’t save in background: fork: Cannot allocate memory

[26641] 21 Dec 09:46:34 * Replication failed, can’t BGSAVE

[26641] 21 Dec 09:46:34 # Received SIGTERM, scheduling shutdown…

[26641] 21 Dec 09:46:34 # User requested shutdown…

很明显的一个问题,系统不能在后台保存,fork进程失败。

翻查了几个月的日志,发觉系统在频繁报错:

[26641] 18 Dec 04:02:14 * 1 changes in 900 seconds. Saving…

[26641] 18 Dec 04:02:14 # Can’t save in background: fork: Cannot allocate memory

系统不能在后台保存,fork进程时无法指定内存。

对源码进行跟踪,在src/rdb.c中定位了这个报错:

int rdbSaveBackground(char *filename) {
    pid_t childpid;
    long long start;

    if (server.bgsavechildpid != -1) return REDIS_ERR;
    if (server.vm_enabled) waitEmptyIOJobsQueue();
    server.dirty_before_bgsave = server.dirty;
    start = ustime();
    if ((childpid = fork()) == 0) {
        /* Child */
        if (server.vm_enabled) vmReopenSwapFile();
        if (server.ipfd > 0) close(server.ipfd);
        if (server.sofd > 0) close(server.sofd);
        if (rdbSave(filename) == REDIS_OK) {
            _exit(0);
        } else {
            _exit(1);
        }
    } else {
        /* Parent */
        server.stat_fork_time = ustime()-start;
        if (childpid == -1) {
            redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
                strerror(errno));
            return REDIS_ERR;
        }
        redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
        server.bgsavechildpid = childpid;
        updateDictResizePolicy();
        return REDIS_OK;
    }
    return REDIS_OK; /* unreached */
}

数据丢失的问题总算搞清楚了!

Redis的数据回写机制分同步和异步两种,

  1. 同步回写即SAVE命令,主进程直接向磁盘回写数据。在数据大的情况下会导致系统假死很长时间,所以一般不是推荐的。
  2. 异步回写即BGSAVE命令,主进程fork后,复制自身并通过这个新的进程回写磁盘,回写结束后新进程自行关闭。由于这样做不需要主进程阻塞,系统不会假死,一般默认会采用这个方法。

个人感觉方法2采用fork主进程的方式很拙劣,但似乎是唯一的方法。内存中的热数据随时可能修改,要在磁盘上保存某个时间的内存镜像必须要冻结。冻结就会导致假死。fork一个新的进程之后等于复制了当时的一个内存镜像,这样主进程上就不需要冻结,只要子进程上操作就可以了。

在小内存的进程上做一个fork,不需要太多资源,但当这个进程的内存空间以G为单位时,fork就成为一件很恐怖的操作。何况在16G内存的主机上fork 14G内存的进程呢?肯定会报内存无法分配的。更可气的是,越是改动频繁的主机上fork也越频繁,fork操作本身的代价恐怕也不会比假死好多少。

找到原因之后,直接修改内核参数vm.overcommit_memory = 1

Linux内核会根据参数vm.overcommit_memory参数的设置决定是否放行。

  1.  如果 vm.overcommit_memory = 1,直接放行
  2. vm.overcommit_memory = 0:则比较 此次请求分配的虚拟内存大小和系统当前空闲的物理内存加上swap,决定是否放行。
  3. vm.overcommit_memory = 2:则会比较 进程所有已分配的虚拟内存加上此次请求分配的虚拟内存和系统当前的空闲物理内存加上swap,决定是否放行。

 

Redis Sentinel

一、哨兵模式概述

哨兵模式是一种特殊的模式,首先Redis提供了哨兵的命令,哨兵是一个独立的进程,作为进程,它会独立运行。其原理是哨兵通过发送命令,等待Redis服务器响应,从而监控运行的多个Redis实例。

001
Redis哨兵

这里的哨兵有两个作用

  • 通过发送命令,让Redis服务器返回监控其运行状态,包括主服务器和从服务器。
  • 当哨兵监测到master宕机,会自动将slave切换成master,然后通过发布订阅模式通知其他的从服务器,修改配置文件,让它们切换主机。

然而一个哨兵进程对Redis服务器进行监控,可能会出现问题,为此,我们可以使用多个哨兵进行监控。各个哨兵之间还会进行监控,这样就形成了多哨兵模式。

用文字描述一下故障切换(failover)的过程。假设主服务器宕机,哨兵1先检测到这个结果,系统并不会马上进行failover过程,仅仅是哨兵1主观的认为主服务器不可用,这个现象成为主观下线。当后面的哨兵也检测到主服务器不可用,并且数量达到一定值时,那么哨兵之间就会进行一次投票,投票的结果由一个哨兵发起,进行failover操作。切换成功后,就会通过发布订阅模式,让各个哨兵把自己监控的从服务器实现切换主机,这个过程称为客观下线。这样对于客户端而言,一切都是透明的。

二、Redis配置哨兵模式

配置3个哨兵和1主2从的Redis服务器来演示这个过程。

服务类型 是否是主服务器 IP地址 端口
Redis 192.168.11.128 6379
Redis 192.168.11.129 6379
Redis 192.168.11.130 6379
Sentinel 192.168.11.128 26379
Sentinel 192.168.11.129 26379
Sentinel 192.168.11.130 26379
002
多哨兵监控Redis

首先配置Redis的主从服务器,修改redis.conf文件如下

# 使得Redis服务器可以跨网络访问
bind 0.0.0.0
# 设置密码
requirepass "123456"
# 指定主服务器,注意:有关slaveof的配置只是配置从服务器,主服务器不需要配置
slaveof 192.168.11.128 6379
# 主服务器密码,注意:有关slaveof的配置只是配置从服务器,主服务器不需要配置
masterauth 123456

上述内容主要是配置Redis服务器,从服务器比主服务器多一个slaveof的配置和密码。

配置3个哨兵,每个哨兵的配置都是一样的。在Redis安装目录下有一个sentinel.conf文件,copy一份进行修改

# 禁止保护模式
protected-mode no
# 配置监听的主服务器,这里sentinel monitor代表监控,mymaster代表服务器的名称,可以自定义,192.168.11.128代表监控的主服务器,6379代表端口,2代表只有两个或两个以上的哨兵认为主服务器不可用的时候,才会进行failover操作。
sentinel monitor mymaster 192.168.11.128 6379 2
# sentinel author-pass定义服务的密码,mymaster是服务名称,123456是Redis服务器密码
# sentinel auth-pass <master-name> <password>
sentinel auth-pass mymaster 123456

上述关闭了保护模式,便于测试。

有了上述的修改,我们可以进入Redis的安装目录的src目录,通过下面的命令启动服务器和哨兵


# 启动Redis服务器进程
./redis-server ../redis.conf
# 启动哨兵进程
./redis-sentinel ../sentinel.conf

注意启动的顺序。首先是主机(192.168.11.128)的Redis服务进程,然后启动从机的服务进程,最后启动3个哨兵的服务进程。

三、Java中使用哨兵模式

/**
 * 测试Redis哨兵模式
 * @author liu
 */
public class TestSentinels {
    @SuppressWarnings("resource")
    @Test
    public void testSentinel() {
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(10);
        jedisPoolConfig.setMaxIdle(5);
        jedisPoolConfig.setMinIdle(5);
        // 哨兵信息
        Set<String> sentinels = new HashSet<>(Arrays.asList("192.168.11.128:26379",
                "192.168.11.129:26379","192.168.11.130:26379"));
        // 创建连接池
        JedisSentinelPool pool = new JedisSentinelPool("mymaster", sentinels,jedisPoolConfig,"123456");
        // 获取客户端
        Jedis jedis = pool.getResource();
        // 执行两个命令
        jedis.set("mykey", "myvalue");
        String value = jedis.get("mykey");
        System.out.println(value);
    }
}

上面是通过Jedis进行使用的,同样也可以使用Spring进行配置RedisTemplate使用。

        <bean id = "poolConfig" class="redis.clients.jedis.JedisPoolConfig">
            <!-- 最大空闲数 -->
            <property name="maxIdle" value="50"></property>
            <!-- 最大连接数 -->
            <property name="maxTotal" value="100"></property>
            <!-- 最大等待时间 -->
            <property name="maxWaitMillis" value="20000"></property>
        </bean>
        
        <bean id="connectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
            <constructor-arg name="poolConfig" ref="poolConfig"></constructor-arg>
            <constructor-arg name="sentinelConfig" ref="sentinelConfig"></constructor-arg>
            <property name="password" value="123456"></property>
        </bean>
        
        <!-- JDK序列化器 -->
        <bean id="jdkSerializationRedisSerializer" class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"></bean>
        
        <!-- String序列化器 -->
        <bean id="stringRedisSerializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"></bean>
        
        <bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
            <property name="connectionFactory" ref="connectionFactory"></property>
            <property name="keySerializer" ref="stringRedisSerializer"></property>
            <property name="defaultSerializer" ref="stringRedisSerializer"></property>
            <property name="valueSerializer" ref="jdkSerializationRedisSerializer"></property>
        </bean>
        
        <!-- 哨兵配置 -->
        <bean id="sentinelConfig" class="org.springframework.data.redis.connection.RedisSentinelConfiguration">
            <!-- 服务名称 -->
            <property name="master">
                <bean class="org.springframework.data.redis.connection.RedisNode">
                    <property name="name" value="mymaster"></property>
                </bean>
            </property>
            <!-- 哨兵服务IP和端口 -->
            <property name="sentinels">
                <set>
                    <bean class="org.springframework.data.redis.connection.RedisNode">
                        <constructor-arg name="host" value="192.168.11.128"></constructor-arg>
                        <constructor-arg name="port" value="26379"></constructor-arg>
                    </bean>
                    <bean class="org.springframework.data.redis.connection.RedisNode">
                        <constructor-arg name="host" value="192.168.11.129"></constructor-arg>
                        <constructor-arg name="port" value="26379"></constructor-arg>
                    </bean>
                    <bean class="org.springframework.data.redis.connection.RedisNode">
                        <constructor-arg name="host" value="192.168.11.130"></constructor-arg>
                        <constructor-arg name="port" value="26379"></constructor-arg>
                    </bean>
                </set>
            </property>
        </bean>

四、哨兵模式的其他配置项

配置项 参数类型 作用
port 整数 启动哨兵进程端口
dir 文件夹目录 哨兵进程服务临时文件夹,默认为/tmp,要保证有可写入的权限
sentinel down-after-milliseconds <服务名称><毫秒数(整数)> 指定哨兵在监控Redis服务时,当Redis服务在一个默认毫秒数内都无法回答时,单个哨兵认为的主观下线时间,默认为30000(30秒)
sentinel parallel-syncs <服务名称><服务器数(整数)> 指定可以有多少个Redis服务同步新的主机,一般而言,这个数字越小同步时间越长,而越大,则对网络资源要求越高
sentinel failover-timeout <服务名称><毫秒数(整数)> 指定故障切换允许的毫秒数,超过这个时间,就认为故障切换失败,默认为3分钟
sentinel notification-script <服务名称><脚本路径> 指定sentinel检测到该监控的redis实例指向的实例异常时,调用的报警脚本。该配置项可选,比较常用

sentinel down-after-milliseconds配置项只是一个哨兵在超过规定时间依旧没有得到响应后,会自己认为主机不可用。对于其他哨兵而言,并不是这样认为。哨兵会记录这个消息,当拥有认为主观下线的哨兵达到sentinel monitor所配置的数量时,就会发起一次投票,进行failover,此时哨兵会重写Redis的哨兵配置文件,以适应新场景的需要。

 

Redis内存溢出的问题

开发的电脑内存增加到24G,启动Redis失败,报:

L480@luo-zip MINGW64 /d/develop/tools/redis64-2.8.12
$ ./redis-server.exe
[13308] 12 Dec 15:13:41.994 #
The Windows version of Redis allocates a memory mapped heap for sharing with
the forked process used for persistence operations. In order to share this
memory, Windows allocates from the system paging file a portion equal to the
size of the Redis heap. At this time there is insufficient contiguous free
space available in the system paging file for this operation (Windows error
0x5AF). To work around this you may either increase the size of the system
paging file, or decrease the size of the Redis heap with the –maxheap flag.
Sometimes a reboot will defragment the system paging file sufficiently for
this operation to complete successfully.

Please see the documentation included with the binary distributions for more
details on the –maxheap flag.

Redis can not continue. Exiting.

解决方法就是启动的时候添加–maxheap参数,如:

./redis-server.exe –maxheap 10240000

 

 

 
Copyright © 2008-2021 lanxinbase.com Rights Reserved. | 粤ICP备14086738号-3 |