【Redis实战】有MQ为啥不用？用Redis作消息队列！？Redis作消息队列使用方法及底层原理高级进阶

简单轻量：Redis是一个内存中的数据存储系统，具有轻量级和简单的特点。相比较专门的消息队列系统，使用Redis作为消息队列不需要引入额外的组件和依赖，可以减少系统的复杂性。
速度快：由于Redis存储在内存中，它具有非常高的读写性能。这对于需要低延迟的应用程序非常有优势。
多种数据结构支持：Redis提供了丰富的数据结构，如列表、发布/订阅、有序集合等。这使得Redis在处理不同类型的消息和任务时更加灵活。
数据持久化：Redis可以通过将数据持久化到磁盘来提供数据的持久性。这意味着即使Redis重启，之前的消息也不会丢失。
广泛的应用场景：Redis不仅可以用作消息队列，还可以用作缓存、数据库、分布式锁等多种用途。如果你的应用程序已经使用了Redis，那么使用Redis作为消息队列可以减少技术栈的复杂性。

缺点也很明显：

缺少一些高级特性：相对于专门的消息队列系统，Redis在消息队列方面的功能可能相对简单。例如，它可能缺乏一些高级消息传递功能，如消息重试、消息路由、持久化消息等。
可靠性和一致性：Redis的主要设计目标是提供高性能和低延迟，而不是强一致性和高可靠性。在某些情况下，Redis可能会丢失消息，或者在出现故障时可能无法提供持久性保证。

应用场景：

适用于简单的中小型项目如果功能简单，访问量并不大可以考虑
如果你的应用程序对可靠性和高级功能有严格要求，并且需要处理大量的消息和复杂的消息路由，那么使用专门的消息队列系统可能更合适。

Redis实现消息队列系统实现步骤：

配置Redis：

首先，确保你已经正确地配置了Redis和Lettuce依赖项，并创建了LettuceConnectionFactory对象。

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

  redis:
    host: 
    port: 6379
    password: 
    lettuce:
      pool:
        max-active: 1000
        max-idle: 1000
        min-idle: 0
        time-between-eviction-runs: 10s
        max-wait: 10000

创建一个RedisTemplate对象，并将LettuceConnectionFactory设置为其连接工厂：

 @Bean
    public RedisTemplate<String, String> redisTemplate(LettuceConnectionFactory connectionFactory) {
        RedisTemplate<String, String> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);
        template.setDefaultSerializer(new StringRedisSerializer());
        return template;
    }

设置RedisTemplate的序列化器。在消息队列中，你可以使用默认的序列化器，即StringRedisSerializer，它会将消息以字符串的形式进行存储和传输。可以通过以下代码设置默认的序列化器：

redisTemplate.setDefaultSerializer(new StringRedisSerializer());

实现消息的发布和订阅功能。

发布消息：
```
redisTemplate.convertAndSend("channel_name", "message_payload");
```
在上述代码中，"channel_name"是消息的通道名称，"message_payload"是要发布的消息内容。
订阅消息：
首先，创建一个MessageListener实现类来处理接收到的消息：

public class MessageListenerImpl implements MessageListener {
    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 处理接收到的消息
        String channel = new String(message.getChannel());
        String payload = new String(message.getBody());
        // 执行自定义的逻辑
    }
}

创建一个LettuceMessageListenerAdapter对象，并提供MessageListener实现类：

LettuceMessageListenerAdapter listenerAdapter = new LettuceMessageListenerAdapter(new MessageListenerImpl());
listenerAdapter.afterPropertiesSet();

创建一个RedisMessageListenerContainer对象，并配置它的LettuceConnectionFactory和监听适配器：

RedisMessageListenerContainer listenerContainer = new RedisMessageListenerContainer();
listenerContainer.setConnectionFactory(lettuceConnectionFactory);
listenerContainer.addMessageListener(listenerAdapter, new ChannelTopic("通道名称"));
listenerContainer.start();

以上的方案好处就是可以很明显的知道监听者在哪个部分监听对应通道的信息然而业务当中如果每一个对应模块的业务和通道都建立一个监听者来进行监听（我们假设每一个就业务所要得到消息以后所执行的逻辑都不相同）那这个工作量就会暴增

于是就有了第二种写法：

实战与改良

/***
 * @title MessageManager
 * @author SUZE
 * @Date 2-17
 **/
@Component
public class ReservedMessageManager {
    private String ListenerId;
    private String UserId;
    private String message;
    private final RedisTemplate<String, String> redisTemplate;

    @Autowired
    public ReservedMessageManager(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
        subscribeToChannel("reserved");
    }
    @Resource
    private SmsServer smsServer;

