关于在使用Redssion发布的订阅遇到大bug

本文主要是介绍关于在使用Redssion发布的订阅遇到大bug，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

故事开始

在我开发一个仿微信的一个项目的时候，有一个功能在服务端接受到客户端消息时需要做出反应

比如：有多个客户端向服务端发送消息 ，然后服务端将消息返回给多个客户端，此时可以利用redis的发布订阅
通过将消息发送给同一个主题监听这个主题是否接受到消息从未做出相应操作。

我使用了redis 的客户端ReissionClient中的publish方法进行推送消息，使用addlisten进行消息监听
代码如下：
redission的配置

package com.jjy.easy_chat.entity.config;import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;/*** 自定义序列化方式*/
@Configuration
public class RedisConfig {private static final Logger logger = LoggerFactory.getLogger(RedisConfig.class);@Value("${spring.redis.host}")private String redisHost;@Value("${spring.redis.port}")private String redisPort;@Bean("redisTemplate")public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();redisTemplate.setKeySerializer(new StringRedisSerializer());redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());redisTemplate.setHashKeySerializer(new StringRedisSerializer());redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());redisTemplate.setConnectionFactory(redisConnectionFactory);return redisTemplate;}@Bean(name = "redissonClient", destroyMethod = "shutdown")public RedissonClient redissonClient() {try {Config config = new Config();String url="redis://" + redisHost + ":" + redisPort;config.useSingleServer().setAddress(url)// 重试间隔时间（单位：毫秒）.setRetryInterval(1500)// 最大重试次数.setRetryAttempts(5)// 连接超时时间（单位：毫秒）.setTimeout(3000)// 连接池大小.setConnectionPoolSize(64)// 最小空闲连接数.setConnectionMinimumIdleSize(10);RedissonClient redissonClient = Redisson.create(config);return redissonClient;} catch (Exception e) {logger.error("redission配置出错", e);}return null;}
//    @Bean
//    public RedissonClient redissonClient() throws IOException {
//        Config config = Config.fromYAML(RedisConfig.class.getClassLoader().getResource("redission-config.yml"));
//        return Redisson.create(config);
//    }}

创建了MessageHandler去控制

package com.jjy.easy_chat.websocket;import com.jjy.easy_chat.entity.dto.MessageSendDto;
import com.jjy.easy_chat.utils.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;@Component("messageHandler")
public class MessageHandler {private final String MESSAGE_TOPIC = "message.topic";private final Logger logger = LoggerFactory.getLogger(MessageHandler.class);@Autowiredprivate RedissonClient redissonClient;@Autowiredprivate ChannelContextUtils channelContextUtils;@PostConstructpublic void init() {logger.info("RedissonClient 连接测试: {}", redissonClient.getConfig().toString());}@PostConstructpublic void listenMessage() {try {logger.info("开始监听");RTopic rTopic = redissonClient.getTopic(MESSAGE_TOPIC);int listenerId = rTopic.addListener(MessageSendDto.class, (channel, sendDto) -> {try {logger.info("收到广播消息: {}", JsonUtils.convertObj2Json(sendDto));// channelContextUtils.sendMsg(sendDto);} catch (Exception e) {logger.error("处理广播消息时出错: {}", e.getMessage(), e);}});logger.info("监听器已添加，ID为: {}", listenerId);} catch (Exception e) {logger.error("监听消息时发生异常: {}", e.getMessage(), e);}}public void sendMessage(MessageSendDto sendDto) {RTopic rTopic = redissonClient.getTopic(MESSAGE_TOPIC);logger.info("发送消息", JsonUtils.convertObj2Json(sendDto));rTopic.publish(sendDto);}}

封装了个接口进行请求

    @RequestMapping("/test")private ResponseVO getTest() {MessageSendDto messageSendDto = new MessageSendDto();messageSendDto.setMessageContent("haahah" + System.currentTimeMillis());messageHandler.sendMessage(messageSendDto);//messageRabbitMqHandler.sendMessage(messageSendDto);return getSuccessResponseVO(messageSendDto);}

最后对该接口进行请求代码没有报错

但是没有监听到消息 如果监听到消息就会输出广播消息加内容

解决过程

我在网上找了许多方法
起初
起初我以为是以为@postConstruct注解没有起到作用,我用了许多方法去解决这个问题 发现并不是这个问题
通过debug 我发现是addlisten 没有起到作用，我到处查资料找到好多方法。

1. 配置问题：确保Redisson客户端配置正确，包括正确配置了Redis服务器的地址、端口、密码等。如果配置有误，可能导致无法正确连接到Redis服务器，从而影响监听功能。

2. 网络问题：网络不稳定或连接数过多可能导致程序与Redis服务断开连接。检查网络连接是否稳定，以及是否有连接数限制。

3. Redis服务状态：检查Redis服务是否正常运行，以及是否有相关的监听功能被禁用。例如，Redis的notify-keyspace-events配置项需要开启以支持键空间通知功能。

4. 版本兼容性：确保使用的Redisson版本与Redis服务器版本兼容。不兼容的版本可能导致某些功能无法正常工作。

5. Redisson线程模型：Redisson使用异步模型来处理事件，如果线程模型配置不当，可能会影响消息的接收。

经过仔细查找 ，配置问题也没有问题，我以为是@Async 不对我重新创建了一个线程去执行这个方法发现也不是
网络上我在校园网和热点中进行切换可惜并没有卵用

昙花一现

最后我选择更新版本，我将pom.xml中的redission的版本调到了最高，redis的版本也调到了最高 然后运行发现成功了

可惜第二天早上我再次运行又不行了，真的气死

最终解决

我无可奈何，想到了消息中间件的rabbitMQ的发布订阅，既然已经到此何不放手一搏，我直接打算换一个技术栈，我创建了rabbitMQ的配置

package com.jjy.easy_chat.entity.config;import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.amqp.core.*;
@Configuration
public class RabbitMqConfig {private final String EXCAHNG_NAME="message.topic";private final String QUNEUNE_NAME="message";//创建交换机@Bean("messageExchange")public Exchange getExchange() {return ExchangeBuilder.topicExchange(EXCAHNG_NAME).durable(true).build();}@Bean("bootQueue")//创建队列public Queue getQueue() {return new Queue(QUNEUNE_NAME);}//绑定交换机//交换机绑定队列@Beanpublic Binding bindMessageQueue(@Qualifier("messageExchange") Exchange exchange, @Qualifier("bootQueue") Queue queue){return BindingBuilder.bind(queue).to(exchange).with("#.message.#").noargs();}
}

MessageRabbitMqHandler

package com.jjy.easy_chat.websocket;import com.jjy.easy_chat.entity.dto.MessageSendDto;
import com.jjy.easy_chat.utils.JsonUtils;
import org.redisson.api.RTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Component("messageRabbitMqHandler")
public class MessageRabbitMqHandler {@Autowiredprivate RabbitTemplate rabbitTemplate;@Autowiredprivate ChannelContextUtils channelContextUtils;@RabbitListener(queues = "message")public void listenMessage(MessageSendDto sendDto){logger.info("收到广播消息: {}", JsonUtils.convertObj2Json(sendDto));channelContextUtils.sendMsg(sendDto);}private final Logger logger = LoggerFactory.getLogger(MessageRabbitMqHandler.class);public void sendMessage(MessageSendDto sendDto) {logger.info("发送消息", JsonUtils.convertObj2Json(sendDto));rabbitTemplate.convertAndSend("message.topic","message",sendDto);}
}

最后我同样去请求这个结果发现成功了，后面也再有没有报错，开心！

如果我的内容对你有帮助，请点赞，评论，收藏。创作不易，大家的支持就是我坚持下去的动力

这篇关于关于在使用Redssion发布的订阅遇到大bug的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---