专业的JAVA编程教程与资源

网站首页 > java教程 正文

Java实现Redis发布订阅

temp10 2025-05-28 20:18:37 java教程 4 ℃ 0 评论

当你做一个小型项目时,需要实现消息队列模式,但是却不想引用常见的MQ,因为它们实在是太重了,引入它们会给项目带来复杂度。而redis作为一个轻量级框架,主要作为缓存使用,而你正好可以通过redis实现消息的发布订阅。

一、如何使用

以SpringBoot项目为模板,来举例说明redis如何在Java中实现。

Java实现Redis发布订阅

1.1 首先引入redis的jar包

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

1.2 配置redis连接池

@Configuration
public class JedisPoolFactory {
    @Bean
    public JedisPool redisFactory(){
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxIdle(10);
        jedisPoolConfig.setMaxWaitMillis(-1);
        jedisPoolConfig.setMinIdle(0);
        return new JedisPool(jedisPoolConfig, "127.0.0.1", 6379);
    }
}

1.3 发布消息

@Component
public class Publisher{

    @Autowired
    private JedisPool jedisPool;

    @Autowired
    private ObjectMapper objectMapper;
  	// 调用run方法传参,实现消息的发布
    public void run(StudentBean studentBean) {
        Jedis jedis = jedisPool.getResource();
        try {
            String value = objectMapper.writeValueAsString(studentBean);
            // 从mychannel的频道上推送消息
            jedis.publish("mychannel", value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }
}

1.4 消费者订阅

  1. 订阅线程
@Slf4j
public class SubscribeThread extends Thread {

    private final JedisPool jedisPool;
    private final Subscriber subscriber = new Subscriber();

    private final String channel = "mychannel";

    public SubscribeThread(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
    }

    @Override
    public void run() {
        Jedis jedis = null;
        try {
            //从redis连接池中取出一个连接
            jedis = jedisPool.getResource();
            //通过subscribe 的api去订阅,入参是订阅者和频道名
            log.info("subscribe redis, channel: {}, thread will be blocked", channel);
            jedis.subscribe(subscriber, channel);
        } catch (Exception e) {
            log.error("subsrcibe channel error: {}", e.getMessage());
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
    }
}
  1. 业务处理,需要继承JedisPubSub,然后实现其发布订阅
public class Subscriber extends JedisPubSub {

    /**
     * 收到消息会调用
     * @param channel
     * @param message
     */
    @Override
    public void onMessage(String channel, String message) {
        // 收到message处理业务
        System.out.printf("received redis published message, channel: %s, message: %s%n",channel, message);
        // TODO message
    }

    /**
     * 订阅了频道会调用
     * @param channel
     * @param subscribedChannels
     */
    @Override
    public void onSubscribe(String channel, int subscribedChannels) {
        System.out.printf("subscribe redis channel success, channel %s, subscribedChannels %d%n",
                channel, subscribedChannels);
    }

    /**
     * 取消订阅会调用
     * @param channel
     * @param subscribedChannels
     */
    @Override
    public void onUnsubscribe(String channel, int subscribedChannels) {
        System.out.printf("unsubscribe redis channel, channel %s, subscribedChannels %d%n",
                channel, subscribedChannels);
    }
}

二、总结

2.1 不能保证消息的可靠性

当发布一条消息时,如果没有订阅者或者订阅者断线时,那么这条消息就会被丢失,订阅者再次启动时无法获得该消息。

2.2 轻量级服务

如果在小型服务里面,只用到redis,而没有用到其它MQ中间件,那么可以用redis来实现消息队列,避免中间件的引入及安装。

本文暂时没有评论,来添加一个吧(●'◡'●)

欢迎 发表评论:

最近发表
标签列表