SpringBoot集成redisCluster(redis集群)实现发布订阅、消息队列

水深无声 2022-03-16 12:42 1172阅读 0赞

SpringBoot+redisCluster封装RedisTemplate，通过RedisTemplate操作Redis,并简单的实现消息队列、发布订阅的功能；

首先导入Maven:

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

在springboot的application.properties配置文件中配置redis集群的地址：

#redis
    spring.redis.cluster.nodes=10.2.102.244:6379,10.2.102.244:6380,10.2.102.244:6381

下图是我的一个项目结构：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x5bHk0NDEz_size_16_color_FFFFFF_t_70][]

**使用@Configuration注解初始化redisCluter配置，返回出一个封装后的RedisTemplate和一个JedisCluster对象用来操作我们redis集群和封装redis的操作命令：RedisCluterConfig类**

package pers.ly.learn.redisCluster.config;
    
    import java.lang.reflect.Field;
    
    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 com.fasterxml.jackson.annotation.PropertyAccessor;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.fasterxml.jackson.annotation.JsonAutoDetect;
    
    import redis.clients.jedis.JedisCluster;
    import redis.clients.jedis.JedisPoolConfig;
    
    @Configuration
    public class RedisCluterConfig extends CachingConfigurerSupport{
    	
    	@Value("${spring.redis.cluster.nodes}")
    	private String clusterNodes;
    	
    	
    	@Bean(name="clusterConfig")
        public RedisClusterConfiguration clusterConfig(){
        	RedisClusterConfiguration config = new RedisClusterConfiguration();
        	String[] nodes = clusterNodes.split(",");
        	for(String node : nodes){
        		String[] host =  node.split(":");
        		RedisNode redis = new RedisNode(host[0], Integer.parseInt(host[1]));
        		config.addClusterNode(redis);
        	}
        	return config;
        }
    	
        @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
        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;
        }
        
        /**
    	 * 通过反射获取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 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;
    	}
    	 
    	 
    	 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;
    	   }
    
    }

**通过@Autowired注入RedisTemplate自己封装一个redis操作的一个辅助类，当然这个可以根据自己的需要来封装：RedisOperationHepler类：**

package pers.ly.learn.redisCluster.hepler;
    
    import java.io.Serializable;
    import java.util.List;
    import java.util.Set;
    import java.util.concurrent.TimeUnit;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.HashOperations;
    import org.springframework.data.redis.core.ListOperations;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.data.redis.core.SetOperations;
    import org.springframework.data.redis.core.ValueOperations;
    import org.springframework.data.redis.core.ZSetOperations;
    import org.springframework.stereotype.Component;
    
    /**
     * redis辅助类
     * @author banma
     */
    @Component
    public class RedisOperationHepler {
    	
    	@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);
        }
        /**
         * 自增数（计数器）
         * @param k
         * @param l
         * @return
         */
        public Long increment(String k, Long l) {
            return redisTemplate.opsForValue().increment(k, l);
        }
    
    }

**刚才我们自己封装了一个redis操作的辅助类RedisOperationHepler，通过这个类我们可以实现写入/读取缓存、删除等基本操作。这就不说了；接下来实现redis的一个消息队列功能伪代码；**

**定义一个生产者接口和一个实现类：Producers、ProducersImpl，redis消息队列的实现是基于数据类型list的推入和弹出的一个功能，当一个生产者生产一个消息存入list中，消费者通过blpop命令阻塞队列弹出生产者的消息。**

