SpringBoot集成redis的JedisCluster和RedisTemplate 实现redis的消息队列消费者-生产者模式，订阅者发布者模式

桃扇骨 2022-03-31 13:16 523阅读 0赞

一般来说，消息队列有两种场景，一种是发布者订阅者模式，一种是生产者消费者模式。利用redis这两种场景的消息队列都能够实现。  
定义：

*  **生产者消费者模式** ：生产者生产消息放到队列里，多个消费者同时监听队列，谁先抢到消息谁就会从队列中取走消息；即对于每个消息只能被最多一个消费者拥有。
 *  **发布者订阅者模式**：发布者生产消息放到队列里，多个监听队列的消费者都会收到同一份消息；即正常情况下每个消费者收到的消息应该都是一样的。

Redis不仅可作为缓存服务器，还可用作消息队列。它的列表类型天生支持用作消息队列。如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NOWllZSA_size_16_color_FFFFFF_t_70]  
由于Redis的列表是使用双向链表实现的，保存了头尾节点，所以在列表头尾两边插取元素都是非常快的。

在Redis中，List类型是按照插入顺序排序的字符串链表。和数据结构中的普通链表一样，我们可以在其头部(left)和尾部(right)添加新的元素。在插入时，如果该键并不存在，Redis将为该键创建一个新的链表。与此相反，如果链表中所有的元素均被移除，那么该键也将会被从数据库中删除。List中可以包含的最大元素数量是4294967295。  
从元素插入和删除的效率视角来看，如果我们是在链表的两头插入或删除元素，这将会是非常高效的操作，即使链表中已经存储了百万条记录，该操作也可以在常量时间内完成。然而需要说明的是，如果元素插入或删除操作是作用于链表中间，那将会是非常低效的。相信对于有良好数据结构基础的开发者而言，这一点并不难理解。

Redis List的主要操作为lpush/lpop/rpush/rpop四种，分别代表从头部和尾部的push/pop，除此之外List还提供了两种pop操作的阻塞版本blpop/brpop，用于阻塞获取一个对象。

Redis通常都被用做一个处理各种后台工作或消息任务的消息服务器。 一个简单的队列模式就是：生产者把消息放入一个列表中，等待消息的消费者用 RPOP 命令（用轮询方式）， 或者用 BRPOP 命令（如果客户端使用阻塞操作会更好）来得到这个消息。

以下列举SpringBoot集成redis的JedisCluster和RedisTemplate

引入依赖到pom.xml

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-data-redis</artifactId>
                <exclusions>
                    <exclusion>
                        <groupId>org.slf4j</groupId>
                        <artifactId>jcl-over-slf4j</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-pool2</artifactId>
            </dependency>
            <dependency>
                <groupId>redis.clients</groupId>
                <artifactId>jedis</artifactId>
            </dependency>

application.yml增加redis集群配置

spring:
      redis:
        password:
        clusters: 10.10.1.238:7000,10.10.1.238:7001,10.10.1.238:7002,10.10.1.238:7003,10.10.1.238:7004,10.10.1.238:7005

RedisConfig配置

