济南网站优化分析,在线制作动画的网站,企业简介ppt模板免费,上海门户网站一网通办为了减轻项目的中间件臃肿#xff0c;由于我们项目本身就应用了 Redis#xff0c;正好 Redis 的也具备订阅发布监听的特性#xff0c;正好应对 Etcd 的功能#xff0c;所以本次给大家讲解如何使用 Redis 消息订阅发布来替代 Etcd 的解决方案。接下来#xff0c;我们先看 R… 为了减轻项目的中间件臃肿由于我们项目本身就应用了 Redis正好 Redis 的也具备订阅发布监听的特性正好应对 Etcd 的功能所以本次给大家讲解如何使用 Redis 消息订阅发布来替代 Etcd 的解决方案。接下来我们先看 Redis 订阅发布的常见情景…… Redis 订阅发布公共类
RedisConfig.java
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import java.net.UnknownHostException;Configuration
ComponentScan({cn.hutool.extra.spring})
public class RedisConfig {BeanRedisMessageListenerContainer container (RedisConnectionFactory redisConnectionFactory){RedisMessageListenerContainer container new RedisMessageListenerContainer();container.setConnectionFactory(redisConnectionFactory);return container;}Beanpublic RedisTemplateString, Object redisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {RedisTemplateString, Object template new RedisTemplate();// 连接工厂template.setConnectionFactory(redisConnectionFactory);// 序列化配置Jackson2JsonRedisSerializer objectJackson2JsonRedisSerializer new Jackson2JsonRedisSerializer(Object.class);ObjectMapper objectMapper new ObjectMapper();objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);objectJackson2JsonRedisSerializer.setObjectMapper(objectMapper);StringRedisSerializer stringRedisSerializer new StringRedisSerializer();// 配置具体序列化// key采用string的序列化方式template.setKeySerializer(stringRedisSerializer);// hash的key采用string的序列化方式template.setHashKeySerializer(stringRedisSerializer);// value序列化采用jacksontemplate.setValueSerializer(objectJackson2JsonRedisSerializer);// hash的value序列化采用jacksontemplate.setHashValueSerializer(objectJackson2JsonRedisSerializer);template.afterPropertiesSet();return template;}
}
RedisUtil.java
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;Slf4j
Component
public class RedisUtil {Resourceprivate RedisTemplate redisTemplate;/*** 消息发送* param topic 主题* param message 消息*/public void publish(String topic, String message) {redisTemplate.convertAndSend(topic, message);}
}
application.yml
server:port: 7077
spring:application:name: redis-demoredis:host: localhosttimeout: 3000jedis:pool:max-active: 300max-idle: 100max-wait: 10000port: 6379
RedisController.java
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;/*** author Lux Sun* date 2023/9/12*/
RestController
RequestMapping(/redis)
public class RedisController {Resourceprivate RedisUtil redisUtil;PostMappingpublic String publish(RequestParam String topic, RequestParam String msg) {redisUtil.publish(topic, msg);return 发送成功: topic - msg;}
} 一、业务情景1 个消费者监听 1 个 Topic
教程三步走下文业务情景类似不再描述
实现接口 MessageListener消息订阅绑定业务 Topic重写 onMessage 消费者业务方法
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;Slf4j
Component
public class RedisReceiver1 implements MessageListener {Resourceprivate RedisMessageListenerContainer container;/*** 重点关注这方法, 进行消息订阅*/PostConstructpublic void init() {MessageListenerAdapter adapter new MessageListenerAdapter(this);// 绑定 Topic 语法为正则表达式container.addMessageListener(adapter, new PatternTopic(topic1.*));}Overridepublic void onMessage(Message message, byte[] bytes) {String key new String(message.getChannel());String value new String(message.getBody());log.info(Key: {}, key);log.info(Value: {}, value);}
}
测试
curl --location 127.0.0.1:7077/redis \
--header Content-Type: application/x-www-form-urlencoded \
--data-urlencode topictopic1.msg \
--data-urlencode msg我是消息1
结果
2023-11-15 10:22:38.445 INFO 59189 --- [ container-2] com.xxx.redis.demo.RedisReceiver1 : Key: topic1.msg
2023-11-15 10:22:38.445 INFO 59189 --- [ container-2] com.xxx.redis.demo.RedisReceiver1 : Value: 我是消息1 二、业务情景1 个消费者监听 N 个 Topic
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;Slf4j
Component
public class RedisReceiver1 implements MessageListener {Resourceprivate RedisMessageListenerContainer container;/*** 重点关注这方法, 进行消息订阅*/PostConstructpublic void init() {MessageListenerAdapter adapter new MessageListenerAdapter(this);// 绑定 Topic 语法为正则表达式container.addMessageListener(adapter, new PatternTopic(topic1.*));// 只需再绑定业务 Topic 即可container.addMessageListener(adapter, new PatternTopic(topic2.*));}Overridepublic void onMessage(Message message, byte[] bytes) {String key new String(message.getChannel());String value new String(message.getBody());log.info(Key: {}, key);log.info(Value: {}, value);}
}
测试
curl --location 127.0.0.1:7077/redis \
--header Content-Type: application/x-www-form-urlencoded \
--data-urlencode topictopic2.msg \
--data-urlencode msg我是消息2
结果
2023-11-15 10:22:38.445 INFO 59189 --- [ container-2] com.xxx.redis.demo.RedisReceiver1 : Key: topic2.msg
2023-11-15 10:22:38.445 INFO 59189 --- [ container-2] com.xxx.redis.demo.RedisReceiver1 : Value: 我是消息2 三、业务情景N 个消费者监听 1 个 Topic
我们看一下现在又新增一个 RedisReceiver2按理讲测试的时候RedisReceiver1 和 RedisReceiver2 会同时收到消息
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;Slf4j
Component
public class RedisReceiver2 implements MessageListener {Resourceprivate RedisMessageListenerContainer container;/*** 重点关注这方法, 进行消息订阅*/PostConstructpublic void init() {MessageListenerAdapter adapter new MessageListenerAdapter(this);// 绑定 Topic 语法为正则表达式container.addMessageListener(adapter, new PatternTopic(topic1.*));}Overridepublic void onMessage(Message message, byte[] bytes) {String key new String(message.getChannel());String value new String(message.getBody());log.info(Key: {}, key);log.info(Value: {}, value);}
}
测试
curl --location 127.0.0.1:7077/redis \
--header Content-Type: application/x-www-form-urlencoded \
--data-urlencode topictopic1.msg \
--data-urlencode msg我是消息1
结果
2023-11-15 10:22:38.445 INFO 59189 --- [ container-2] com.xxx.redis.demo.RedisReceiver1 : Key: topic1.msg
2023-11-15 10:22:38.449 INFO 59189 --- [ container-3] com.xxx.redis.demo.RedisReceiver2 : Key: topic1.msg
2023-11-15 10:22:38.545 INFO 59189 --- [ container-2] com.xxx.redis.demo.RedisReceiver1 : Value: 我是消息1
2023-11-15 10:22:38.645 INFO 59189 --- [ container-3] com.xxx.redis.demo.RedisReceiver2 : Value: 我是消息1 四、业务情景N 个消费者监听 N 个 Topic
都到这阶段了应该不难理解了吧~
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;Slf4j
Component
public class RedisReceiver2 implements MessageListener {Resourceprivate RedisMessageListenerContainer container;/*** 重点关注这方法, 进行消息订阅*/PostConstructpublic void init() {MessageListenerAdapter adapter new MessageListenerAdapter(this);// 绑定 Topic 语法为正则表达式container.addMessageListener(adapter, new PatternTopic(topic1.*));// 只需再绑定业务 Topic 即可container.addMessageListener(adapter, new PatternTopic(topic2.*));}Overridepublic void onMessage(Message message, byte[] bytes) {String key new String(message.getChannel());String value new String(message.getBody());log.info(Key: {}, key);log.info(Value: {}, value);}
}
测试
curl --location 127.0.0.1:7077/redis \
--header Content-Type: application/x-www-form-urlencoded \
--data-urlencode topictopic2.msg \
--data-urlencode msg我是消息2
结果
2023-11-15 10:22:38.445 INFO 59189 --- [ container-2] com.xxx.redis.demo.RedisReceiver1 : Key: topic2.msg
2023-11-15 10:22:38.449 INFO 59189 --- [ container-3] com.xxx.redis.demo.RedisReceiver2 : Key: topic2.msg
2023-11-15 10:22:38.545 INFO 59189 --- [ container-2] com.xxx.redis.demo.RedisReceiver1 : Value: 我是消息2
2023-11-15 10:22:38.645 INFO 59189 --- [ container-3] com.xxx.redis.demo.RedisReceiver2 : Value: 我是消息2 好了Redis 订阅发布的教程到此为止。接下来我们看下如何用它来替代 Etcd 的业务情景
这之前我们先大概聊下 Etcd 的 2 个要点
Etcd 消息事件类型Etcd 持久层数据
那么问题来了Redis 虽然具备基本的消息订阅发布但是如何契合 Etcd 的这 2 点特性我们目前给出对应的解决方案是
使用 Redis K-V 的 value 作为 Etcd 消息事件类型使用 MySQL 作为 Etcd 持久层数据字段 id 随机 UUID、字段 key 对应 Etcd key、字段 value 对应 Etcd value这样做的一个好处是无需重构数据结构
SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS 0;DROP TABLE IF EXISTS t_redis_msg;
CREATE TABLE t_redis_msg (
id varchar(32) NOT NULL,
key varchar(255) NOT NULL,
value longtext,
PRIMARY KEY (id)
) ENGINEInnoDB DEFAULT CHARSETutf8mb4;SET FOREIGN_KEY_CHECKS 1;
所以如果想平替 Etcd 的事件类型和持久层数据的解决方案需要 MySQL Redis 结合接下来直接上代码…… Redis MySQL 整合
application.yml升级
spring:application:name: redis-demodatasource:username: rootpassword: 123456url: jdbc:mysql://localhost:3306/db_demo?useUnicodetruecharacterEncodingutf-8useSSLfalseserverTimezoneAsia/Shanghaidriver-class-name: com.mysql.cj.jdbc.Driverhikari:connection-test-query: SELECT 1idle-timeout: 40000max-lifetime: 1880000connection-timeout: 40000minimum-idle: 1validation-timeout: 60000maximum-pool-size: 20
RedisMsg.java
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;/*** author Lux Sun* date 2021/2/19*/
Data
SuperBuilder
NoArgsConstructor
AllArgsConstructor
TableName(value t_redis_msg, autoResultMap true)
public class RedisMsg {TableId(type IdType.ASSIGN_UUID)private String id;TableField(value key)private String key;private String value;
}
RedisMsgEnum.java
/*** author Lux Sun* date 2022/11/11*/
public enum RedisMsgEnum {PUT(PUT),DEL(DEL);private String code;RedisMsgEnum(String code) {this.code code;}public String getCode() {return code;}}
RedisMsgService.java
import com.baomidou.mybatisplus.extension.service.IService;
import java.util.List;
import java.util.Map;/*** author Lux Sun* date 2020/6/16*/
public interface RedisMsgService extends IServiceRedisMsg {/*** 获取消息* param key*/RedisMsg get(String key);/*** 获取消息列表* param key*/MapString, String map(String key);/*** 获取消息值* param key*/String getValue(String key);/*** 获取消息列表* param key*/ListRedisMsg list(String key);/*** 插入消息* param key* param value*/void put(String key, String value);/*** 删除消息* param key*/void del(String key);
}
RedisMsgServiceImpl.java
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;/*** author Lux Sun* date 2020/6/16*/
Slf4j
Service
public class RedisMsgServiceImpl extends ServiceImplRedisMsgDao, RedisMsg implements RedisMsgService {Resourceprivate RedisMsgDao redisMsgDao;Resourceprivate RedisUtil redisUtil;/*** 获取消息** param key*/Overridepublic RedisMsg get(String key) {LambdaQueryWrapperRedisMsg lqw new LambdaQueryWrapper();lqw.eq(RedisMsg::getKey, key);return redisMsgDao.selectOne(lqw);}/*** 获取消息列表** param key*/Overridepublic MapString, String map(String key) {ListRedisMsg redisMsgs this.list(key);return redisMsgs.stream().collect(Collectors.toMap(RedisMsg::getKey, RedisMsg::getValue));}/*** 获取消息值** param key*/Overridepublic String getValue(String key) {RedisMsg redisMsg this.get(key);return redisMsg.getValue();}/*** 获取消息列表** param key*/Overridepublic ListRedisMsg list(String key) {LambdaQueryWrapperRedisMsg lqw new LambdaQueryWrapper();lqw.likeRight(RedisMsg::getKey, key);return redisMsgDao.selectList(lqw);}/*** 插入消息** param key* param value*/Overridepublic void put(String key, String value) {log.info(开始添加 - key: {}value: {}, key, value);LambdaQueryWrapperRedisMsg lqw new LambdaQueryWrapper();lqw.eq(RedisMsg::getKey, key);this.saveOrUpdate(RedisMsg.builder().key(key).value(value).build(), lqw);redisUtil.putMsg(key);log.info(添加成功 - key: {}value: {}, key, value);}/*** 删除消息** param key*/Overridepublic void del(String key) {log.info(开始删除 - key: {}, key);LambdaQueryWrapperRedisMsg lqw new LambdaQueryWrapper();lqw.likeRight(RedisMsg::getKey, key);redisMsgDao.delete(lqw);redisUtil.delMsg(key);log.info(删除成功 - key: {}, key);}
}
RedisUtil.java升级
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;Slf4j
Component
public class RedisUtil {Resourceprivate RedisTemplate redisTemplate;/*** 消息发送* param topic 主题* param message 消息*/public void publish(String topic, String message) {redisTemplate.convertAndSend(topic, message);}/*** 消息发送 PUT* param topic 主题*/public void putMsg(String topic) {redisTemplate.convertAndSend(topic, RedisMsgEnum.PUT);}/*** 消息发送 DELETE* param topic 主题*/public void delMsg(String topic) {redisTemplate.convertAndSend(topic, RedisMsgEnum.DEL);}
} 演示 DEMO
RedisMsgController.java
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;/*** author Lux Sun* date 2023/9/12*/
RestController
RequestMapping(/redisMsg)
public class RedisMsgController {Resourceprivate RedisMsgService redisMsgService;PostMappingpublic String publish(RequestParam String topic, RequestParam String msg) {redisMsgService.put(topic, msg);return 发送成功: topic - msg;}
}
RedisMsgReceiver.java
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;Slf4j
Component
public class RedisMsgReceiver implements MessageListener {Resourceprivate RedisMsgService redisMsgService;Resourceprivate RedisMessageListenerContainer container;PostConstructpublic void init() {MessageListenerAdapter adapter new MessageListenerAdapter(this);container.addMessageListener(adapter, new PatternTopic(topic3.*));}Overridepublic void onMessage(Message message, byte[] bytes) {String key new String(message.getChannel());String event new String(message.getBody());String value redisMsgService.getValue(key);log.info(Key: {}, key);log.info(Event: {}, event);log.info(Value: {}, value);}
}
测试
curl --location 127.0.0.1:7077/redisMsg \
--header Content-Type: application/x-www-form-urlencoded \
--data-urlencode topictopic3.msg \
--data-urlencode msg我是消息3
结果
2023-11-16 10:24:35.721 INFO 43794 --- [nio-7077-exec-1] c.c.redis.demo.RedisMsgServiceImpl : 开始添加 - key: topic3.msgvalue: 我是消息3
2023-11-16 10:24:35.935 INFO 43794 --- [nio-7077-exec-1] c.c.redis.demo.RedisMsgServiceImpl : 添加成功 - key: topic3.msgvalue: 我是消息3
2023-11-16 10:24:35.950 INFO 43794 --- [ container-2] c.xxx.redis.demo.RedisMsgReceiver : Key: topic3.msg
2023-11-16 10:24:35.950 INFO 43794 --- [ container-2] c.xxx.redis.demo.RedisMsgReceiver : Event: PUT
2023-11-16 10:24:35.950 INFO 43794 --- [ container-2] c.xxx.redis.demo.RedisMsgReceiver : Value: 我是消息3