    public void publishMessage(String channel, reserveMessage message) {
        String  Message=serialize(message);
        redisTemplate.convertAndSend("channel_name", "message_payload");
        redisTemplate.convertAndSend(channel, Message);
    }
    // 接收到消息时触发的事件
    private void handleReserveMessage(String channel, reserveMessage reserveMessage) {
        if (reserveMessage != null) {
            String userId = reserveMessage.getUserId();
            String ListenerId=reserveMessage.getListenerId();
            String message = reserveMessage.getMessage();
            //TODO 处理接收到的消息逻辑 这里后续要对Message进行一个检测他有wait agree和refused和over四种状态 思种状态就是不一样的发送内容
            switch (message){
                //TODO 消息要给两边都发 所以要发两份 发信息的文案
                case "wait":

                    smsServer.sendSms(userId,ListenerId,message);
                    break;
                case "agree":

                    smsServer.sendSms(userId,ListenerId,message);
                    break;
                case "refuse":

                    smsServer.sendSms(userId,ListenerId,message);
                    break;
                case "over":
                    //这里要操作文档系统了

                    //拒绝的话 那就要监听一下
                    smsServer.sendSms(userId,ListenerId,message);
                    break;

            }
            //smsServer.sendSms(userId,ListenerId,message);
            // 其他处理逻辑...
        }
    }

    public void subscribeToChannel(String channel) {
        redisTemplate.execute((RedisCallback<Object>) (connection) -> {
            connection.subscribe((message, pattern) -> {
                String channelName = new String(message.getChannel());
                byte[] body = message.getBody();
                // 解析接收到的消息
                switch (channelName){
                    case "reserved":
                        reserveMessage reserveMessage = deserializeMessage(new String(body));
                        handleReserveMessage(channelName, reserveMessage);
                        break;
                    //还有其他的通道 例如refuse就是一个 拒绝通道 专门监听拒绝的理由
                }
            }, channel.getBytes());
            return null;
        });
    }
    // 反序列化
    private reserveMessage deserializeMessage(String body) {
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            return objectMapper.readValue(body, reserveMessage.class);
        } catch (IOException e) {
            // 处理反序列化异常
            e.printStackTrace();
            return null;
        }
    }

    // 序列化
    public String serialize(reserveMessage reserveMessage) throws SerializationException {
        if (reserveMessage == null) {
            return null;
        }
        try {
            ObjectMapper objectMapper = new ObjectMapper();
            return objectMapper.writeValueAsString(reserveMessage);
        } catch (JsonProcessingException e) {
            throw new SerializationException("Error serializing object", e);
        }
    }

}

代码解释

我把消息处理的系统中心化处理，也就是说是这个监听系统他可以监听reserved通道的所有业务类型，我这里列举了四种wait，agree，refuse，over四种但如果是更大的业务体系同一个通道可能面临着更多可能性分支那如果按照第一套的方案需要对每一个具体业务实现一个监听者工作量就很大（可能这样耦合会低一些吧）

但是我这样把消息集中处理然后短信发送系统就专门只做短信发送的事情 xx系统就只做对应的工作就能把工作上的耦合度大大降低

那么大家应该也注意到我的两个负责序列化和反序列化的方法了吧这是因为业务需求要把对象封装成一个类所以这里的方案就是在信息中心处理器上自定义一个序列化方案（如果再做得好一点其实可以把这个序列器抽理出来封装为一个抽象方法用泛型来定义返回结果和参数这样就能序列化所有引用的类型了）

遇到的问题：

对了中途遇到了这样一个错误

原因与分析：

这个是改好的封装类：

@Data
public class reserveMessage {
    private String UserId;
    private String ListenerId;
    private String message;


    public reserveMessage() {
        // 默认构造函数
    }
    public reserveMessage(String userId, String ListenerId,String message) {
        this.UserId = userId;
        this.ListenerId = ListenerId;
        this.message=message;
    }


}

实际业务中的测试

发布服务

【Redis实战】有MQ为啥不用？用Redis作消息队列！？Redis作消息队列使用方法及底层原理高级进阶-LMLPHP

订阅服务（监听服务）

【Redis实战】有MQ为啥不用？用Redis作消息队列！？Redis作消息队列使用方法及底层原理高级进阶-LMLPHP

测试结果

【Redis实战】有MQ为啥不用？用Redis作消息队列！？Redis作消息队列使用方法及底层原理高级进阶-LMLPHP

成功

这里面的打印是代替了原本业务中的短信发送也算是成了

这一期就到这感谢观看

图灵重生我名苏泽