Redisson发布订阅学习

介绍

Redisson 的消息订阅功能遵循 Redis 的发布/订阅模式，该模式包括以下几个核心概念：

• 发布者（Publisher）：发送消息到特定频道的客户端。在 Redis 中，这通过 PUBLISH 命令实现。

• 订阅者（Subscriber）：监听频道并接收消息的客户端。Redis 提供了 SUBSCRIBE 命令来实现订阅功能。

• 频道（Channel）：消息传递的媒介。发布者和订阅者通过频道进行通信。

• 消息（Message）：通过频道从发布者传递到订阅者的数据。

Redisson 对 Redis 的发布/订阅机制进行了封装，提供了更易于使用的 Java API。具体实现如下：

• 发布消息（Publish）：当一个客户端想要发送消息时，它使用 Redis 的 PUBLISH 命令将消息发送到特定的频道。Redisson 提供了 RTopic 接口来实现这一功能。例如：

RTopic topic = redisson.getTopic("testTopic");
topic.publish("Hello, World!");

• 订阅消息（Subscribe）：另一端的客户端使用 Redis 的 SUBSCRIBE 命令订阅一个或多个频道。当有消息发送到这些频道时，Redis 会自动将消息推送给所有订阅者。Redisson 提供了 addListener 方法来添加消息监听器。例如：

topic.addListener(String.class, new MessageListener<String>() {
    @Override
    public void onMessage(CharSequence channel, String msg) {
        System.out.println("Received message from " + channel + ": " + msg);
    }
});

 消息处理

Redisson 为 RTopic 实现了监听器接口，允许应用程序定义如何处理接收到的消息。当消息到达时，Redisson 会调用这些监听器。例如，可以在 onMessage 方法中处理消息，或者在处理过程中捕获异常并进行适当的兜底策略。

优点

• 解耦：发布者和订阅者之间不需要知道对方，可以独立进行扩展和修改。

• 简单易用：Redisson 提供了简单的 API 来进行消息的发布和订阅，开发者可以轻松集成到自己的应用中。

• 实时性：Redis 的发布/订阅机制提供了低延迟的消息传递，适合需要实时通信的应用。

• 可扩展性：可以有多个订阅者同时订阅同一个主题，而且可以通过增加 Redis 实例来水平扩展系统。

• 高吞吐量：Redis 作为内存数据结构存储，能够处理大量的消息。

 缺点

• 消息丢失：

  Redis 的 Pub/Sub 是即时的，如果订阅者不在线，消息会丢失。如果需要消息持久化，可以考虑使用 Redis Stream。

• 网络断开：

  如果客户端与 Redis 服务器的连接断开，需要重新订阅频道。

• 性能问题：

  如果订阅的频道过多或消息量过大，可能会影响 Redis 服务器的性能。

代码示例

添加依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.15.6</version> 
</dependency>

初始化 Redisson 客户端

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

public class RedissonConfig {

    public static RedissonClient createRedissonClient() {

        Config config = new Config();
        config.useSingleServer()
                 //Redis 服务器地址
                .setAddress("redis://127.0.0.1:6379")
                //密码
                .setPassword("password") 
                .setKeepAlive(true);
        return Redisson.create(config);
    }

}

创建消息发布者

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.api.listener.MessageListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@AllArgsConstructor
public class MessagePublisher {

    private RedissonClient redissonClient;

    public void publishMessage(String topicName, String message) {

        RTopic topic = redissonClient.getTopic(topicName);

        topic.publish(message);
    }
}

创建消息订阅者

import lombok.AllArgsConstructor;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

@Component
@AllArgsConstructor
public class DemoMessagePublisher{

	private RedissonClient redissonClient;

	public void publishMessage(String topicName, String message) {

		RTopic topic = redissonClient.getTopic(topicName);

		topic.publish(message);
	}

}

消息绑定

import com.jlcloud.tenant.mq.listener.TenantMessageListener;
import lombok.AllArgsConstructor;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

@Component
public class MessageTestRunner implements CommandLineRunner {

	private DemoMessageListener messagePublisher;

	@Override
	public void run(String... args) throws Exception {
		// 订阅主题
		messagePublisher.listener("test_update");
	}
}

创建发消息

import com.chengxuyuanshitang.mq.publisher.DemoMessagePublisher;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.AllArgsConstructor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@AllArgsConstructor
@RequestMapping("/front/test")
@Tag(name = "TestController", description = "TestController")
public class TestController {

	private final DemoMessagePublisher mssagePublisher;

	@GetMapping("/msg/send")
	@Operation(summary = "消息发送")
	public Boolean sendMsg() {
		mssagePublisher.publishMessage("test_update", "== test a msg==");
		return Boolean.TRUE;
	}
}