package com.example.myframe.config;
    
    import com.fasterxml.jackson.annotation.JsonAutoDetect;
    import com.fasterxml.jackson.annotation.PropertyAccessor;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.cache.annotation.CachingConfigurerSupport;
    import org.springframework.cache.interceptor.KeyGenerator;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.connection.RedisClusterConfiguration;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    import org.springframework.data.redis.connection.RedisNode;
    import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
    import redis.clients.jedis.JedisCluster;
    import redis.clients.jedis.JedisPoolConfig;
    
    import java.lang.reflect.Field;
    
    @Configuration
    public class RedisConfig extends CachingConfigurerSupport { 
    	/**redis密码**/
    	@Value("${spring.redis.password}")
    	public String password;
    	
    	@Value("${spring.redis.clusters}")
    	public String cluster;
    	
        
        @Bean
        public KeyGenerator keyGenerator() { 
            return (target, method, params) -> { 
    			StringBuilder sb = new StringBuilder();
    			sb.append(target.getClass().getName());
    			sb.append(method.getName());
    			for (Object obj : params) { 
    				sb.append(obj.toString());
    			}
    			return sb.toString();
    		};
        }
    
    	public  Object getFieldValueByObject (Object object , String targetFieldName) throws Exception { 
    		// 获取该对象的Class
    		Class objClass = object.getClass();
    		// 获取所有的属性数组
    		Field[] fields = objClass.getDeclaredFields();
    		for (Field field:fields) { 
    			// 属性名称
    			field.setAccessible(true);
    			String currentFieldName = field.getName();
    			if(currentFieldName.equals(targetFieldName)){ 
    				return field.get(object); // 通过反射拿到该属性在此对象中的值(也可能是个对象)
    			}
    		}
    		return null;
    	}
    
    	/** * 通过反射获取JedisCluster * @param factory * @return */
    	@Bean
    	public JedisCluster redisCluster(RedisConnectionFactory factory){ 
    		Object object =null;
    		try { 
    			object= getFieldValueByObject(factory,"cluster");
    		} catch (Exception e) { 
    			e.printStackTrace();
    		}
    		return (JedisCluster)object;
    
    	}
    
        @Bean
        public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory factory) { 
        	StringRedisTemplate template = new StringRedisTemplate(factory);
            Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
            ObjectMapper om = new ObjectMapper();
            om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
            om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
            jackson2JsonRedisSerializer.setObjectMapper(om);
            template.setValueSerializer(jackson2JsonRedisSerializer);
            template.afterPropertiesSet();
            return template;
        }
        
        @Bean(name="factory")
        public RedisConnectionFactory factory(RedisClusterConfiguration clusterConfig){ 
            JedisConnectionFactory redisConnectionFactory=new JedisConnectionFactory(clusterConfig);
            String redisPassword = password;
        	redisConnectionFactory.setPassword(redisPassword);
            redisConnectionFactory.setPoolConfig(createJedisPoolConfig());
            redisConnectionFactory.setTimeout(30000);
            redisConnectionFactory.setUsePool(true);  
            return redisConnectionFactory;  
        }
        
        @Bean(name="clusterConfig")
        public RedisClusterConfiguration clusterConfig(){ 
        	RedisClusterConfiguration config = new RedisClusterConfiguration();
        	String[] nodes = cluster.split(",");
        	for(String node : nodes){ 
        		String[] host =  node.split(":");
        		RedisNode redis = new RedisNode(host[0], Integer.parseInt(host[1]));
        		config.addClusterNode(redis);
        	}
        	
        	return config;
        }
    
    
        public JedisPoolConfig createJedisPoolConfig(){ 
       	 JedisPoolConfig config = new JedisPoolConfig();
       	//连接耗尽时是否阻塞, false报异常,ture阻塞直到超时, 默认true
       	 config.setBlockWhenExhausted(false);
       	  
       	 //设置的逐出策略类名, 默认DefaultEvictionPolicy(当连接超过最大空闲时间,或连接数超过最大空闲连接数)
       	 config.setEvictionPolicyClassName("org.apache.commons.pool2.impl.DefaultEvictionPolicy");
       	  
       	 //是否启用pool的jmx管理功能, 默认true
       	 config.setJmxEnabled(true);
       	  
       	 //MBean ObjectName = new ObjectName("org.apache.commons.pool2:type=GenericObjectPool,name=" + "pool" + i); 默 认为"pool", JMX不熟,具体不知道是干啥的...默认就好.
       	 config.setJmxNamePrefix("pool");
       	  
       	 //是否启用后进先出, 默认true
       	 config.setLifo(true);
       	  
       	 //最大空闲连接数, 默认8个
       	 config.setMaxIdle(2000);
       	  
       	 //最大连接数, 默认8个
       	 config.setMaxTotal(5000);
       	 
       	 //获取连接时的最大等待毫秒数(如果设置为阻塞时BlockWhenExhausted),如果超时就抛异常, 小于零:阻塞不确定的时间, 默认-1
       	 config.setMaxWaitMillis(10000);
       	  
       	 //逐出连接的最小空闲时间 默认1800000毫秒(30分钟)
       	 config.setMinEvictableIdleTimeMillis(1800000);
       	  
       	 //最小空闲连接数, 默认0
       	 config.setMinIdle(0);
       	  
       	 //每次逐出检查时 逐出的最大数目 如果为负数就是 : 1/abs(n), 默认3
       	 config.setNumTestsPerEvictionRun(3);
       	  
       	 //对象空闲多久后逐出, 当空闲时间>该值 且 空闲连接>最大空闲数 时直接逐出,不再根据MinEvictableIdleTimeMillis判断 (默认逐出策略) 
       	 config.setSoftMinEvictableIdleTimeMillis(1800000);
       	  
       	 //在获取连接的时候检查有效性, 默认false
       	 config.setTestOnBorrow(false);
       	  
       	 //在空闲时检查有效性, 默认false
       	 config.setTestWhileIdle(false);
       	  
       	 //逐出扫描的时间间隔(毫秒) 如果为负数,则不运行逐出线程, 默认-1
       	 config.setTimeBetweenEvictionRunsMillis(-1);
       	 
       	 return config;
       }
        
    
    }

