jedis订阅发布故障处理

Posted 2023-03-15

参考技术A redis订阅主题数过一段时间就减少，导致订阅的进程无法收到生产者publish的消息。

启动订阅进程，启动发布进程。分别断开与redis之前的网络连接，发现jedis断开后又重连连上了，通过redis命令pubsub命令查询订阅数，发现订阅数减少了。由此证明当网络中断后订阅丢失。然而jedis.subscribe并不会异常退出，而是一直卡在这个方法中，故没法根据异常退出方法重新订阅。

redis发布订阅还是不太适合商用，并不支持消息重试，即便重新注册上了，之前的消息也丢了。

Redis的客户端Jedis

1、 Redis支持消息的订阅与发布

Redis的消息订阅支持：先订阅后发布

订阅：subscribe c1 c2

发布：publish c2 hello-redis

支持通配符的订阅：psubscribe new*

实际开发中没用，RabbitMQ

2、Redis对于事务支持

Redis对于事务也是支持，但是支持的不太好。

multi:开启事务

exec:提交事务

discard:回滚事务

3、Redis的Java客户端Jedis

3.1先导入依赖包

 

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>2.9.0</version>
</dependency>

 

3.2，测试代码

 public static void main( String[] args ){
        //1.创建Jedis客户端，并指定Redis服务器的ip和端口号
        Jedis jedis = new Jedis("192.168.58.10",6379);
        //2.通过Jedis客户端操作Redis服务器
        jedis.set("key1","value1");
        String  value1 = jedis.get("key1");
        System.out.println(value1);
        //3.关闭Jedis客户端
        jedis.close();
    }
这种还需要一个个创建！！不太好 

3.3，创建jedisPool

这种每用一次都需要创建一个连接池  消耗内存  

public static void main( String[] args ){
        //1.创建JedisPool对象，从JedisPool对象中获取Jedis连接
        JedisPool jedisPool = new JedisPool("192.168.58.10",6379);
        Jedis jedis = jedisPool.getResource();
        //2.通过Jedis客户端操作Redis服务器
        String  value1 = jedis.get("key1");
        System.out.println(value1);
        //3.关闭Jedis客户端
        jedis.close();
    }

 

四、Jedis和Spring的整合

4.1导入spring 和jedis的包

4.2 :配置配置文件  ：这种就不用每次用 就不用创建jedis池了 只创建一次 就行。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

    <bean id="jedisPoolConfig" class="redis.clients.jedis.JedisPoolConfig">
        <!-- 最大空闲连接数 -->
        <property name="maxIdle" value="1" />
        <!-- 最大连接数 -->
        <property name="maxTotal" value="5" />
        <!-- 连接耗尽时是否阻塞, false报异常,ture阻塞直到超时, 默认true -->
        <property name="blockWhenExhausted" value="true" />
        <!-- 获取连接时的最大等待毫秒数(如果设置为阻塞时BlockWhenExhausted),如果超时就抛异常, 小于零:阻塞不确定的时间,  默认-1 -->
        <property name="maxWaitMillis" value="30000" />
        <!-- 在获取连接的时候检查有效性 -->
        <property name="testOnBorrow" value="true" />
    </bean>

    <bean class="redis.clients.jedis.JedisPool" destroy-method="close">
        <constructor-arg name="host" value="192.168.248.99"/>
        <constructor-arg name="port" value="6379"/>
        <constructor-arg name="poolConfig" ref="jedisPoolConfig"/>
    </bean>

</beans>

五、Jedis和SSM整合

5.1:导包

5.2：配置ssm和jedis配置文件的整合

5.2:然后redis缓存只能应用在服务层。

第一种

这种是利用自定义的工具类 是使用字符串转化成数组 ，数组幻化成字符串的方式进行缓存。

先创建自定义序列化，反序列化工具类

 

package com.ujy.maven.utils;
import java.io.*;
public class SerizlizableUtil {

    //完成序列化
    public  static byte[] serializable(Object object){
      /*
      * 接收一个对象，传到writeObject里面，通过对象输出流写入到字节输出流
      * ，然后从字节数组输出流的toByteArray获取这个数组，然后返回给调用者
      * */
       try {
           //1.字节数组输出流
           ByteArrayOutputStream baos = new ByteArrayOutputStream(); //3
           //2.对象输出流
           ObjectOutputStream oos = new ObjectOutputStream(baos);//2
           oos.writeObject(object);  //1
           byte[] bytes = baos.toByteArray();//4
           return  bytes;//5
       }catch (Exception e){
           e.printStackTrace();
       }
       return  null;
    }
    //完成反序列化
    public static Object unSerializable(byte [] bytes){
   //先判断是否非空 非空直接返回
      if (bytes==null){
          return  null;
      }
        try {
        //1.字节数组输入流
        ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
        //2.对象输入流
         ObjectInputStream ois = new ObjectInputStream(bais);
            Object o = ois.readObject();
            return   o;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return  null;
    }
}

 

然后在服务层应用：

