Java 中使用 Redis 实现消息订阅/发布

Redis 提供了 Pub/Sub (发布/订阅) 模式，允许客户端订阅频道并接收发布到这些频道的消息。以下是 Java 中使用 Redis 实现消息订阅的几种方式。

1. 使用 Jedis 客户端

添加依赖

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

基本订阅示例

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;public class RedisSubscriber {public static void main(String[] args) {// 创建 Jedis 连接Jedis jedis = new Jedis("localhost", 6379);// 创建订阅者JedisPubSub subscriber = new JedisPubSub() {@Overridepublic void onMessage(String channel, String message) {System.out.println("收到消息 - 频道: " + channel + ", 内容: " + message);}@Overridepublic void onSubscribe(String channel, int subscribedChannels) {System.out.println("订阅成功 - 频道: " + channel);}};// 订阅频道jedis.subscribe(subscriber, "myChannel");}
}

发布消息

import redis.clients.jedis.Jedis;public class RedisPublisher {public static void main(String[] args) {Jedis jedis = new Jedis("localhost", 6379);jedis.publish("myChannel", "Hello, Redis Pub/Sub!");jedis.close();}
}

2. 使用 Lettuce 客户端 (推荐)

Lettuce 是另一个流行的 Redis Java 客户端，支持响应式编程。

添加依赖

<dependency><groupId>io.lettuce</groupId><artifactId>lettuce-core</artifactId><version>6.2.3.RELEASE</version>
</dependency>

订阅示例

import io.lettuce.core.RedisClient;
import io.lettuce.core.pubsub.RedisPubSubListener;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;public class LettuceSubscriber {public static void main(String[] args) {RedisClient client = RedisClient.create("redis://localhost");StatefulRedisPubSubConnection<String, String> connection = client.connectPubSub();connection.addListener(new RedisPubSubListener<String, String>() {@Overridepublic void message(String channel, String message) {System.out.println("收到消息 - 频道: " + channel + ", 内容: " + message);}@Overridepublic void message(String pattern, String channel, String message) {// 模式匹配的消息}@Overridepublic void subscribed(String channel, long count) {System.out.println("订阅成功 - 频道: " + channel);}// 其他需要实现的方法...});RedisPubSubCommands<String, String> sync = connection.sync();sync.subscribe("myChannel");// 保持程序运行以持续接收消息try {Thread.sleep(Long.MAX_VALUE);} catch (InterruptedException e) {e.printStackTrace();}connection.close();client.shutdown();}
}

3. Spring Data Redis 集成

如果你使用 Spring Boot，可以更方便地集成 Redis Pub/Sub，这也是比较常用的方式

添加依赖

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

配置 Redis 容器

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;@Configuration
public class RedisConfig {@Beanpublic RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,RedisMessageSubscriber subscriber) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.addMessageListener(subscriber, new ChannelTopic("myChannel"));return container;}
}

配置订阅

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;@Component
public class RedisMessageSubscriber implements MessageListener {@Overridepublic void onMessage(Message message, byte[] pattern) {System.out.println("收到消息: " + new String(message.getBody()));}
}

发布消息

import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;@Service
public class RedisMessagePublisher {private final RedisTemplate<String, Object> redisTemplate;public RedisMessagePublisher(RedisTemplate<String, Object> redisTemplate) {this.redisTemplate = redisTemplate;}public void publish(String message) {redisTemplate.convertAndSend("myChannel", message);}
}

高级功能

模式匹配订阅：可以使用 psubscribe 订阅匹配模式的频道

取消订阅

处理连接断开：实现 onPMessage, onPUnsubscribe 等方法处理各种事件

//模式匹配订阅
// Jedis
jedis.psubscribe(subscriber, "news.*");// Lettuce
sync.psubscribe("news.*");//取消订阅
subscriber.unsubscribe("myChannel");
subscriber.punsubscribe("news.*");