RedisService.java辅助类

package com.example.myframe.redis.service;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.*;
    import org.springframework.stereotype.Service;
    
    import java.io.Serializable;
    import java.util.List;
    import java.util.Set;
    import java.util.concurrent.TimeUnit;
    
    @Service("redisService")
    public class RedisService { 
        @Autowired
        private RedisTemplate redisTemplate;
    
        /** * 写入缓存 * @param key * @param value * @return */
        public boolean set(final String key, Object value) { 
            boolean result = false;
            try { 
                ValueOperations<Serializable, Object> operations = redisTemplate.opsForValue();
                operations.set(key, value);
                result = true;
            } catch (Exception e) { 
                e.printStackTrace();
            }
            return result;
        }
        /** * 写入缓存设置时效时间 * @param key * @param value * @return */
        public boolean set(final String key, Object value, Long expireTime) { 
            boolean result = false;
            try { 
                ValueOperations<Serializable, Object> operations = redisTemplate.opsForValue();
                operations.set(key, value);
                redisTemplate.expire(key, expireTime, TimeUnit.SECONDS);
                result = true;
            } catch (Exception e) { 
                e.printStackTrace();
            }
            return result;
        }
        /** * 批量删除对应的value * @param keys */
        public void remove(final String... keys) { 
            for (String key : keys) { 
                remove(key);
            }
        }
    
        /** * 批量删除key * @param pattern */
        public void removePattern(final String pattern) { 
            Set<Serializable> keys = redisTemplate.keys(pattern);
            if (keys.size() > 0){ 
            	redisTemplate.delete(keys);
            }
                
        }
        /** * 删除对应的value * @param key */
        public void remove(final String key) { 
            if (exists(key)) { 
                redisTemplate.delete(key);
            }
        }
        /** * 判断缓存中是否有对应的value * @param key * @return */
        public boolean exists(final String key) { 
            return redisTemplate.hasKey(key);
        }
        /** * 读取缓存 * @param key * @return */
        public Object get(final String key) { 
            Object result = null;
            ValueOperations<Serializable, Object> operations = redisTemplate.opsForValue();
            result = operations.get(key);
            return result;
        }
        /** * 哈希 添加 * @param key * @param hashKey * @param value */
        public void hmSet(String key, Object hashKey, Object value){ 
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            hash.put(key,hashKey,value);
        }
    
        /** * 哈希 获取哈希的key集合 * @param key * @return */
        public Set<Object> hmKeys(String key){ 
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            return hash.keys(key);        
        }
    
        /** * 哈希 删除哈希的key * @param key * @param hashKey */
        public void hmDelete(String key,String hashKey){ 
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            hash.delete(key, hashKey);       
        }    
        
        /** * 哈希获取数据 * @param key * @param hashKey * @return */
        public Object hmGet(String key, Object hashKey){ 
            HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
            return hash.get(key,hashKey);
        }
        
        /** * 获取所有key值 * @param key * @return */
        public Set<Object>  hmKeySet(String key){ 
        	HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
        	return hash.keys(key);	
        }
        
        /** * 获取所有key值 * @param key * @return */
        public void  hmRemove(String key, Object hashKey){ 
        	HashOperations<String, Object, Object> hash = redisTemplate.opsForHash();
        	hash.delete(key, hashKey); 	
        }
        
        
        
    
        /** * 列表添加 * @param k * @param v */
        public void lPush(String k,Object v){ 
            ListOperations<String, Object> list = redisTemplate.opsForList();
            list.rightPush(k,v);
        }
    
        /** * 列表获取 * @param k * @param l * @param l1 * @return */
        public List<Object> lRange(String k, long l, long l1){ 
            ListOperations<String, Object> list = redisTemplate.opsForList();
            return list.range(k,l,l1);
        }
    
        /** * 集合添加 * @param key * @param value */
        public void add(String key,Object value){ 
            SetOperations<String, Object> set = redisTemplate.opsForSet();
            set.add(key,value);
        }
    
        /** * 集合获取 * @param key * @return */
        public Set<Object> setMembers(String key){ 
            SetOperations<String, Object> set = redisTemplate.opsForSet();
            return set.members(key);
        }
    
        /** * 集合长度 * @param key * @return */
        public Long setSize(String key){ 
            SetOperations<String, Object> set = redisTemplate.opsForSet();
            return set.size(key);
        }
    
    
        /** * 集合获取 * @param key * @param count * @return */
        public Set<Object> setMembers(String key, int count){ 
        	 SetOperations<String, Object> set = redisTemplate.opsForSet();
        	 return set.distinctRandomMembers(key, count);
        }
        
        /** * 删除集合数据 * @param key * @param value */
        public void remove(String key, Object value){ 
        	 SetOperations<String, Object> set = redisTemplate.opsForSet();
        	 set.remove(key, value);
        }
    
        /** * 有序集合添加 * @param key * @param value * @param scoure */
        public void zAdd(String key,Object value,double scoure){ 
            ZSetOperations<String, Object> zset = redisTemplate.opsForZSet();
            zset.add(key,value,scoure);
        }
    
        /** * 有序集合获取 * @param key * @param scoure * @param scoure1 * @return */
        public Set<Object> rangeByScore(String key,double scoure,double scoure1){ 
            ZSetOperations<String, Object> zset = redisTemplate.opsForZSet();
            return zset.rangeByScore(key, scoure, scoure1);
        }
        
        /** * 消息队列实现 * @param channel * @param message */
        public void convertAndSend(String channel, Object message){ 
        	redisTemplate.convertAndSend(channel, message);
        }
        
        /** * 数列添加 * @param key * @param value */
        public void addList(String key,Object value){ 
            ListOperations<String, Object> list = redisTemplate.opsForList();
            list.rightPush(key, value);
        }
    
        /** * 数列获取 * @param key * @return */
        public List<Object> getList(String key){ 
            ListOperations<String, Object> list = redisTemplate.opsForList();
            return list.range(key, 0, list.size(key));
        }
        
        /** * 左弹出数列 * @param key * @return */
        public Object popList(String key) { 
            ListOperations<String, Object> list = redisTemplate.opsForList();
            return list.leftPop(key);
        }
    
        public Long increment(String k, Long l) { 
            return redisTemplate.opsForValue().increment(k, l);
        }
        
    }

