91 lines
3.5 KiB
Java
91 lines
3.5 KiB
Java
package com.webgame.webgamebackend.common.utils;
|
|
|
|
import com.alibaba.fastjson2.JSON;
|
|
import io.micrometer.common.util.StringUtils;
|
|
import lombok.RequiredArgsConstructor;
|
|
import lombok.extern.slf4j.Slf4j;
|
|
import org.springframework.data.redis.connection.MessageListener;
|
|
import org.springframework.data.redis.core.RedisTemplate;
|
|
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 java.util.function.Consumer;
|
|
|
|
/**
|
|
* Redis消息队列工具类
|
|
* @author xzg
|
|
*/
|
|
@Slf4j
|
|
@Component
|
|
@RequiredArgsConstructor
|
|
public class RedisMqUtil {
|
|
private final RedisTemplate<String, String> redisTemplate;
|
|
private final RedisMessageListenerContainer redisMessageListenerContainer;
|
|
|
|
public void pub(String channel, Object message) {
|
|
try {
|
|
if (StringUtils.isBlank(channel)) {
|
|
log.warn("[REDIS消息队列] 发布信息失败: 通道名称为 null");
|
|
return;
|
|
}
|
|
if (message == null) {
|
|
log.warn(STR."[REDIS消息队列] 发布信息至通道(\{channel})失败: 信息体为 null");
|
|
return;
|
|
}
|
|
|
|
// 信息体转JSON
|
|
var json = "";
|
|
try {
|
|
json = JSON.toJSONString(message);
|
|
} catch (Exception e) {
|
|
log.warn(STR."[REDIS消息队列] 发布信息至通道(\{channel})异常: 信息实体无法正确的序列化为JSON字符串, \{e.getMessage()}", e);
|
|
return;
|
|
}
|
|
|
|
redisTemplate.convertAndSend(channel, json);
|
|
} catch (Exception e) {
|
|
log.warn(STR."[REDIS消息队列] 发布信息至通道(\{channel})异常: \{e.getMessage()}", e);
|
|
}
|
|
}
|
|
|
|
public <T> void sub(String channel, Class<T> clazz, Consumer<T> callback) {
|
|
if (StringUtils.isBlank(channel)) {
|
|
log.warn("[REDIS消息队列] 订阅通道失败, 通道名称不可为空");
|
|
return;
|
|
}
|
|
if (callback == null) {
|
|
log.warn(STR."[REDIS消息队列] 订阅通道(\{channel})失败, 回调方法不可为空");
|
|
return;
|
|
}
|
|
|
|
try {
|
|
var adapter = new MessageListenerAdapter((MessageListener) (message, _) -> {
|
|
var body = message.getBody();
|
|
var channelName = new String(message.getChannel());
|
|
|
|
T dto;
|
|
try {
|
|
dto = JSON.parseObject(new String(body), clazz);
|
|
} catch (Exception e) {
|
|
log.error(STR."[REDIS消息队列] 消费通道(\{channelName})信息实体类型转换处理异常: \{e.getMessage()}, 信息: \{body}", e);
|
|
return;
|
|
}
|
|
try {
|
|
callback.accept(dto);
|
|
} catch (Exception e) {
|
|
log.error(STR."[REDIS消息队列] 消费通道(\{channelName})信息回调处理异常: \{e.getMessage()}, 信息: \{body}", e);
|
|
}
|
|
});
|
|
|
|
redisMessageListenerContainer.addMessageListener(adapter, new PatternTopic(channel));
|
|
} catch (Exception e) {
|
|
log.error(STR."[REDIS消息队列] 订阅通道(\{channel})异常: \{e.getMessage()}", e);
|
|
return;
|
|
}
|
|
|
|
log.info(STR."[REDIS消息队列] 订阅通道(\{channel})成功");
|
|
}
|
|
}
|