写在开头
写这篇教程的目的是为了给那些有毕设需求而准备的，实际开放中应该没多少人会把mirai拉进微服务这个圈子吧？
本篇中，Mirai Core充当着用户终端的角色，把用户输入传出、系统信息传送至用户；对分布式环境做了初步适配；结合Nacos进行动态更新配置，并利用Sentinel进行端口的动态流控；使用RocketMQ；
项目默认使用WebFlux、但Servlet我也做了支持，branch中带有-servlet则是完全采用Servlet。
各层必要方法的逻辑我会详细说明。
项目地址
传送门
PS：实际业务逻辑请以项目为准，文档有时候来不及更新！
版本预览
如果最新版代码中有对上版示例代码的进行大量变更时，将在最新版中附上完整的示例代码，否则只会针对需要更新的业务方法进行详细说明。
若无必要，将不会删除各版间的留言！
第一版：传送门
PS：代码我并没有实机测试（特指集群环境），若有BUG，望不吝赐教！
第二版：传送门
PS： 正在更新，目前已更新了Mysql版，MongoDB正在调试中。项目中的一些代码我事后回想起来是有坑的，正在结合源码进行填坑…… 更新完毕，暂时解决了消息ID自定义，不过RocketMQ Spring Boot Starter异步回调确实不完美，实际使用还需注意。
目录
主要组件版本控制
其他相关组件版本控制
主POM
Bot
4.1
多Bot的坑
4.2
多Bot的坑（集群）
4.3
其他业务对多Bot的适配
Nacos
5.1
整合Nacos
5.2
接入配置中心
5.3
整合Nacos
5.4
设置配置文件
5.5
监听Nacos配置刷新事件
Sentinel
6.1
整合Sentinel
6.2
配置Sentinel客户端
6.3
自定义流控返回
6.4
其他类对Sentinel的适配
6.5
配置持久化
配置参考
7.1
Redis配置参考
7.2
Redisson配置参考
主要组件版本控制：
Adopt OpenJDK
Spring Boot
Spring Cloud
Spring Cloud Alibaba
Mirai Core
11.0.10
2.4.2
2020.0.1
2021.1
2.6.8
其他相关组件版本控制
Nacos
Sentinel
Hutool
1.4.1
1.8.0
5.7.15
主POM：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.4.2</version>
<relativePath/>
</parent>
<groupId>cyou.wssy001.cloud</groupId>
<artifactId>bot</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>bot</name>
<packaging>pom</packaging>
<properties>
<java.version>11</java.version>
<spring.cloud.version>2020.0.1</spring.cloud.version>
<spring.cloud.alibaba.version>2021.1</spring.cloud.alibaba.version>
<lombok.version>1.18.20</lombok.version>
<redisson.version>3.16.4</redisson.version>
<hutool.version>5.7.15</hutool.version>
<fastjson.version>1.2.78</fastjson.version>
<commons.pool2.version>2.11.1</commons.pool2.version>
<mirai.core.version>2.6.8</mirai.core.version>
</properties>
<dependencyManagement>
<dependencies>
<!-- Spring Cloud Alibaba-->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-alibaba-dependencies</artifactId>
<version>${spring.cloud.alibaba.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<!-- Spring Cloud-->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring.cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<!-- Lombok-->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>${lombok.version}</version>
</dependency>
<!-- Redisson-->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>${redisson.version}</version>
</dependency>
<!-- Hutool-->
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-core</artifactId>
<version>${hutool.version}</version>
</dependency>
<!-- Hutool-http-->
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-http</artifactId>
<version>${hutool.version}</version>
</dependency>
<!-- FastJSON-->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>${fastjson.version}</version>
</dependency>
<!-- Commons Pool2-->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<version>${commons.pool2.version}</version>
</dependency>
<!-- Mirai-->
<dependency>
<groupId>net.mamoe</groupId>
<artifactId>mirai-core-jvm</artifactId>
<version>${mirai.core.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
</project>
Bot
引入Mirai依赖，由于Spring Boot 2.4.2自带的kotlin依赖版本是
<kotlin.version>1.4.21</kotlin.version>
<kotlin-coroutines.version>1.4.2</kotlin-coroutines.version>
所以可以直接引入mirai-core-jvm，无需再指定kotlin相关依赖的版本
<dependency>
<groupId>net.mamoe</groupId>
<artifactId>mirai-core-jvm</artifactId>
</dependency>
定义一个一个存储Bot账户的类
@Data
public class BotAccount implements Serializable {
private static final long serialVersionUID = 1L;
private Long account;
private String password;
}
然后有一个用户机器人管理的Service，Bot登录后可以不放入IOC，使用Bot.getInstanceOrNull(botQQNumber)获得指定的Bot实例。
@Slf4j
@Service
@RequiredArgsConstructor
public class RobotService {
private final DynamicProperty dynamicProperty;
private final BotOnlineHandler botOnlineHandler;
private final BotOfflineHandler botOfflineHandler;
private final GroupMessageHandler groupMessageHandler;
private final MemberJoinHandler memberJoinHandler;
private final MemberLeaveHandler memberLeaveHandler;
@PostConstruct
private void init() {
login(stringToBotAccountList(dynamicProperty.getAccounts()));
}
public void refreshBot() {
List<BotAccount> botAccounts = stringToBotAccountList(dynamicProperty.getAccounts())
.stream()
.filter(v -> Bot.getInstanceOrNull(v.getAccount()) == null)
.collect(Collectors.toList());
login(botAccounts);
}
private void login(List<BotAccount> botAccounts) {
BotConfiguration botConfiguration = new BotConfiguration();
botConfiguration.fileBasedDeviceInfo("device.json");
// botConfiguration.noNetworkLog();
// botConfiguration.noBotLog();
for (BotAccount botAccountInfo : botAccounts) {
Bot bot = BotFactory.INSTANCE.newBot(botAccountInfo.getAccount(), botAccountInfo.getPassword(), botConfiguration);
try {
bot.getEventChannel().subscribeAlways(BotOnlineEvent.class, botOnlineHandler::handle);
bot.getEventChannel().subscribeAlways(BotOfflineEvent.class, botOfflineHandler::handle);
bot.getEventChannel().subscribeAlways(GroupMessageEvent.class, groupMessageHandler::handle);
bot.getEventChannel().subscribeAlways(MemberJoinEvent.class, memberJoinHandler::handle);
bot.getEventChannel().subscribeAlways(MemberLeaveEvent.class, memberLeaveHandler::handle);
bot.login();
log.info("******RobotService QQ：{} 登陆成功", botAccountInfo.getAccount());
// 去掉break即可实现多Bot在线
break;
} catch (LoginFailedException e) {
bot.close();
log.error("******Bot LoginFailedException：{} ，QQ：{}", e.getMessage(), botAccountInfo.getAccount());
}
}
}
private List<BotAccount> stringToBotAccountList(String s) {
return JSON.parseArray(s, BotAccount.class);
}
}
多Bot的坑
当你的项目同时启用多个Bot时，可能会遇到同时监听一个群消息的问题，这时我们需要加锁，在监听群消息上强行“单线程”。由于只部署一个实例，所以用JVM级别的锁就行。
逻辑
使用synchronized可以确保同一时间只有一个机器人能够调用该方法，同时检查消息ID是否存在于Redis来避免重复消费（这块我使用Redis的有序集合，通过设置scope为过期时间来实现过期，同时利用有序集合对存储已存在数据会执行更新操作且返回0的性质完成对消息IDs的存储和是否全部存在的判断），同时做了优化，只有存在多个机器人同时监听的群才会被强制“单线程”
@Component
@RequiredArgsConstructor
public class GroupMessageHandler2 {
private final RepetitiveGroupService repetitiveGroupService;
@Resource
private ReactiveZSetOperations<Long, Integer> reactiveZSetOperations;
public void handle(@NotNull GroupMessageEvent event) {
long groupId = event.getGroup().getId();
long botId = event.getBot().getId();
RepetitiveGroup repetitiveGroup = repetitiveGroupService.get(groupId)
.share()
.block();
if (hasBeenConsumed(event, groupId)) return;
if (containMultiBots(repetitiveGroup, botId)) {
synchronized (this) {
consume();
}
} else {
consume();
}
}
// 检查事件是否以及被消费过
private boolean hasBeenConsumed(@NotNull GroupMessageEvent event, long groupId) {
long second = new Date().getTime();
Range<Double> range = Range.closed(0.0, second + 0.0);
reactiveZSetOperations.removeRangeByScore(groupId, range);
int[] ids = event.getSource().getIds();
List<DefaultTypedTuple<Integer>> list = new ArrayList<>();
for (int id : ids) {
list.add(new DefaultTypedTuple<>(id, second + 10.0));
}
Long unExist = reactiveZSetOperations.addAll(groupId, list)
.share()
.block();
return unExist == null || unExist == 0;
}
// 检查这个群是否有多个机器人监听群消息
private boolean containMultiBots(RepetitiveGroup repetitiveGroup, Long botId) {
return repetitiveGroup.getId() == null || !repetitiveGroup.getBotIds().contains(botId);
}
// 消费事件
private void consume() {
}
}
多Bot的坑（集群）
集群环境下，JVM级别的锁失效，考虑到接入了Redis，所以引用Redisson来实现分布式锁。
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
</dependency>
代码逻辑大致和非集群的类似，redisson拿到的锁是有超时时间的，时间到了自动释放。当事件A同时被Bot A、Bot B监听时，若Bot A拿到了锁但是消费事件时突然宕机，无法正常释放锁。若锁的超时时间大于事件Ids的有效时间，那Bot B可以正常消费事件。反之Bot B会被系统判定为事件重复消费。
@Component
@RequiredArgsConstructor
public class GroupMessageHandler {
private final RedissonClient redissonClient;
private final RepetitiveGroupService repetitiveGroupService;
@Resource
private ReactiveZSetOperations<Long, Integer> reactiveZSetOperations;
public void handle(@NotNull GroupMessageEvent event) {
long groupId = event.getGroup().getId();
long botId = event.getBot().getId();
RepetitiveGroup repetitiveGroup = repetitiveGroupService.get(groupId)
.share()
.block();
RLock lock = null;
if (hasBeenConsumed(event, groupId)) return;
if (containMultiBots(repetitiveGroup, botId)) {
lock = redissonClient.getLock(IdUtil.fastSimpleUUID());
lock.lock(20, TimeUnit.SECONDS);
}
consume();
if (lock != null) lock.unlock();
}
// 检查事件是否以及被消费过
private boolean hasBeenConsumed(@NotNull GroupMessageEvent event, long groupId) {
long second = new Date().getTime();
Range<Double> range = Range.closed(0.0, second + 0.0);
reactiveZSetOperations.removeRangeByScore(groupId, range);
int[] ids = event.getSource().getIds();
List<DefaultTypedTuple<Integer>> list = new ArrayList<>();
for (int id : ids) {
list.add(new DefaultTypedTuple<>(id, second + 10.0));
}
Long unExist = reactiveZSetOperations.addAll(groupId, list)
.share()
.block();
return unExist == null || unExist == 0;
}
// 检查这个群是否有多个机器人监听群消息
private boolean containMultiBots(RepetitiveGroup repetitiveGroup, Long botId) {
return repetitiveGroup.getId() != null && repetitiveGroup.getBotIds().contains(botId);
}
// 消费事件
private void consume() {
}
}
其他业务对多Bot的适配
RepetitiveGroup，存储加入了多个机器人的群
@Data
@AllArgsConstructor
@NoArgsConstructor
public class RepetitiveGroup implements Serializable {
private static final long serialVersionUID = 1L;
@Id
private Long id;
private List<Long> botIds;
}
RepetitiveGroupService实现CRUD
@Service
public class RepetitiveGroupService {
@Resource
private ReactiveRedisOperations<String, RepetitiveGroup> reactiveRedisOperations;
public static final String HASH_KEY = RepetitiveGroup.class.getSimpleName();
public Mono<RepetitiveGroup> get(Long groupId) {
return reactiveRedisOperations.opsForHash()
.get(HASH_KEY, groupId)
.map(v -> (RepetitiveGroup) v)
.switchIfEmpty(Mono.defer(() -> Mono.just(new RepetitiveGroup())));
}
public Flux<RepetitiveGroup> getAll() {
return reactiveRedisOperations.opsForHash()
.values(HASH_KEY)
.map(v -> (RepetitiveGroup) v);
}
public Mono<Boolean> upset(RepetitiveGroup repetitiveGroup) {
return reactiveRedisOperations.opsForHash()
.put(HASH_KEY, repetitiveGroup.getId(), repetitiveGroup);
}
public Mono<Boolean> delete(RepetitiveGroup repetitiveGroup) {
return reactiveRedisOperations.opsForHash()
.remove(HASH_KEY, repetitiveGroup.getId())
.map(v -> v.equals(1L));
}
}
BotOnlineHandler：
@Slf4j
@Component
@RequiredArgsConstructor
public class BotOnlineHandler {
private final SentinelService sentinelService;
private final RepetitiveGroupService repetitiveGroupService;
@Resource
private ReactiveSetOperations<String, Long> reactiveSetOperations;
public void handle(@NotNull BotOnlineEvent event) {
Bot bot = event.getBot();
long id = bot.getId();
log.info("******BotOnlineHandler：QQ：{}", id);
List<Bot> instances = Bot.getInstances();
Set<Long> groupIds = bot.getGroups()
.parallelStream()
.map(Group::getId)
.collect(Collectors.toSet());
groupIds.parallelStream()
.forEach(v -> reactiveSetOperations.add(id + "", v));
instances.removeIf(v -> v.getId() == id);
instances.parallelStream()
.forEach(v -> sinter(id, v.getId()));
}
private void sinter(Long self, Long target) {
ArrayList<Long> list = new ArrayList<>();
list.add(self);
list.add(target);
reactiveSetOperations.intersect(self + "", target + "")
.subscribe(v -> insert(v, list));
}
private void insert(Long groupId, List<Long> botIds) {
RepetitiveGroup repetitiveGroup = repetitiveGroupService.get(groupId)
.share()
.block();
if (repetitiveGroup.getId() == null)
repetitiveGroup.setBotIds(new ArrayList<>());
Set<Long> set = new HashSet<>(botIds);
set.addAll(repetitiveGroup.getBotIds());
repetitiveGroup.setBotIds(new ArrayList<>(set));
repetitiveGroupService.upset(repetitiveGroup);
}
}
BotOfflineHandler：
@Slf4j
@Component
@RequiredArgsConstructor
public class BotOfflineHandler {
private final RepetitiveGroupService repetitiveGroupService;
public void handle(@NotNull BotOfflineEvent event) {
Bot bot = event.getBot();
long id = bot.getId();
String reason = "未知";
if (event instanceof BotOfflineEvent.Active)
reason = "主动下线";
if (event instanceof BotOfflineEvent.Force)
reason = "被挤下线";
if (event instanceof BotOfflineEvent.Dropped)
reason = "被服务器断开或因网络问题而掉线";
if (event instanceof BotOfflineEvent.RequireReconnect)
reason = "服务器主动要求更换另一个服务器";
log.info("******BotOfflineHandler：QQ：{}，Reason：{}", id, reason);
repetitiveGroupService.getAll()
.parallel()
.runOn(Schedulers.parallel())
.sequential()
.filter(v -> v.getBotIds().contains(id))
.map(v -> updateBotIdList(id, v))
.subscribe(this::updateOrDeleteRepetitiveGroup);
bot.close();
}
@NotNull
private RepetitiveGroup updateBotIdList(long id, RepetitiveGroup repetitiveGroup) {
List<Long> botIds = repetitiveGroup.getBotIds();
botIds.remove(id);
repetitiveGroup.setBotIds(botIds);
return repetitiveGroup;
}
private void updateOrDeleteRepetitiveGroup(RepetitiveGroup repetitiveGroup) {
if (repetitiveGroup.getBotIds().size() < 2) {
repetitiveGroupService.delete(repetitiveGroup);
} else {
repetitiveGroupService.upset(repetitiveGroup);
}
}
}
MemberJoinHandler：
@Component
@RequiredArgsConstructor
public class MemberJoinHandler {
private final RedissonClient redissonClient;
private final RepetitiveGroupService repetitiveGroupService;
public void handle(@NotNull MemberJoinEvent event) {
long memberId = event.getMember().getId();
Group group = event.getGroup();
if (event.getBot().getId() == memberId) return;
RLock lock = redissonClient.getLock(IdUtil.fastSimpleUUID());
lock.lock(30, TimeUnit.SECONDS);
List<Long> idList = Bot.getInstances()
.parallelStream()
.map(Bot::getId)
.collect(Collectors.toList());
if (!idList.contains(memberId)) return;
RepetitiveGroup repetitiveGroup = repetitiveGroupService.get(group.getId())
.share()
.block();
List<Long> botIds = idList.parallelStream()
.filter(group::contains)
.collect(Collectors.toList());
if (repetitiveGroup.getId() != null && botIds.size() == repetitiveGroup.getBotIds().size()) return;
repetitiveGroup.setId(group.getId());
repetitiveGroupService.upset(repetitiveGroup);
lock.unlock();
}
}
MemberLeaveHandler：
@Component
@RequiredArgsConstructor
public class MemberLeaveHandler {
private final RedissonClient redissonClient;
private final RepetitiveGroupService repetitiveGroupService;
public void handle(@NotNull MemberLeaveEvent event) {
long memberId = event.getMember().getId();
Group group = event.getGroup();
if (event.getBot().getId() == memberId) return;
RLock lock = redissonClient.getLock(IdUtil.fastSimpleUUID());
lock.lock(30, TimeUnit.SECONDS);
List<Long> idList = Bot.getInstances()
.parallelStream()
.map(Bot::getId)
.collect(Collectors.toList());
if (!idList.contains(memberId)) return;
RepetitiveGroup repetitiveGroup = repetitiveGroupService.get(group.getId())
.share()
.block();
if (repetitiveGroup.getId() == null) return;
List<Long> botIds = idList.parallelStream()
.filter(group::contains)
.collect(Collectors.toList());
if (botIds.size() > 1) {
repetitiveGroup.setId(group.getId());
repetitiveGroup.setBotIds(botIds);
repetitiveGroupService.upset(repetitiveGroup);
}
if (botIds.size() == 1)
repetitiveGroupService.delete(repetitiveGroup);
lock.unlock();
}
}
至此，Bot相关配置到此结束，下面介绍Nacos↓↓↓
Nacos
整合Nacos
这里只做Nacos Client的相关配置，Nacos Server端的请参阅Nacos官方文档
相关依赖：（注：因为要配合OpenFeign，所以将ribbon替换成了Loadbalancer）
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
<exclusions>
<exclusion>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
</exclusion>
</exclusions>
</dependency>
接入配置中心
本项目中使用Nacos的主要目的还是其动态下发配置的便利。
先设置bootstrap.yml，高版本的Spring Boot不支持bootstrap.yml，可以引入以下依赖：
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-bootstrap</artifactId>
</dependency>
bootstrap.yml配置如下：
spring:
application:
name: qqrobot
cloud:
nacos:
server-addr: localhost:8848
config:
# 如果在Nacos Web端没设置命名空间的可以删除
namespace: 3e4a06ab-3139-46e8-bc82-888154383de0
# 可以指定配置文件名，也可以不配置，用默认规则
name: qqrobot-dev.yml
main:
# 允许Bean覆盖
allow-bean-definition-overriding: true
使用@EnableDiscoveryClient就可以让模块注册进Nacos了
设置配置文件
Nacos Web端设置配置文件的相关操作我这边直接跳过，说说后端这边，
可以用@Value或@ConfigurationProperties(prefix = "xxx")取值，使用@RefreshScope注解即可开启自动刷新
@Data
@Component
@RefreshScope
public class DynamicProperty {
@Value("${accounts}")
private String accounts;
}
这边说几个坑。有几个List写法我试了几次，后端自动装配没成功（从Nacos获取配置）
list<Bean>
test:
list:
- a:1
b:1
test:
list:
- {a:1,b:1}
List<Map<String,String>>
test:
list:
- a:1
b:1
所以我建议JSON
监听Nacos配置刷新事件
当Nacos刷新本地项目的配置后会被RefreshEventListener(org.springframework.cloud.endpoint.event.RefreshEventListener)监听，通过参考onApplicationEvent()方法我们可以实现一个监听器
其中robotService是Bot的控制类，示例中我将机器人账号信息存入Nacos，当相关信息刷新后调用refreshBot()更新相关机器人的状态（上线、下线）
@Configuration
@RequiredArgsConstructor
public class RefreshEventListenerConfig {
private final RobotService robotService;
private AtomicBoolean ready = new AtomicBoolean(false);
@EventListener
public void listen(ApplicationEvent event) {
if (event instanceof ApplicationReadyEvent)
this.ready.compareAndSet(false, true);
if (event instanceof RefreshEvent && this.ready.get())
robotService.refreshBot();
}
}
至此，Nacos相关的适配到此结束
Sentinel
接入Sentinel的目的是进行动态流控，当机器人全部离线时，如果有请求打进来，立刻执行兜底方法，并存储请求参数做延迟处理。当机器人至少有一个在线时则开放端口，同时对已存储的请求进行消费。这块功能未来会持续优化。
整合Sentinel
我们要引入sentinel组件，sentinel对web框架的适配（这里我选择了webflux）以及配置持久化方式
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.csp</groupId>
<artifactId>sentinel-spring-webflux-adapter</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.csp</groupId>
<artifactId>sentinel-datasource-nacos</artifactId>
</dependency>
配置Sentinel客户端
在bootstrap.yml中配置Sentinel Dashboard的地址
spring:
cloud:
sentinel:
transport:
dashboard: localhost:38081
port: 38719
这里只做Sentinel客户端的适配，服务端配置请参考Sentinel官方文档
@Configuration
@RequiredArgsConstructor
public class SentinelConfig {
private final List<ViewResolver> viewResolvers;
private final ServerCodecConfigurer serverCodecConfigurer;
@Bean
@Order(-1)
public SentinelBlockExceptionHandler sentinelBlockExceptionHandler() {
return new SentinelBlockExceptionHandler(viewResolvers, serverCodecConfigurer);
}
@Bean
@Order(-1)
public SentinelWebFluxFilter sentinelWebFluxFilter() {
return new SentinelWebFluxFilter();
}
}
自定义流控返回
先写一个请求参数存储类UnhandledHttpRequest
@Data
@AllArgsConstructor
public class UnhandledHttpRequest implements Serializable {
private static final long serialVersionUID = 1L;
@Id
private String id;
private String body;
private String method;
public UnhandledHttpRequest() {
id = IdUtil.fastSimpleUUID();
}
public UnhandledHttpRequest(String body) {
id = IdUtil.fastSimpleUUID();
this.body = body;
}
}
相关的服务类UnhandledHttpRequestService
@Service
public class UnhandledHttpRequestService {
@Resource
private ReactiveRedisOperations<String, UnhandledHttpRequest> reactiveRedisOperations;
public static final String HASH_KEY = RepetitiveGroup.class.getSimpleName();
public Mono<UnhandledHttpRequest> get(String id) {
return reactiveRedisOperations.opsForHash()
.get(HASH_KEY, id)
.map(v -> (UnhandledHttpRequest) v)
.switchIfEmpty(Mono.defer(() -> Mono.just(new UnhandledHttpRequest())));
}
public Flux<UnhandledHttpRequest> getAll() {
return reactiveRedisOperations.opsForHash()
.values(HASH_KEY)
.map(v -> (UnhandledHttpRequest) v);
}
public Mono<Boolean> upset(UnhandledHttpRequest unhandledHttpRequest) {
return reactiveRedisOperations.opsForHash()
.put(HASH_KEY, unhandledHttpRequest.getId(), unhandledHttpRequest);
}
public Mono<Boolean> delete(UnhandledHttpRequest unhandledHttpRequest) {
return reactiveRedisOperations.opsForHash()
.remove(HASH_KEY, unhandledHttpRequest.getId())
.map(v -> v.equals(1L));
}
}
最后定义一个返回内容处理类，WebFlux这边获取request的body没有Servlet容易，我试了几种方法都失败了，示例代码仅展示获取在URL上的请求参数。
@Component
@RequiredArgsConstructor
public class RobotBlockExceptionHandler implements BlockRequestHandler {
private final UnhandledHttpRequestService unhandledHttpRequestService;
@SneakyThrows
@Override
public Mono<ServerResponse> handleRequest(ServerWebExchange exchange, Throwable t) {
URI uri = exchange.getRequest().getURI();
JSONObject jsonObject = new JSONObject();
jsonObject.putAll(decodeBody(uri.getRawQuery()));
UnhandledHttpRequest unhandledHttpRequest = new UnhandledHttpRequest(jsonObject.toJSONString());
if (uri.getPath().contains("/send/msg")) {
unhandledHttpRequest.setMethod("/send/msg");
} else {
unhandledHttpRequest.setMethod("/collect/msg");
}
unhandledHttpRequestService.upset(unhandledHttpRequest);
jsonObject.clear();
jsonObject.put("msg", "机器人服务正忙，请稍后重试");
jsonObject.put("code", HttpStatus.TOO_MANY_REQUESTS.value());
return ServerResponse.status(HttpStatus.TOO_MANY_REQUESTS)
.contentType(MediaType.APPLICATION_JSON)
.bodyValue(jsonObject.toJSONString());
}
private Map<String, Object> decodeBody(String body) {
if (StrUtil.isBlank(body)) return new HashMap<>();
if (body.contains("&") && body.contains("="))
return Arrays.stream(body.split("&"))
.map(s -> s.split("="))
.collect(Collectors.toMap(arr -> arr[0], arr -> arr[1]));
return JSON.parseObject(body)
.getInnerMap();
}
}
其他类对Sentinel的适配
逻辑
我这边仅仅简单实现了，当Bot全部离线时，相关端口全部熔断，当有Bot在线时恢复。这个做法不可实际应用，更好点的做法则是依据request进行的限流。
BotOfflineHandler
private final SentinelService sentinelService;
if (Bot.getInstances().isEmpty()) {
FlowRule rule = new FlowRule("/send/msg");
rule.setCount(0);
rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
rule.setLimitApp("default");
sentinelService.saveOrUpdateFlowRule(rule);
rule.setResource("/collect/msg");
sentinelService.saveOrUpdateFlowRule(rule);
}
BotOnlineHandler
private final SentinelService sentinelService;
if (Bot.getInstances().isEmpty()) {
FlowRule rule = new FlowRule("/send/msg");
rule.setCount(10);
rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
rule.setLimitApp("default");
sentinelService.saveOrUpdateFlowRule(rule);
rule.setResource("/collect/msg");
sentinelService.saveOrUpdateFlowRule(rule);
}
配置持久化
修改bootstrap.yml，这里仅做参考，具体参数详见官网
spring:
cloud:
sentinel:
transport:
dashboard: localhost:38081
port: 38719
datasource:
flow-rule:
nacos:
server-addr: localhost:8848
username: nacos
password: nacos
namespace: 3e4a06ab-3139-46e8-bc82-888154383de0
data-id: qqrobot-sentinel-flow-dev.json
rule-type: flow
至此，Sentinel相关的适配到此结束
配置参考
Redis配置参考
@Configuration
public class RedisConfig extends CachingConfigurerSupport {
@Bean
public ReactiveRedisTemplate<String, String> reactiveRedisTemplate(ReactiveRedisConnectionFactory connectionFactory) {
return new ReactiveRedisTemplate<>(connectionFactory, RedisSerializationContext.string());
}
@Bean
public ReactiveRedisConnection connection(ReactiveRedisConnectionFactory connectionFactory) {
return connectionFactory.getReactiveConnection();
}
@Bean
ReactiveRedisOperations<String, Object> reactiveRedisOperations(ReactiveRedisConnectionFactory factory) {
return new ReactiveRedisTemplate<>(factory, getObjectRedisSerializationContext());
}
@Bean
ReactiveSetOperations<String, Object> reactiveSetOperations(ReactiveRedisConnectionFactory factory) {
return new ReactiveRedisTemplate<>(factory, getObjectRedisSerializationContext()).opsForSet();
}
@Bean
ReactiveZSetOperations<String, Object> reactiveZSetOperations(ReactiveRedisConnectionFactory factory) {
return new ReactiveRedisTemplate<>(factory, getObjectRedisSerializationContext()).opsForZSet();
}
private static RedisSerializationContext<String, Object> getObjectRedisSerializationContext() {
return RedisSerializationContext
.<String, Object>newSerializationContext(new GenericFastJsonRedisSerializer())
.key(new FastJsonRedisSerializer<>(String.class))
.value(new FastJsonRedisSerializer<>(Object.class))
.hashKey(new FastJsonRedisSerializer<>(String.class))
.hashValue(new FastJsonRedisSerializer<>(Object.class))
.build();
}
}
Redisson配置参考
@Configuration
public class RedissonConfig {
@Bean
public RedissonClient redissonClient() {
Config config = new Config();
config.setLockWatchdogTimeout(10000L);
SingleServerConfig singleServerConfig = config.useSingleServer();
singleServerConfig.setPassword("password");
singleServerConfig.setAddress("redis://localhost:6379");
singleServerConfig.setDatabase(1);
return Redisson.create(config);
}
}