TestRedisController.java

package com.example.myframe.controller;
    
    import com.example.myframe.redis.RedisUtil;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    @RequestMapping(value = "/redis")
    public class TestRedisController { 
    	@Autowired
    	RedisUtil redisUtil;
    
    	/** * 生产者 通过此方法来往redis的list的尾部插入数据 */
    	@RequestMapping("/shengchangzhe")
    	public void shengChangZhe() { 
    		redisUtil.dealShengChangZhe();
    	}
    
    	/** * 消费者 ，通过此方法往redis的list的头部获取数据，直到list没有数据 阻塞，等到一有数据又继续获取 */
    	@RequestMapping("/xiaoFeiZhe")
    	public void xiaoFeiZhe() { 
    		redisUtil.dealXiaoFeiZhe();
    	}
    
    	/** * 发布者，发布信息，监听器一旦接收到监听，就进行操作 */
    	@RequestMapping("/faBuDingYue")
    	public void faBuDingYue() { 
    		redisUtil.dealFaBuDingYue();
    	}
    
    }

package com.example.myframe.redis;
    
    import com.example.myframe.redis.service.RedisService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    import redis.clients.jedis.JedisCluster;
    
    import java.util.List;
    
    @Component
    public class RedisUtil { 
    	@Autowired
    	JedisCluster jedisCluster;
    
    	@Autowired
    	private RedisService redisService;
    
    	/** * 生产者 */
    	public void dealShengChangZhe() { 
    		for (int i = 0; i < 10; i++) { 
    			jedisCluster.rpush("ceshi", "value1_" + i);
    		}
    	}
    
    	/** * 消费者 */
    	public void dealXiaoFeiZhe() { 
    		while (true) { 
    			//阻塞式brpop，List中无数据时阻塞，参数0表示一直阻塞下去，直到List出现数据
    			List<String> listingList = jedisCluster.blpop(0, "ceshi");
    			System.out.println("线程取数据：{}" + listingList.get(1));
    		}
    	}
    
    	/** * 发布订阅模式 */
    	public void dealFaBuDingYue() { 
    		redisService.convertAndSend("dealFaBuDingYue", "我是来发布信息的");
    	}
    }

