时间戳

首页 > 技术文章 > SpringBoo+Redis实现简单的发布订阅功能

SpringBoo+Redis实现简单的发布订阅功能

yiMro 2022-04-18 14:59 原文

Redis的发布订阅模式

发布订阅（Pub/Sub）：目前广泛使用的通信模型，它采用事件作为基本的通信机制，提供大规模系统所要求的松散耦合的交互模式：订阅者(如客户端)以事件订阅的方式表达出它有兴趣接收的一个事件或一类事件；发布者(如服务器)可将订阅者感兴趣的事件随时通知相关订阅者,其就是基于观察者模式设计的。不过redis的发布订阅缺点也很明显。

对于旧版的redis来说，如果一个客户端订阅了某个或者某些频道，但是它读取消息的速度不够快，那么不断的积压的消息就会使得redis输出缓冲区的体积越来越大，这可能会导致redis的速度变慢，甚至直接崩溃。也可能会导致redis被操作系统强制杀死，甚至导致操作系统本身不可用。新版的redis不会出现这种问题，因为它会自动断开不符合client-output-buffer-limit pubsub配置选项要求的订阅客户端。

任何网络系统在执行操作时都可能会遇到断网的情况。而断线产生的连接错误通常会使得网络连接两端中的一端进行重新连接。如果客户端在执行订阅操作的过程中断线，那么客户端将会丢失在断线期间的消息，这在很多业务场景下是不可忍受的。

适用场景

(1).优惠券到期到期失效

(2).微信粉丝关注公众号

(3).消息及时通讯(redis不像中间件一样,不会对发布的消息进行缓存,确认,重试等机制

添加maven的依赖文件

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

在项目的启动类什么加上

@SpringBootApplication(exclude=DataSourceAutoConfiguration.class)

因为只是简单的操作下看看效果就不配置数据库了，yml文件也不需要写直接用默认的就行。前提是本地有装redis (>_<)

通过配置的方式注入了RedisTemplate实例。

@Configuration
public class RedisConfig {

    // 编写redistamle
    @Bean
    @SuppressWarnings("all")
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory)
        throws UnknownException {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        // json序列号
        Jackson2JsonRedisSerializer<Object> serializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        serializer.setObjectMapper(om);
        // string序列号
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

        // key 采用string的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        // Hash的key 采用string的序列化方式
        template.setHashKeySerializer(stringRedisSerializer);
        // value采用Jackson2JsonRedisSerializer的序列化方式
        template.setValueSerializer(serializer);
        // Hash的value 采用Jackson2JsonRedisSerializer的序列化方式
        template.setHashValueSerializer(serializer);


        // 配置具体的序列化方式
        template.afterPropertiesSet();

        return template;
    }
}
View Code

发布者：使用Redis实现的发布，其实就是通过redisTemplate.convertAndSend()方法实现。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.stereotype.Component;

import java.util.Calendar;

@Component
public class MsgPublisher {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private ChannelTopic topic;

    public void sendMsg(String msg){
        redisTemplate.convertAndSend( topic.getTopic(), "Message: " + msg +
                ";Time:" + Calendar.getInstance().getTime());
    }
}

定义一个发布者的主题，该主题可以直接在MsgPublisher类中通过new一个出来，反正还是会加到spring容器里面的

@Bean 
public ChannelTopic topic() {
 　　return new ChannelTopic( "pubsub:queue" );
}

使用Redis实现的订阅者，需要两个步骤：

1、定义消息监听器，需要实现MessageListener接口，

2、需要配置RedisMessageListenerContainer容器，即把消息监听器添加到容器中。

监听器：主要是实现了MessageListener 接口的onMessage()方法，该方法即是用来处理监听结果的代码块。

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;

@Component
public class MsgListener implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        System.out.println( "Message received: " + message.toString());
        // 这里写咱们业务就行拉

    }
}

订阅者配置：主要定义了RedisMessageListenerContainer 容器，并把监听器放到容器中，并添加监听器监听的主题。

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.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

@Configuration
public class SubConfig {

    @Bean
    MessageListenerAdapter messageListener() {
        return new MessageListenerAdapter( new MsgListener() );
    }

    @Bean
    RedisMessageListenerContainer redisContainer(RedisConnectionFactory factory) {
        final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.addMessageListener(messageListener(), new ChannelTopic( "pubsub:queue" ));
        return container;
    }
}

定义一个测试类，通过接口调用发布者，传递msg参数，然后订阅者可以监听到该参数

import com.lb.farm.provider.config.MsgPublisher;
import lombok.AllArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/TestRedisController")
@AllArgsConstructor
public class TestRedisController {

    @Autowired
    private MsgPublisher msgPublisher;

    @GetMapping("/sendMsg/")
    public String sendMsg(String msg){
        msgPublisher.sendMsg(msg);
        return "发送成功！";
    }
}

直接用浏览器访问一下。就可以在控制台看到东西啦！

 

推荐阅读