package pers.ly.learn.redisCluster.service;
    
    import org.springframework.stereotype.Service;
    
    @Service
    public interface Producers {
    	
    	void productCar();
    	
    	void productPhone();
    
    }
    
    
    
    package pers.ly.learn.redisCluster.service.impl;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import pers.ly.learn.redisCluster.service.Producers;
    import redis.clients.jedis.JedisCluster;
    
    @Service
    public class ProducersImpl implements Producers {
    
    	@Autowired
    	private JedisCluster jedisCluster;
    	
    	@Override
    	public void productCar() {
    		System.out.println("我是一个汽车生产者，正在生产汽车");
    		for (int i = 0; i < 10; i++) {
    			jedisCluster.rpush("car", "宝马" + i);
    		}
    	}
    
    	@Override
    	public void productPhone() {
    		System.out.println("我是一个手机生产者，正在生产汽车");
    		for (int i = 0; i < 10; i++) {
    			jedisCluster.rpush("phone", "iphone" + i);
    		}	}
    
    }

**定义一个消费只接口和实现类：Consumers、ConsumersImpl去消费对应的消息：可以有多个消费者去消费一条消息，但是一条消息只能被消费一次，也就是说一条消息只能被A消费者或者B消费者中的一个去消费，因为消息消费完之后会被弹出也就是删除。**

package pers.ly.learn.redisCluster.service;
    
    import org.springframework.stereotype.Service;
    
    @Service
    public interface Consumers {
    	
    	void carConsumer();
    	
    	void phoneConsumer();
    
    }
    
    
    package pers.ly.learn.redisCluster.service.impl;
    
    import java.util.List;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import pers.ly.learn.redisCluster.service.Consumers;
    import redis.clients.jedis.JedisCluster;
    @Service
    public class ConsumersImpl implements Consumers {
    	
    	@Autowired
    	private JedisCluster jedisCluster;
    
    	@Override
    	public void carConsumer() {
    		System.out.println("我是一个汽车消费者，正在消费汽车");
    		while (true) {
    			//阻塞式brpop，List中无数据时阻塞，参数0表示一直阻塞下去，直到List出现数据
    			List<String> listingList = jedisCluster.blpop(10, "car");
    			System.out.println("正在消费汽车：{}" + listingList.get(1));
    		}
    		
    	}
    
    	@Override
    	public void phoneConsumer() {
    		System.out.println("我是一个手机消费者，正在消费手机");
    		while (true) {
    			//阻塞式brpop，List中无数据时阻塞，参数0表示一直阻塞下去，直到List出现数据
    			List<String> listingList = jedisCluster.blpop(10, "phone");
    			System.out.println("正在消费手机：{}" + listingList.get(1));
    		}
    	}
    
    }

**测试代码：使用spring的Junit单元测试**

package pers.ly.learn;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import pers.ly.learn.redisCluster.service.Consumers;
    import pers.ly.learn.redisCluster.service.Producers;
    import pers.ly.learn.redisCluster.service.Publisher;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class RedisTest {
    	@Autowired
    	private Consumers consumers;
    	@Autowired
    	private Producers producers;
    	@Autowired
    	private Publisher publisher;
    	
    	
    	/**
    	 * x消息队列测试
    	 * @throws InterruptedException
    	 */
    	@Test
    	public void MessageQueueTest() throws InterruptedException {
    		producers.productCar();
    		Thread.sleep(2000);
    		consumers.carConsumer();
    		
    	}
    	
    	
    	/**
    	 * 发布订阅测试
    	 * @throws InterruptedException
    	 */
    	@Test
    	public void PublishAndsubscriTest() {
    		publisher.topic1Publisher();
    		publisher.topic2Publisher();
    	}
    
    }

**以上是redis的消息队列实现伪代码，接下来实现redis的一个发布订阅功能；发布订阅：首先订阅者会通过RedisMessageListenerContainer这个对象去监听订阅的Topics,当发布者往这些Topic发布消息， 对应的监听器会通知下游的订阅者去消费这些消息，这个模式下，一条消息可以被多个订阅者同时消费； **

**定义一个发布者接口和实现类：Publisher、PublisherImpl：**

