爱站云网站建设heikw,网站建设 李奥贝纳,做网站用哪个服务器不用备案,opencart网站建设Redis 的发布订阅#xff08;pub/sub#xff09;机制是一种消息传递模式#xff0c;允许消息的发送者#xff08;发布者#xff09;和消息的接收者#xff08;订阅者#xff09;通过一个中介层#xff08;频道#xff09;进行通信#xff0c;而无需彼此直接交互。以下…Redis 的发布订阅pub/sub机制是一种消息传递模式允许消息的发送者发布者和消息的接收者订阅者通过一个中介层频道进行通信而无需彼此直接交互。以下是 Redis 发布订阅机制的工作原理的详细解释 基于事件驱动的架构 Redis 服务器使用一个事件驱动的模型来处理所有的网络通信和客户端请求。这种架构允许 Redis 以非阻塞方式高效地处理多个并发连接。 发布Publish 当一个客户端想要发送消息时它会将消息发布到一个指定的频道channel上。 发布操作是通过 PUBLISH 命令实现的。 例如PUBLISH channel_name message 会将 message 发送到名为 channel_name 的频道。 这里返回的代表有3个监听者 订阅Subscribe 客户端使用 SUBSCRIBE 命令来监听一个或多个频道的消息。 当客户端订阅了一个频道后它会接收到发送到这个频道的所有消息。例如通过执行 SUBSCRIBE channel_name客户端就可以监听 channel_name 频道上的消息。 订阅会返回3个参数 1代表订阅 2订阅频道 3发布者数量
当发布着发布消息后订阅者会被推送如下消息
消息传递 一旦有消息被发布到频道上Redis 服务器会将这个消息分发给所有订阅了该频道的客户端。 这种消息传递方式是多对多的多个发布者可以向同一个频道发送消息而所有订阅该频道的客户端都可以接收到这些消息。非持久化的消息 Redis 发布订阅机制中的消息是非持久化的。这意味着一旦消息被发送它不会被存储在服务器上无法被之后订阅该频道的客户端接收。多路复用和非阻塞 I/O Redis 使用多路复用技术如 epoll、kqueue来同时监听多个网络连接这使得服务器能够同时处理多个发布和订阅操作。 使用非阻塞 I/O 确保单个操作不会阻塞整个服务器从而提高整体性能。
在springboot项目中可以如下 发布
Resource
private RedisTemplate redisTemplate;PostMapping(value /add)
public Result? add(RequestBody SysApplication sysApplication) {sysApplicationService.save(sysApplication);//这里需要用JSONObject.toJSONString转成string序列化否则直接上传对象会带上对象路径信息等redisTemplate.convertAndSend(RedisListenerConstant.APP_CHANNEL, JSONObject.toJSONString(sysApplication));return Result.OK(添加成功);
}订阅者 这里写了2个listener因为想说明可以在RedisMessageConfig 同时监听多个频道
package com.sungrowpower.redis;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sungrowpower.common.system.dto.SysApplicationPermissionDto;
import com.sungrowpower.service.IGoodsSpuService;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;import javax.annotation.Resource;
Configuration
public class RedisMessageAppListener implements MessageListener {Resourceprivate IGoodsSpuService iGoodsSpuService;Overridepublic void onMessage(Message message, byte[] bytes) {// 假设 message 体是一个 JSON 字符串String body new String(message.getBody());// 去除类路径
// String json body.substring(body.indexOf({), body.lastIndexOf(}) 1);String msg (String) JSON.parse(body);SysApplicationPermissionDto app JSONObject.parseObject(msg, SysApplicationPermissionDto.class);System.out.println(smsg);try {//这里可以调用本身的业务逻辑iGoodsSpuService.operateGoodsSpu(app);} catch (Exception e) {throw new RuntimeException();}}
}package com.sungrowpower.redis;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sungrowpower.common.exception.AutoBizException;
import com.sungrowpower.common.system.dto.SysPermission;
import com.sungrowpower.service.IGoodsSpuService;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;import javax.annotation.Resource;Component
public class RedisMessageMenuListener implements MessageListener {Resourceprivate IGoodsSpuService iGoodsSpuService;Overridepublic void onMessage(Message message, byte[] bytes) {// 假设 message 体是一个 JSON 字符串String body new String(message.getBody());// 去除类路径
// String json body.substring(body.indexOf({), body.lastIndexOf(}) 1);String msg (String) JSON.parse(body);SysPermission menu JSONObject.parseObject(msg, SysPermission.class);System.out.println(smsg);try {//这里可以调用本身的业务逻辑iGoodsSpuService.operateAttrValue(menu);} catch (Exception e) {throw new RuntimeException();}}}package com.sungrowpower.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.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
Configuration
public class RedisMessageConfig {/*** 消息监听器适配器menu* return org.springframework.data.redis.listener.adapter.MessageListenerAdapter*/Beanpublic MessageListenerAdapter listenerAdapterMenu(RedisMessageMenuListener receiver) {//这个地方是给 messageListenerAdapter 传入一个消息接受的处理器利用反射的方法调用“onMessage”return new MessageListenerAdapter(receiver, onMessage);}/*** 消息监听器适配器app* return org.springframework.data.redis.listener.adapter.MessageListenerAdapter*/Beanpublic MessageListenerAdapter listenerAdapterApp(RedisMessageAppListener receiver) {//这个地方是给 messageListenerAdapter 传入一个消息接受的处理器利用反射的方法调用“onMessage”return new MessageListenerAdapter(receiver, onMessage);}/*** redis消息监听器容器* return org.springframework.data.redis.listener.RedisMessageListenerContainer*/Beanpublic RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapterMenu,MessageListenerAdapter listenerAdapterApp) {RedisMessageListenerContainer container new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);// 订阅了一个叫 my-channel 的频道container.addMessageListener(listenerAdapterMenu, new PatternTopic(menu_channel));container.addMessageListener(listenerAdapterApp, new PatternTopic(app_channel));// 这个container 可以添加多个 messageListenerreturn container;}}