redis消息队列监听信息

package com.example.myframe.redis.msg;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    import org.springframework.data.redis.listener.PatternTopic;
    import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
    
    @Configuration
    public class RedisMsgListener { 
    	@Autowired
    	Receiver receiver;
    
    	/** * redis消息队列监听信息 * * @param connectionFactory * @param listenerAdapter * @return */
    	@Bean
    	RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
    											MessageListenerAdapter listenerAdapter) { 
    		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    		container.setConnectionFactory(connectionFactory);
    		container.addMessageListener(listenerAdapter, new PatternTopic("dealFaBuDingYue"));
    		return container;
    	}
    
    	/** * 监听方法 * * @return */
    	@Bean(name = "listenerAdapter")
    	MessageListenerAdapter listenerAdapter() { 
    		// 回调数据处理方法
    		return new MessageListenerAdapter(receiver, "dealJt");
    	}
    }

package com.example.myframe.redis.msg;
    
    import com.example.myframe.redis.service.RedisService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.stereotype.Service;
    
    @Service("receiver")
    public class Receiver{ 
    
    	@Autowired
    	private RedisService redisService;
    	@Autowired
    	private StringRedisTemplate redisTemplate;
    
        /** * 清除外部广告位本地缓存 * @param message */
        public  void dealJt(String message){ 
            System.out.println("我是用来监听信息的");
            System.out.println(message);
        }
    
    	/** * 清除外部广告位本地缓存 * @param message */
    	public  void dealJt1(String message){ 
    		System.out.println("我是用来监听信息的1");
    		System.out.println(message);
    	}
    
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NOWllZSA_size_16_color_FFFFFF_t_70]: /images/20220331/77d1bf9e4a4c4cac96850a884bd4210b.png