package pers.ly.learn.redisCluster.service;
    
    import org.springframework.stereotype.Service;
    
    @Service
    public interface Publisher {
    	
    	void topic1Publisher();
    	
    	void topic2Publisher();
    
    }
    
    
    
    
    package pers.ly.learn.redisCluster.service.impl;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import pers.ly.learn.redisCluster.hepler.RedisOperationHepler;
    import pers.ly.learn.redisCluster.service.Publisher;
    @Service
    public class PublisherImpl implements Publisher {
    
    	
    	@Autowired
    	private RedisOperationHepler redisOperationHepler;
    	
    	@Override
    	public void topic1Publisher() {
    		System.out.println("我正在往Topic1发消息...");
    		redisOperationHepler.convertAndSend("topic1", "我是topic1发布的信息");
    	}
    
    	@Override
    	public void topic2Publisher() {
    		System.out.println("我正在往Topic2发消息...");
    		redisOperationHepler.convertAndSend("topic2", "我是topic2发布的信息");
    	}
    
    }

**定义一个订阅者接口和实现类：Subscriber、SubscriberImpl：**

package pers.ly.learn.redisCluster.service;
    
    import org.springframework.stereotype.Service;
    
    @Service
    public interface Subscriber {
    	
    	void subscriberTopic1(String message);
    	
    	void subscriberTopic2(String message);
    
    }
    
    
    
    package pers.ly.learn.redisCluster.service.impl;
    
    import org.springframework.stereotype.Service;
    
    import pers.ly.learn.redisCluster.service.Subscriber;
    @Service
    public class SubscriberImpl implements Subscriber {
    
    
    	@Override
    	public void subscriberTopic1(String message) {
    		 System.out.println("我是订阅频道1收到T消息：{}" + message);
    		
    	}
    
    	@Override
    	public void subscriberTopic2(String message) {
    		System.out.println("我是订阅频道2收到T消息：{}" + message);
    		
    	}
    
    }

**定义一个发布订阅的一个监听器：TopicMsgListener类**

package pers.ly.learn.redisCluster.msgListener;
    
    import java.util.ArrayList;
    import java.util.List;
    
    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;
    
    import pers.ly.learn.redisCluster.service.Subscriber;
    
    @Configuration
    public class TopicMsgListener {
    	
    	@Autowired
    	private Subscriber subscriber;
    
    	/**
    	 * redis发布订阅的监听信息
    	 * @param connectionFactory
    	 * @param listenerAdapter
    	 * @return
    	 */
    	@Bean
    	RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
    											MessageListenerAdapter listenerAdapter,MessageListenerAdapter listenerAdapter2) {
    		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    		container.setConnectionFactory(connectionFactory);
    //		List<PatternTopic> topics = new ArrayList<>();
    //		topics.add(new PatternTopic("topic1"));
    //		topics.add(new PatternTopic("topic2"));
    //		container.addMessageListener(listenerAdapter, topics);
    //		container.addMessageListener(listenerAdapter2, topics);
    		
    		//监听器1
    		container.addMessageListener(listenerAdapter, new PatternTopic("topic1"));
    		
    		//监听器2
    		container.addMessageListener(listenerAdapter2, new PatternTopic("topic2"));
    		return container;
    	}
    
    	/**
    	 * 监听方法
    	 * @return
    	 */
    	@Bean(name = "listenerAdapter")
    	MessageListenerAdapter listenerAdapter() {
    		// 回调数据处理方法
    		return new MessageListenerAdapter(subscriber, "subscriberTopic1");
    	}
    	
    	/**
    	 * 监听方法2
    	 * @return
    	 */
    	@Bean(name = "listenerAdapter2")
    	MessageListenerAdapter listenerAdapter2() {
    		// 回调数据处理方法
    		return new MessageListenerAdapter(subscriber, "subscriberTopic2");
    	}
    
    }

**测试发布订阅功能：**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x5bHk0NDEz_size_16_color_FFFFFF_t_70]: /images/20220316/b92761807e924d35bdbbf6351d900659.png