 @Autowired
    private PersonDao personDao;
    @Autowired
    private JedisPool jedisPool;

 public List<Person> findAll() {
        Jedis jedis = jedisPool.getResource();
        List<Person> list=null;

        long startTime = System.currentTimeMillis();
        System.out.println(startTime);
        //1.先去缓存中去查数据 ，
        byte[] lists = jedis.get(SerizlizableUtil.serializable("personList"));
        //2. 如果没有就从数据库查 然后存入缓存 下次可以从缓存取。
        if (lists==null){
             list = personDao.findAll();
            long endTime1 = System.currentTimeMillis();
            System.out.println("去数据库中查询数据用的时间为:"+(endTime1-startTime));
            jedis.set(SerizlizableUtil.serializable("personList") ,SerizlizableUtil.serializable(list));
         }else {
            //3.如果缓存中有数据，直接反序列化，返回给控制层
             list=(List<Person>) SerizlizableUtil.unSerializable(lists);
            System.out.println(list);
            long endTime2 = System.currentTimeMillis();
            System.out.println("去缓存中查询数据用的时间为:"+(endTime2-startTime));
        }
          jedis.close();
         return  list;
    }

 

 

第二种：

这种是利用FastJson这种定义好的工具类 ，是转换成字符串，然后字符串转换成对象或者集合或者map。

先来两个测试代码：这种是任意类型转换成String类型。

 

  @Test
    public void test1(){
        //对象--->字符串
      Person person = new Person(6,"王子",1,22);
        String s = JSON.toJSONString(person);
        System.out.println(s);

        //对象列表(集合)--->字符串
        List<Person>  list = new ArrayList<Person>();
        list.add(new Person(7,"格格",0,20));
        list.add(new Person(8 ,"公主",0,21));
        list.add(new Person(9,"逍遥",1,23));
        String s1 = JSON.toJSONString(list);
        System.out.println(s1);

        //Map--->字符串
        Map<String,Object> map = new HashMap<String, Object>();
        map.put("小李飞刀","刀无虚发");
        map.put("飞檐走壁","乃真人也");
        String s = JSON.toJSONString(map);
        System.out.println(s);

    }

 

这种是String转换成 对象，List ，数组，Map

无论任意数据类型转换成字符串就通过：JSON.toJSONString()
     字符串转换成其他数据类型：
          String --->对象类型JSON.parseObj()
           String--->数组类型JSON.parseArray()
           String --->Map类型JSON.parseObj()




@Test
    public void test2(){
       // String --->对象类型JSON.parseObj()
        Person person = new Person(6,"王子",1,22);
        String s = JSON.toJSONString(person);

        Person person1 = JSON.parseObject(s, Person.class);
        System.out.println(person1);

        //String--->数组类型JSON.parseArray()
     List<Person>  list = new ArrayList<Person>();
        list.add(new Person(7,"格格",0,20));
        list.add(new Person(8 ,"公主",0,21));
        list.add(new Person(9,"逍遥",1,23));
        String s1 = JSON.toJSONString(list);

        List<Person> people = JSON.parseArray(s1, Person.class);
        System.out.println(people);

        //String --->Map类型JSON.parseObj()
        Map<String,Object> map = new HashMap<String, Object>();
        map.put("小李飞刀","刀无虚发");
        map.put("飞檐走壁","乃真人也");
        String s = JSON.toJSONString(map);

        Map map1 = JSON.parseObject(s, Map.class);
        System.out.println(map1);
    }

然后应用到服务层

  @Autowired
    private PersonDao personDao;
    @Autowired
    private JedisPool jedisPool;


 public List<Person> findAll() {
        Jedis jedis = jedisPool.getResource();
        List<Person> list=null;

        long startTime = System.currentTimeMillis();
        System.out.println(startTime);
        //1.先去缓存中去查数据 ，
        String sss =jedis.get("personList");
        //2. 如果没有就从数据库查 然后存入缓存 下次可以从缓存取。
        if (sss==null){
            list = personDao.findAll();
            long endTime1 = System.currentTimeMillis();
            System.out.println("去数据库中查询数据用的时间为:"+(endTime1-startTime));
            jedis.set(JSON.toJSONString("personList") , JSON.toJSONString(list));
        }else {
            //3.如果缓存中有数据，直接反序列化，返回给控制层
            list=JSON.parseArray(sss,Person.class);
            System.out.println(list);
            long endTime2 = System.currentTimeMillis();
            System.out.println("去缓存中查询数据用的时间为:"+(endTime2-startTime));
        }
        jedis.close();
        return  list;
    }