如果发布的redis是当前springboot配置的Redis,那就直接进行订阅!
import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.listener.PatternTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; @Configuration public class RedisSubscriber extends JedisPubSub{ @Bean public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); //订阅了一个叫chat 的通道 container.addMessageListener(new MessageListener(){ @Override public void onMessage(Message message, byte[] pattern) { String msg = new String(message.getBody()); System.out.println(new String(pattern) + "主题发布:" + msg); //获取到通道消息进行解析msg,并处理 } }, new PatternTopic("xxx")); //xxx:具体的通道名。如果是多个通道,那就直接复制setConnectionFactory就好 return container; } }
如果订阅的消息不是,当前配置的Redis,那就需要从新起一个线程,不然直接获取连接,会导致线程阻塞,因为redis本身是但线程的。
port org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import redis.clients.jedis.JedisPool; import redis.clients.jedis.JedisPoolConfig; import redis.clients.jedis.JedisPubSub; @Configuration public class RedisSubscriber extends JedisPubSub{ //在yaml里面的自定义的redis的链接进行获取 @Value("${video.host}") //ip private String host; @Value("${video.port}") //端口 private String port; @Value("${video.password}") //密码 private String password; @Bean public void redisMessageListenerContainer() { JedisPoolConfig jedisPoolConfig = new JedisPoolConfig(); jedisPoolConfig.setMaxIdle(100); jedisPoolConfig.setMaxWaitMillis(10000); jedisPoolConfig.setMinIdle(8); jedisPoolConfig.setMaxTotal(500); jedisPoolConfig.setJmxEnabled(true); jedisPoolConfig.setTestOnBorrow(true); jedisPoolConfig.setTestOnReturn(true); JedisPool jedisPool = new JedisPool(jedisPoolConfig, host, Integer.valueOf(port), 100000, password); SubThread subThread=new SubThread(jedisPool); subThread.start(); } }
import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; public class SubThread extends Thread { private final JedisPool jedisPool; private final Subscriber subscriber = new Subscriber(); //订阅通道 private final String Channel1 = "Channel1"; private final String Channel2= "Channel2"; public SubThread(JedisPool jedisPool) { super("SubThread"); this.jedisPool = jedisPool; } @Override public void run() { Jedis jedis = null; try { jedis = jedisPool.getResource(); //取出一个连接 jedis.subscribe(subscriber, Channel1,Channel2) 【通道可配置单个或者多个,底层源码有体现】 } catch (Exception e) { System.out.println(e); } finally { if (jedis != null) { jedis.close(); } } } }
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import redis.clients.jedis.JedisPubSub; @Component public class Subscriber extends JedisPubSub { // 这样写是因为 service无法注入的原因 public static XxxxService xxxxxService; @Autowired public void setSenderService(XxxxService xxxxxService){ Subscriber.XxxxService = xxxxxService; } public Subscriber(){} @Override public void onMessage(String channel, String message) { //收到消息会调用 //视频告警 //System.out.println("【"+channel + "】主题发布:" + message); if("Channel1".equals(channel)){ //具体逻辑编写 } } }