Skip to content

fix: 增加 BrokerClientItem 连接失败自动重试机制 #542

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mcqwy wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: 增加 BrokerClientItem 连接失败自动重试机制 #542

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 120 additions & 11 deletions ...bolt/bolt-core/src/main/java/com/iohao/game/bolt/broker/core/client/BrokerClientItem.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
import com.iohao.game.bolt.broker.core.message.InnerModuleMessage;
import com.iohao.game.bolt.broker.core.message.InnerModuleVoidMessage;
import com.iohao.game.common.kit.CollKit;
import com.iohao.game.common.kit.concurrent.timer.delay.DelayTaskKit;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.Setter;
Expand All @@ -53,10 +54,13 @@
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.net.UnknownHostException;
import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
* 客户连接项
Expand Down Expand Up @@ -100,8 +104,23 @@ public enum Status {
AwareInject awareInject;
int brokerServerWithNo;

// ================== 重试相关字段 ==================
/** 重试任务的唯一标识（实例级别） */
private final String retryTaskId;
/** 最大重试次数,默认0不重试,-1无限重试 */
private int maxRetryCount = IoGameGlobalConfig.maxRetryCount;
/** 重试间隔（毫秒） 默认30秒*/
private long retryDelayMillis = IoGameGlobalConfig.retryDelayMillis;
/** 当前重试次数 */
private int retryCount = 0;
/** 是否正在重试中（防止并发触发） */
private final AtomicBoolean isRetrying = new AtomicBoolean(false);
// ====================================================

public BrokerClientItem(String address) {
this.address = address;
// 使用 address + 实例哈希码 作为唯一标识，确保多实例不冲突
this.retryTaskId = "broker_reconnect_" + address + "_" + System.identityHashCode(this);
this.rpcClient = new RpcClient();
// 重连选项
rpcClient.option(BoltClientOption.CONN_RECONNECT_SWITCH, true);
Expand Down Expand Up @@ -289,42 +308,132 @@ private void internalOneway(Object responseObject) {
}
}

// ================== 核心修改：替换原有 startup / send / registerToBroker ==================
/**
* 注册到网关 broker
* 客户端服务器注册到网关服
* 注册到网关（保持 public，供外部调用）
* <p>
* 此方法只执行注册逻辑，不触发重试。
* 外部调用时（如网关请求），不触发重试，由外部决定是否重试。
* @return true 注册成功，false 注册失败
*/
public void registerToBroker() {
public boolean registerToBroker() {
try {
// 客户端服务器注册到游戏网关服
BrokerClientModuleMessage brokerClientModuleMessage = this.brokerClient.getBrokerClientModuleMessage();
this.rpcClient.oneway(address, brokerClientModuleMessage);

TimeUnit.MILLISECONDS.sleep(100);
this.status = Status.ACTIVE;
this.brokerClient.getBrokerClientManager().resetSelector();

this.barSkeleton.getRunners().onStartAfter();

this.with();

log.info("逻辑服注册网关成功: {}", address);
return true;
} catch (RemotingException | InterruptedException e) {
log.error(e.getMessage(), e);
log.error("注册网关失败: {}", address, e);
// 注意：外部调用时（如 Processor），不触发重试，仅记录日志
// 内部调用时（如 tryConnect），由调用方负责重试
return false;
}
}

/**
* 启动客户端（非阻塞）
* <p>
* 首次尝试连接，若失败则通过 DelayTaskKit 自动重试，直到成功或达到最大重试次数。
* 解决了 DNS 临时失效、网关未启动等场景下的连接恢复问题。
*/
public void startup() {
this.rpcClient.startup();
this.send();
// 首次尝试连接（非阻塞）
this.tryConnect();
}

private void send() {
/**
* 尝试连接网关（内部方法）
* <p>
* 发送握手消息，成功后调用 registerToBroker()，若失败则调度重试
*/
private void tryConnect() {
if (this.status == Status.ACTIVE) {
return;
}

if (!isRetrying.compareAndSet(false, true)) {
log.debug("已有重试任务进行中，忽略本次触发");
return;
}

try {
// 1. 发送连接握手消息（原 send 逻辑）
var message = new BrokerClientItemConnectMessage();
this.rpcClient.oneway(address, message);

// 2. 连接成功，执行注册
log.info("网关连接成功: {}", address);
this.status = Status.ACTIVE;
// 直接调用 public 方法注册
// 如果注册失败，回退状态并触发重试
if (!this.registerToBroker()) {
this.status = Status.DISCONNECT;
isRetrying.set(false);//先释放锁，再调度重试
this.scheduleRetry(new RemotingException("注册失败"));
return;
}
// 重置重试计数
retryCount = 0;
isRetrying.set(false);

} catch (RemotingException | InterruptedException e) {
log.error(e.getMessage(), e);
// 连接失败（握手失败），释放锁并调度重试
isRetrying.set(false);
// 注意：如果是因为注册失败导致异常，这里也会重试，但由于 registerToBroker 内部捕获了异常，不会向外抛出
// 所以这里捕获的主要是握手异常（如连接被拒绝、DNS 解析失败等）
this.scheduleRetry(e);
} catch (Exception e) {
isRetrying.set(false);
log.error("连接网关时发生未知异常，将重试", e);
this.scheduleRetry(e);
}
}

/**
* 使用 DelayTaskKit 调度下一次重试
*/
private void scheduleRetry(Exception e) {
// 检查是否达到最大重试次数（-1 表示无限重试）
if (maxRetryCount != -1 && retryCount >= maxRetryCount) {
log.error("重试次数已达上限 ({}), 放弃连接: {}", maxRetryCount, address);
return;
}

retryCount++;
// 创建延时任务，覆盖之前的同名任务
DelayTaskKit.of(retryTaskId, () -> {
log.info("执行第 {} 次重试，连接网关: {}", retryCount, address);
// 递归调用 tryConnect 进行重试
this.tryConnect();
}).plusTime(Duration.ofMillis(retryDelayMillis)).task();

// 日志记录（区分 DNS 错误和其他错误）
if (e.getCause() instanceof UnknownHostException) {
log.warn("DNS解析失败 ({}), {} 秒后重试 (第 {} 次)",
address, retryDelayMillis / 1000, retryCount);
} else {
log.warn("连接网关失败 ({}), {} 秒后重试 (第 {} 次): {}",
address, retryDelayMillis / 1000, retryCount, e.getMessage());
}
}

/**
* 取消重试任务（可在应用关闭时调用）
*/
public void cancelRetry() {
DelayTaskKit.cancel(retryTaskId);
isRetrying.set(false);
log.info("已取消网关重连任务: {}", address);
}


private void with() {
int withNo = this.brokerClient.getWithNo();

Expand Down
4 changes: 4 additions & 0 deletions ...lt/bolt-core/src/main/java/com/iohao/game/bolt/broker/core/common/IoGameGlobalConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,10 @@ public class IoGameGlobalConfig {
public boolean externalLog;
/** true 开启广播相关日志，默认为 false */
public boolean broadcastLog;
/** 重连网关间隔,默认30秒 */
public long retryDelayMillis = 30000;
/** 重连网关最大重试次数,默认0不重试,-1无限重试 */
public int maxRetryCount = 0;
/**
* Broker（游戏网关）转发消息容错配置
* <pre>
Expand Down