SpringBoot整合Redis实现消息发布订阅

旧城等待， 2023-08-17 17:29 114阅读 0赞

## 前言： ##

*  SpringBoot整合Redis此处不作赘述
 *  为减少篇幅，只有部分代码（不影响使用）
 *  本文基于lombok插件以及logback日志，所以如果某些注解报错或者无法导入基本是这两处的问题
 *  勿喷

## 正文 ##

*  Redis常量配置（个人习惯，本类按需求而定）
    
        public class RedisConstant {
            /**
           	 * 消息订阅点赞主题
           	 */
           	public static final String TOPIC_PRAISE = "TOPIC_PRAISE";
           	/**
           	 * 消息订阅收藏主题
           	 */
           	public static final String TOPIC_COLLECT = "TOPIC_COLLECT";
           	/**
           	 * 消息订阅评论主题
           	 */
           	public static final String TOPIC_COMMENT = "TOPIC_COMMENT";
           	/**
           	 * 消息订阅关注主题
           	 */
           	public static final String TOPIC_FOCUS = "TOPIC_FOCUS";
        }
 *  配置消息处理器
    
        package com.java.single.service.impl;
        import java.util.concurrent.CountDownLatch;
        import com.java.single.entity.res.MessageCollect;
        import com.java.single.entity.res.MessageComment;
        import com.java.single.entity.res.MessageFocus;
        import com.java.single.util.PushUtil;
        import org.springframework.beans.factory.annotation.Autowired;
        import org.springframework.stereotype.Service;
        import com.alibaba.fastjson.JSONObject;
        import lombok.extern.slf4j.Slf4j;
        
        @Slf4j
        @Service
        public class ReceiverServiceImpl {
        
        	private CountDownLatch latch;
        
        	@Autowired
        	public ReceiverServiceImpl(CountDownLatch latch) {
        		this.latch = latch;
        	}
        
        	public void praiseReceive(MessagePraise messagePraise) {
        		log.info("消费点赞数据:[{}]", messagePraise);
        		PushUtil.pushOne(messagePraise.getToUserId(), "新消息",
        				messagePraise.getUserName() + "点赞您的" + messagePraise.getTitle() + " 作品");
        		latch.countDown();
        	}
        
        	public void collectReceive(MessageCollect messageCollect) {
        		log.info("消费收藏数据:[{}]", messageCollect);
        		PushUtil.pushOne(messageCollect.getToUserId(), "新消息",
        				messageCollect.getUserName() + "收藏了您的" + messageCollect.getTitle() + " 作品");
        		latch.countDown();
        	}
        
        	public void commentReceive(MessageComment messageComment) {
        		log.info("消费评论数据:[{}]", messageComment);
        		PushUtil.pushOne(messageComment.getToUserId(), "您有新的评论",
        				messageComment.getUserName() + "对" + messageComment.getTitle() + "进行了评论:" + messageComment.getText());
        		latch.countDown();
        	}
        
        	public void focusReceive(MessageFocus messageFocus) {
        		log.info("消费关注数据:[{}]", messageFocus);
        		PushUtil.pushOne(messageFocus.getToUserId(), "新消息", messageFocus.getUserName() + "关注了您");
        		latch.countDown();
        	}
        }
    
    上述为消息处理器，目前无用。需要后面配置，PushUtil为个人封装的个推消息推送工具类，可以删除，根据自己逻辑决定相关操作。  
    上述一共写了四种主题，以其中一种主题为例，实际项目中按照自己需求更改
    
        package com.java.single.entity.res;
         import java.io.Serializable;
         import lombok.Data;
         import lombok.NoArgsConstructor;
         
         @Data
         @NoArgsConstructor
         public class MessageCollect implements Serializable {
         	/**
         	 * 
         	 */
         	private static final long serialVersionUID = 1L;
         	/**
         	 * 收藏数据标识
         	 */
         	private String id;
         	/**
         	 * 收藏用户标识
         	 */
         	private String userId;
         	/**
         	 * 收藏用户名字
         	 */
         	private String userName;
         	/**
         	 * 收藏数据名称
         	 */
         	private String title;
         	/**
         	 * 收藏目标用户
         	 */
         	private String toUserId;
         	/**
         	 * 
         	 * @param id
         	 *            收藏数据标识
         	 * @param userId
         	 *            收藏用户标识
         	 * @param userName
         	 *            收藏用户名字
         	 * @param title
         	 *            收藏数据名称
         	 * @param toUserId
         	 *            收藏目标用户
         	 */
         	public MessageCollect(String id, String userId, String userName, String title, String toUserId) {
         		super();
         		this.id = id;
         		this.userId = userId;
         		this.userName = userName;
         		this.title = title;
         		this.toUserId = toUserId;
         	}
        }
    
    注意，这个类有两点需要注意：1、必须实现序列化。2、必须有无参构造方法。
 *  配置RedisMessageListener配置类
    
        import java.util.concurrent.CountDownLatch;
        import com.java.single.constant.RedisConstant;
        import com.java.single.service.impl.ReceiverServiceImpl;
        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 org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
        import com.fasterxml.jackson.annotation.JsonAutoDetect;
        import com.fasterxml.jackson.annotation.PropertyAccessor;
        import com.fasterxml.jackson.databind.ObjectMapper;
        
        @Configuration
        public class RedisMessageListener {
        
        	/**
        	 * redis消息监听器容器
        	 * 
        	 * @param connectionFactory
        	 * @param praiseListenerAdapter
        	 *            点赞消息订阅处理器
        	 * @param collectListenerAdapter
        	 *            收藏消息订阅处理器
        	 * @param commentListenerAdapter
        	 *            评论消息订阅处理器
        	 * @param focusListenerAdapter
        	 *            关注消息订阅处理器
        	 * @return
        	 */
        	@Bean
        	RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
        			MessageListenerAdapter praiseListenerAdapter, MessageListenerAdapter collectListenerAdapter,
        			MessageListenerAdapter commentListenerAdapter, MessageListenerAdapter focusListenerAdapter) {
        		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        		container.setConnectionFactory(connectionFactory);
        		
        		// 以下为修改默认的序列化方式，网上大多数消息发布订阅都是String类型,但是实际中数据类型肯定不止String类型
        		Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(
        				Object.class);
        		ObjectMapper objectMapper = new ObjectMapper();
        		objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        		objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        		jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        		
        		// 针对每一个消息处理可以设置不同的序列化方式
        		
        		praiseListenerAdapter.setSerializer(jackson2JsonRedisSerializer);
        		// 点赞主题并绑定消息订阅处理器
        		container.addMessageListener(praiseListenerAdapter, new PatternTopic(RedisConstant.TOPIC_PRAISE));
        		// 收藏主题并绑定消息订阅处理器
        		collectListenerAdapter.setSerializer(jackson2JsonRedisSerializer);
        		container.addMessageListener(collectListenerAdapter, new PatternTopic(RedisConstant.TOPIC_COLLECT));
        		// 评论主题并绑定消息订阅处理器
        		commentListenerAdapter.setSerializer(jackson2JsonRedisSerializer);
        		container.addMessageListener(commentListenerAdapter, new PatternTopic(RedisConstant.TOPIC_COMMENT));
        		// 关注主题并绑定消息订阅处理器
        		focusListenerAdapter.setSerializer(jackson2JsonRedisSerializer);
        		container.addMessageListener(focusListenerAdapter, new PatternTopic(RedisConstant.TOPIC_FOCUS));
        		return container;
        	}
        
        	/**
        	 * 点赞消息订阅处理器,并指定处理方法
        	 * 
        	 * @param receiver
        	 * @return
        	 */
        	@Bean
        	MessageListenerAdapter praiseListenerAdapter(ReceiverServiceImpl receiver) {
        		MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "praiseReceive");
        		return messageListenerAdapter;
        	}
        
        	/**
        	 * 收藏消息订阅处理器,并指定处理方法
        	 * 
        	 * @param receiver
        	 * @return
        	 */
        	@Bean
        	MessageListenerAdapter collectListenerAdapter(ReceiverServiceImpl receiver) {
        		MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "collectReceive");
        		return messageListenerAdapter;
        	}
        
        	/**
        	 * 评论消息订阅处理器,并指定处理方法
        	 * 
        	 * @param receiver
        	 * @return
        	 */
        	@Bean
        	MessageListenerAdapter commentListenerAdapter(ReceiverServiceImpl receiver) {
        		MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "commentReceive");
        		return messageListenerAdapter;
        	}
        
        	/**
        	 * 关注消息订阅处理器,并指定处理方法
        	 * 
        	 * @param receiver
        	 * @return
        	 */
        	@Bean
        	MessageListenerAdapter focusListenerAdapter(ReceiverServiceImpl receiver) {
        		MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "focusReceive");
        		return messageListenerAdapter;
        	}
        
        	@Bean
        	ReceiverServiceImpl receiver(CountDownLatch latch) {
        		return new ReceiverServiceImpl(latch);
        	}
        
        	@Bean
        	CountDownLatch latch() {
        		return new CountDownLatch(1);
        	}
        }
    
    注释应该已经足够，注意一定要修改序列化方法，否则使用的是默认是String类型的序列化方法，ReceiverServiceImpl方法接收到String类型的参数还需要再次转化为指定类！！！！
 *  发布消息
    
        redisTemplate.convertAndSend(RedisConstant.TOPIC_COLLECT,
        					new MessageCollect(id, userPublish.getUserId(), userName, title, userId));
    
    在使用的时候需要将构造函数的参数设置成自己的，这样项目启动之后执行上述方法后ReceiverServiceImpl的collectReceive方法就会自动处理这条消息，决定什么方法执行什么操作的配置在RedisMessageListener中实现。

日志打印：

消费收藏数据:[MessageCollect(id=c8738aed3ef9421898ebb4be62a11111, userId=6e5c7e33cbd34453b3273ed775e11111, userName=single_cong, title=CSDN博客, toUserId=d8e87f6615f64368a53bd6b4dbebc111)]

## 总结 ##

网上绝大部分都是String类型的消息发布订阅，肯定无法满足现实需求，所以要自定义序列化方式并实现主题与对应的处理方法的配置，在这里需要注意两点，第一点是序列化类型的选择，第二是这里序列化的类需要有无参构造方法（给人的感觉类似MyBatis返回结果集时报构造函数不存在时添加无参构造函数就能解决一样）。