当前位置：首页 > news >正文

php 实现mqtt 订阅监听功能出错_Redis发布订阅原理以及java实现

news 来源：原创 2024/5/6 19:37:52

一、Redis发布订阅原理

Redis的架构包括两个部分：Redis Client和Redis Server，即客户端和服务端。客户端负责向服务器端发送请求并接受来自服务器端的响应。服务器端负责处理客户端请求，例如：数据存储，获取数据，修改数据等。

Redis通常用作数据库，缓存以及消息系统。

发布订阅的框架

其中Publisher和Subscriber为Redis Client，channel为Redis server,而且发布者和订阅者是一对多的关系。

1、发布原理：

客户端和服务端可以理解为都各自维护着一个channel列表。

（1）PUBLISH

当客户端向某个频道发送消息时，Redis首先在结构体redisServer中的pubsub_channels中找出键为该频道的结点，遍历该结点的值，找出所有的客户端，将消息发送给这些客户端。然后，遍历结构体Redis Client中的pubsub_patterns，找出包含该频道的模式的结点，将消息发送给订阅了该模式的客户端。

（2）SUBSCRIBE

首先，在客户端结构体client中，有一个属性为pubsub_channels，该属性表明了该客户端订阅的所有频道，它是一个字典类型，通过哈希表实现，其中的每个元素都包含了一个键值对以及指向下一个元素的指针，每次订阅都要向其中插入一个结点，键表示订阅的频道，值为空。然后，在表示服务器端的结构体redisServer中，也有一个属性为pubsub_channels，但此处它表示的是该服务器端中的所有频道以及订阅了这个频道的客户端，它也是一个字典类型，插入结点时，键表示频道，值则是订阅了这个频道的所有客户端组成的链表。最后Redis通知客户端其订阅成功。

(3)PSUBSCRIBE

当客户端订阅某个模式时，Redis同样需要将该模式和该客户端绑定。首先，在结构体client中，有一个属性为pubsub_patterns，该属性表示该客户端订阅的所有模式，它是一个链表类型，每个结点包括了订阅的模式和指向下一个结点的指针，每次订阅某个模式时，都要向其中插入一个结点。然后，在结构体redisServer中，有一个属性也叫pubsub_patterns，它表示了该服务器端中的所有模式和订阅了这些模式的客户端，它也是一个链表类型，插入结点时，每个结点都要包含订阅的模式，以及订阅这个模式的客户端，和指向下一个结点的指针,个人理解为是一种模糊匹配

（4）UNSUBSCRIBE（退订）

UNSUBSCRIBE命令可以退订指定的频道，这个命令执行的是订阅的反操作：它从 pubsub_channels 字典的给定频道（键）中，删除关于当前客户端的信息，这样被退订频道的信息就不会再发送给这个客户端。

二、REDIS发布订阅和监听REDIS队列的区别

使用jedis的subscribe和publish实现的发布订阅系统 PK 使用jedis的BRPOP和BLPOP实现的阻塞时消息队列

1、redis队列为阻塞队列，获取完一个信息后会主动退出，如果想一直获取信息则需要开启一个监听；而发布订阅中的订阅端是自动完成的监听。

2、redis队列中的数据取出后就消失了，无法满足多端口；而发布订阅可以将数据发布到多个channel。

3、redis队列的数据不取出就会一直在缓存中；而发布订阅中订阅获取的数据不处理就消失了。

三、ActiveMQ和REDIS发布订阅的比较

1、ActiveMQ支持多种消息协议，包括AMQP，MQTT，Stomp等（ActiveMQ支持的传输协议 - winner_0715 - 博客园

2、ActiveMQ提供持久化功能；Redis消息被发送，如果没有订阅者接收，那么消息就会丢失

3、ActiveMQ提供了消息传输保障，当客户端连接超时或事务回滚等情况发生时，消息会被重新发送给客户端；Redis没有提供消息传输保障

4、ActiveMQ所提供的功能远比Redis发布订阅要复杂，毕竟Redis不是专门做发布订阅的；如果系统能够通过mq实现，则没必要使用mq

四、REDIS在客户端的操作

PUBLISH 和 subscribe

publish 和 PSUBSCRIBE

五、JAVA的实现

MAVEN依赖

 <!-- redis.clients " jedis -->
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>3.1.0</version>
        </dependency>

PUBLISH

public Long publish(){
    System.out.println("发布者 ");
    Long res = null;
    try {
        Jedis jr = new Jedis("127.0.0.1", 6379, 0);// redis服务地址和端口号
        // jr客户端配置监听两个channel
        new Thread(()->{
            try {
                Thread.sleep(20000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            jr.publish("mychannel", "test");
        }).start();
    } catch (Exception e) {
        e.printStackTrace();
    }
    return res;
}

实现下面抽象类即可

public abstract class SubscriberService extends JedisPubSub {
 
    public abstract String getChannel();
 
    public abstract boolean dealMessage(String message);
 
    public void onMessage(String channel, String message) {
        dealMessage(message);
    }
 
    @PostConstruct
    public void init(){
        System.out.println("订阅者 ");
        Jedis jr = null;
        try {
            jr = new Jedis("127.0.0.1", 6379, 0);// redis服务地址和端口号
            // jr客户端配置监听两个channel
            jr.subscribe(this, "mychannel");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (jr != null) {
                jr.disconnect();
            }
        }
    }
}

实现例子：

public class SubScribeRedis extends SubscriberService {
    @Override
    public String getChannel() {
        return "mychannel";
    }
 
    @Override
    public boolean dealMessage(String message) {
        System.out.println("--------"+message);
        return false;
    }
}

若有问题，下方留言~

---------------------

原文链接：https://blog.csdn.net/Theo9/article/details/98869532

作者：轻易科技-轻易贷研发部赵少超