为了减轻项目的中间件臃肿，由于我们项目本身就应用了 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 {
@Bean
RedisMessageListenerContainer container (RedisConnectionFactory redisConnectionFactory){
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(redisConnectionFactory);
return container;
}
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {
RedisTemplate<String, 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序列化采用jackson
template.setValueSerializer(objectJackson2JsonRedisSerializer);
// hash的value序列化采用jackson
template.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 {
@Resource
private 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-demo
redis:
host: localhost
timeout: 3000
jedis:
pool:
max-active: 300
max-idle: 100
max-wait: 10000
port: 6379RedisController.java
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;
/**
* @author Lux Sun
* @date 2023/9/12
*/
@RestController
@RequestMapping("/redis")
public class RedisController {
@Resource
private RedisUtil redisUtil;
@PostMapping
public 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 {
@Resource
private RedisMessageListenerContainer container;
/**
* 重点关注这方法, 进行消息订阅
*/
@PostConstruct
public void init() {
MessageListenerAdapter adapter = new MessageListenerAdapter(this);
// 绑定 Topic 语法为正则表达式
container.addMessageListener(adapter, new PatternTopic("topic1.*"));
}
@Override
public 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 'topic=topic1.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 {
@Resource
private RedisMessageListenerContainer container;
/**
* 重点关注这方法, 进行消息订阅
*/
@PostConstruct
public void init() {
MessageListenerAdapter adapter = new MessageListenerAdapter(this);
// 绑定 Topic 语法为正则表达式
container.addMessageListener(adapter, new PatternTopic("topic1.*"));
// 只需再绑定业务 Topic 即可
container.addMessageListener(adapter, new PatternTopic("topic2.*"));
}
@Override
public 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 'topic=topic2.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 {
@Resource
private RedisMessageListenerContainer container;
/**
* 重点关注这方法, 进行消息订阅
*/
@PostConstruct
public void init() {
MessageListenerAdapter adapter = new MessageListenerAdapter(this);
// 绑定 Topic 语法为正则表达式
container.addMessageListener(adapter, new PatternTopic("topic1.*"));
}
@Override
public 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 'topic=topic1.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 {
@Resource
private RedisMessageListenerContainer container;
/**
* 重点关注这方法, 进行消息订阅
*/
@PostConstruct
public void init() {
MessageListenerAdapter adapter = new MessageListenerAdapter(this);
// 绑定 Topic 语法为正则表达式
container.addMessageListener(adapter, new PatternTopic("topic1.*"));
// 只需再绑定业务 Topic 即可
container.addMessageListener(adapter, new PatternTopic("topic2.*"));
}
@Override
public 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 'topic=topic2.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`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
SET FOREIGN_KEY_CHECKS = 1;所以，如果想平替 Etcd 的事件类型和持久层数据的解决方案需要 MySQL & Redis 结合，接下来直接上代码……
Redis & MySQL 整合
application.yml（升级）
spring:
application:
name: redis-demo
datasource:
username: root
password: 123456
url: jdbc:mysql://localhost:3306/db_demo?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
driver-class-name: com.mysql.cj.jdbc.Driver
hikari:
connection-test-query: SELECT 1
idle-timeout: 40000
max-lifetime: 1880000
connection-timeout: 40000
minimum-idle: 1
validation-timeout: 60000
maximum-pool-size: 20RedisMsg.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 IService<RedisMsg> {
/**
* 获取消息
* @param key
*/
RedisMsg get(String key);
/**
* 获取消息列表
* @param key
*/
Map<String, String> map(String key);
/**
* 获取消息值
* @param key
*/
String getValue(String key);
/**
* 获取消息列表
* @param key
*/
List<RedisMsg> 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 ServiceImpl<RedisMsgDao, RedisMsg> implements RedisMsgService {
@Resource
private RedisMsgDao redisMsgDao;
@Resource
private RedisUtil redisUtil;
/**
* 获取消息
*
* @param key
*/
@Override
public RedisMsg get(String key) {
LambdaQueryWrapper<RedisMsg> lqw = new LambdaQueryWrapper<>();
lqw.eq(RedisMsg::getKey, key);
return redisMsgDao.selectOne(lqw);
}
/**
* 获取消息列表
*
* @param key
*/
@Override
public Map<String, String> map(String key) {
List<RedisMsg> redisMsgs = this.list(key);
return redisMsgs.stream().collect(Collectors.toMap(RedisMsg::getKey, RedisMsg::getValue));
}
/**
* 获取消息值
*
* @param key
*/
@Override
public String getValue(String key) {
RedisMsg redisMsg = this.get(key);
return redisMsg.getValue();
}
/**
* 获取消息列表
*
* @param key
*/
@Override
public List<RedisMsg> list(String key) {
LambdaQueryWrapper<RedisMsg> lqw = new LambdaQueryWrapper<>();
lqw.likeRight(RedisMsg::getKey, key);
return redisMsgDao.selectList(lqw);
}
/**
* 插入消息
*
* @param key
* @param value
*/
@Override
public void put(String key, String value) {
log.info("开始添加 - key: {}，value: {}", key, value);
LambdaQueryWrapper<RedisMsg> 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
*/
@Override
public void del(String key) {
log.info("开始删除 - key: {}", key);
LambdaQueryWrapper<RedisMsg> 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 {
@Resource
private 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 {
@Resource
private RedisMsgService redisMsgService;
@PostMapping
public 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 {
@Resource
private RedisMsgService redisMsgService;
@Resource
private RedisMessageListenerContainer container;
@PostConstruct
public void init() {
MessageListenerAdapter adapter = new MessageListenerAdapter(this);
container.addMessageListener(adapter, new PatternTopic("topic3.*"));
}
@Override
public 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 'topic=topic3.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.msg，value: 我是消息3
2023-11-16 10:24:35.935 INFO 43794 --- [nio-7077-exec-1] c.c.redis.demo.RedisMsgServiceImpl : 添加成功 - key: topic3.msg，value: 我是消息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
