robot
/
robot_webFenli
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

红中机器人线程池

master
zhouwei 2026-02-26 19:28:22 +08:00
parent a3b46394c4
commit 74ce44217e
5 changed files with 242 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
robots/majiang/robot_mj_hz/src/main/java/robot/mj/EXGameController.java
View File

@ -388,7 +388,7 @@ public class EXGameController extends GameController {
robotConnectionManager.disconnectFromGameServer(connecId);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.error("加入房间超时异常");
}
});
robotUser.setIntoRoomTime(robotConnectionManager.getTime());

182
robots/majiang/robot_mj_hz/src/main/java/robot/mj/RobotConnectionManager.java
View File

@ -9,6 +9,7 @@ import com.taurus.core.events.IEventListener;
import com.taurus.core.plugin.redis.Redis;
import com.taurus.core.util.ICallback;
import com.taurus.core.util.StringUtil;
import org.apache.log4j.Logger;
import robot.mj.business.AccountBusiness;
import robot.mj.handler.HuNanHongZhong;
import robot.mj.info.RobotUser;
@ -22,6 +23,7 @@ import java.util.*;
import java.util.concurrent.*;
import static robot.mj.EXGameController.robotRoomMapping;
import static robot.mj.thread.ThreadPoolConfig.scheduleDelay;
/**
* 机器人连接管理器 - 管理与游戏服务器的连接
@ -29,6 +31,7 @@ import static robot.mj.EXGameController.robotRoomMapping;
public class RobotConnectionManager {
private static final Map<String, HuNanHongZhong> huNanHongZhongInstances = new ConcurrentHashMap<>();
private static final Logger log = Logger.getLogger(RobotConnectionManager.class);
private final EXGameController exGameController;
private final String host= Config.GAME_SERVER_HOST;
@ -346,68 +349,78 @@ public class RobotConnectionManager {
}
//玩家加入房间
else if ("2001".equalsIgnoreCase(command)) {
CompletableFuture.runAsync(() -> {
sleepTime(6000);
scheduleDelay(() -> {
try {
String roomKey = String.valueOf(robotUser.getCurrentRoomId());
String roomKey = String.valueOf(robotUser.getCurrentRoomId());
//查询该房间的玩家信息
String playersStr = jedis0.hget("room:"+roomKey, "players");
if (!playersStr.equals("[]")) {
String players = playersStr.substring(1, playersStr.length() - 1);
String[] playerIds = players.split(",");
//查询该房间的玩家信息
String playersStr = jedis0.hget("room:"+roomKey, "players");
if (!playersStr.equals("[]")) {
String players = playersStr.substring(1, playersStr.length() - 1);
String[] playerIds = players.split(",");
//判断只有当前机器人一个玩家
if (playerIds.length == 1) {
int playerId = Integer.parseInt(playerIds[0].trim());
if (playerId == robotId) {
//判断只有当前机器人一个玩家
if (playerIds.length == 1) {
int playerId = Integer.parseInt(playerIds[0].trim());
if (playerId == robotId) {
//发送退出房间协议
ITObject params = TObject.newInstance();
client.send("1005", params, response -> {
EXGameController.removeRobotRoomInfo(String.valueOf(robotId));
//更新机器人剩余数量
updateLeftoverRobot(robotId);
disconnectFromGameServer(connecId);
System.out.println("2002发送退出房间协议1005，robotId: {"+robotId+"}");
});
//发送退出房间协议
ITObject params = TObject.newInstance();
client.send("1005", params, response -> {
EXGameController.removeRobotRoomInfo(String.valueOf(robotId));
//更新机器人剩余数量
updateLeftoverRobot(robotId);
disconnectFromGameServer(connecId);
System.out.println("2002发送退出房间协议1005，robotId: {"+robotId+"}");
});
}
}
}
} catch (Exception e) {
log.error("处理玩家加入房间检查时发生异常", e);
}
});
}, 6, TimeUnit.SECONDS);
System.out.println("玩家{"+ robotUser.getCurrentRoomId()+"}加入房间:"+ param);
}
//玩家退出房间也要检查
else if ("2002".equalsIgnoreCase(command)) {
CompletableFuture.runAsync(() -> {
sleepTime(6000);
//直接使用定时任务替代Thread.sleep，避免嵌套异步调用
scheduleDelay(() -> {
Jedis jedis = Redis.use().getJedis();
try {
String roomKey = String.valueOf(robotUser.getCurrentRoomId());
String roomKey = String.valueOf(robotUser.getCurrentRoomId());
//查询该房间的玩家信息
String playersStr = jedis.hget("room:"+roomKey, "players");
if (!playersStr.equals("[]")) {
String players = playersStr.substring(1, playersStr.length() - 1);
String[] playerIds = players.split(",");
//查询该房间的玩家信息
String playersStr = jedis0.hget("room:"+roomKey, "players");
if (!playersStr.equals("[]")) {
String players = playersStr.substring(1, playersStr.length() - 1);
String[] playerIds = players.split(",");
//判断只有当前机器人一个玩家
if (playerIds.length == 1) {
int playerId = Integer.parseInt(playerIds[0].trim());
if (playerId == robotId) {
//判断只有当前机器人一个玩家
if (playerIds.length == 1) {
int playerId = Integer.parseInt(playerIds[0].trim());
if (playerId == robotId) {
//发送退出房间协议
ITObject params = TObject.newInstance();
client.send("1005", params, response -> {
EXGameController.removeRobotRoomInfo(String.valueOf(robotId));
//更新机器人剩余数量
updateLeftoverRobot(robotId);
disconnectFromGameServer(connecId);
System.out.println("2002发送退出房间协议1005，robotId: {"+robotId+"}");
});
//发送退出房间协议
ITObject params = TObject.newInstance();
client.send("1005", params, response -> {
EXGameController.removeRobotRoomInfo(String.valueOf(robotId));
//更新机器人剩余数量
updateLeftoverRobot(robotId);
disconnectFromGameServer(connecId);
System.out.println("2002发送退出房间协议1005，robotId: {"+robotId+"}");
});
}
}
}
} catch (Exception e) {
log.error("处理玩家退出房间检查时发生异常", e);
} finally {
if (jedis != null) {
jedis.close();
}
}
});
}, 6, TimeUnit.SECONDS);
}
//玩家解散房间
else if ("2005".equalsIgnoreCase(command)) {
@ -423,45 +436,52 @@ public class RobotConnectionManager {
disconnectFromGameServer(connecId);
}
else if ("2009".equalsIgnoreCase(command)) {
CompletableFuture.runAsync(() -> {
Integer paramRobotId = param.getInt("aid");
sleepTime(6000);
scheduleDelay(() -> {
Jedis localJedis = Redis.use().getJedis();
try {
Integer paramRobotId = param.getInt("aid");
if (robotUser != null) {
String roomKey = String.valueOf(robotUser.getCurrentRoomId());
if (robotUser != null) {
String roomKey = String.valueOf(robotUser.getCurrentRoomId());
//查询该房间的玩家信息
String playersStr = jedis0.hget(roomKey, "players");
if (!playersStr.equals("[]")) {
String players = playersStr.substring(1, playersStr.length() - 1);
String[] playerIds = players.split(",");
//查询该房间的玩家信息
String playersStr = jedis0.hget(roomKey, "players");
if (!playersStr.equals("[]")) {
String players = playersStr.substring(1, playersStr.length() - 1);
String[] playerIds = players.split(",");
//判断只有当前机器人一个玩家
if (playerIds.length == 1) {
int playerId = Integer.parseInt(playerIds[0].trim());
if (playerId == paramRobotId) {
//判断只有当前机器人一个玩家
if (playerIds.length == 1) {
int playerId = Integer.parseInt(playerIds[0].trim());
if (playerId == paramRobotId) {
String gpid = jedis0.hget(roomKey, "gpid");
String gpId = jedis0.hget(roomKey, "group");
String gpid = jedis0.hget(roomKey, "gpid");
String gpId = jedis0.hget(roomKey, "group");
//发送退出房间协议
ITObject params = TObject.newInstance();
client.send("1005", params, response -> {
EXGameController.removeRobotRoomInfo(String.valueOf(paramRobotId));
//断开连接
disconnectFromGameServer(connecId);
//更新机器人剩余数量
updateLeftoverRobot(paramRobotId);
System.out.println("2009发送退出房间协议1005，robotId: {"+paramRobotId+"}");
});
//发送退出房间协议
ITObject params = TObject.newInstance();
client.send("1005", params, response -> {
EXGameController.removeRobotRoomInfo(String.valueOf(paramRobotId));
//断开连接
disconnectFromGameServer(connecId);
//更新机器人剩余数量
updateLeftoverRobot(paramRobotId);
System.out.println("2009发送退出房间协议1005，robotId: {"+paramRobotId+"}");
});
}
}
}
}
} catch (Exception e) {
log.error("处理机器人房间映射检查时发生异常", e);
} finally {
if (localJedis != null) {
localJedis.close();
}
}
});
}, 6, TimeUnit.SECONDS);
}
} catch (Exception e) {
throw new RuntimeException(e);
log.error("处理接收到的游戏协议异常");
} finally {
jedis0.close();
jedis2.close();
@ -525,14 +545,14 @@ public class RobotConnectionManager {
}
});
} catch (Exception e) {
throw new RuntimeException(e);
log.error("机器人登录异常");
}
}
public void connectGame(RobotUser robotUser){
if(robotUser.isLogin){
if(robotUser.getClient()==null){
TaurusClient client = new TaurusClient(robotUser.getGameHost()+":"+robotUser.getGamePort(), "game", TaurusClient.ConnectionProtocol.Tcp);
TaurusClient client = new TaurusClient(robotUser.getGameHost()+":"+robotUser.getGamePort(), "hz"+robotUser.getRobotId(), TaurusClient.ConnectionProtocol.Tcp);
client.setSession(robotUser.getLoginsession());
client.connect();
setupEventListeners(client, robotUser.getCurrentRoomId()+"_"+robotUser.getRobotId());
@ -540,7 +560,7 @@ public class RobotConnectionManager {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
log.error("连接超时异常");
}
robotUser.setClient(client);
EXGameController.robotRoomMapping.put(robotUser.getCurrentRoomId()+"_"+robotUser.getRobotId(), robotUser);
@ -551,14 +571,14 @@ public class RobotConnectionManager {
robotUser.setIsconnect(true);
}else{
System.out.println("reconnect"+robotUser.getClient().getGameID());
TaurusClient client = new TaurusClient(robotUser.getGameHost()+":"+robotUser.getGamePort(), "game", TaurusClient.ConnectionProtocol.Tcp);
TaurusClient client = new TaurusClient(robotUser.getGameHost()+":"+robotUser.getGamePort(), "hz"+robotUser.getRobotId(), TaurusClient.ConnectionProtocol.Tcp);
client.setSession(robotUser.getLoginsession());
client.connect();
robotUser.setIsconnect(client.isConnected());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
log.error("连接超时异常");
}
robotUser.setClient(client);
EXGameController.robotRoomMapping.put(robotUser.getCurrentRoomId()+"_"+robotUser.getRobotId(), robotUser);
@ -597,7 +617,7 @@ public class RobotConnectionManager {
//添加延迟
Thread.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
log.error("连接超时异常");
}
}

26
robots/majiang/robot_mj_hz/src/main/java/robot/mj/business/AccountBusiness.java
View File

@ -21,12 +21,14 @@ import com.taurus.web.Controller;
import com.taurus.web.WebException;
import redis.clients.jedis.Jedis;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
public class AccountBusiness extends Controller {
private static final org.apache.log4j.Logger log = org.apache.log4j.Logger.getLogger(AccountBusiness.class);
private static Logger logger = Logger.getLogger(AccountBusiness.class);
private final ITObject fillLoginData(String session, int accountid) {
@ -81,7 +83,7 @@ public class AccountBusiness extends Controller {
}
public final ITObject fastLogin(int userid) throws Exception {
public final ITObject fastLogin(int userid){
Jedis jedis = Redis.use("group1_db0").getJedis();
ITObject resData = null;
try {
@ -102,7 +104,7 @@ public class AccountBusiness extends Controller {
if (StringUtil.isNotEmpty(idPwdBan))
{
logger.error("id:"+acc_bean.id+" ban login");
throw new WebException(ErrorCode.BAN_LOGIN);
//throw new WebException(ErrorCode.BAN_LOGIN);
}
resData.putString("token", token);
return resData;
@ -117,7 +119,7 @@ public class AccountBusiness extends Controller {
}
public final ITObject idPasswordLogin(int id, String password) throws Exception {
public final ITObject idPasswordLogin(int id, String password){
logger.info("id:" + id + " login");
@ -149,12 +151,17 @@ public class AccountBusiness extends Controller {
if (StringUtil.isNotEmpty(idPwdBan)) {
System.out.println("进入了77777777777777777777");
logger.error("id:" + id + " ban login");
throw new WebException(ErrorCode.BAN_LOGIN);
//throw new WebException(ErrorCode.BAN_LOGIN);
}
System.out.println("进入了9999999999999");
ITArray resultArray = DataBase.use().executeQueryByTArray(sql);
if (resultArray.size() == 0) {
ITArray resultArray = null;
try {
resultArray = DataBase.use().executeQueryByTArray(sql);
} catch (SQLException e) {
log.error(e);
}
if (resultArray.size() == 0) {
if (Redis.use("group1_db0").exists(id + "_pwd_token")) {
Redis.use("group1_db0").incrBy(id + "_pwd_token", 1);
} else {
@ -171,13 +178,13 @@ public class AccountBusiness extends Controller {
logger.error("pwd error count:" + count + " not login");
System.out.println("进入了00000000000");
throw new WebException(ErrorCode._NO_SESSION);
//throw new WebException(ErrorCode._NO_SESSION);
}
}
System.out.println("进入了111111111111");
throw new WebException(ErrorCode._FAILED);
//throw new WebException(ErrorCode._FAILED);
}
ITObject userData = resultArray.getTObject(0);
@ -279,9 +286,8 @@ public class AccountBusiness extends Controller {
/**
*
* @return
* @throws Exception
*/
private final int UpdateUserData(ITObject reqData, long id) throws Exception {
private final int UpdateUserData(ITObject reqData, long id) {
ITObject userData = TObject.newInstance();
userData.putInt("id", (int) id);

25
robots/majiang/robot_mj_hz/src/main/java/robot/mj/handler/HuNanHongZhong.java
View File

@ -17,6 +17,8 @@ import taurus.util.Util;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import static robot.mj.thread.ThreadPoolConfig.getBusinessThreadPool;
public class HuNanHongZhong {
public static int hongZhongCard = 0;
@ -347,8 +349,19 @@ public class HuNanHongZhong {
// cardInhand.remove(1);
System.out.println("执行id为：" + 0 + "的操作");
client.send("612", params, response -> {
System.out.println("操作成功: " + response.returnCode);
//使用线程池执行延迟动作
getBusinessThreadPool().execute(() -> {
try {
int delaySeconds = 1 + new Random().nextInt(2);
System.out.println("执行延迟" + delaySeconds + "秒");
Thread.sleep(delaySeconds * 1000);
client.send("612", params, response -> {
});
} catch (Exception e) {
System.err.println("执行动作时发生异常: " + e.getMessage());
}
});
return null;
}
@ -405,18 +418,16 @@ public class HuNanHongZhong {
System.out.println("目前机器人剩余手牌：" + hongZhongCardInhand.toString());
params.putString("session", session + "," + token);
CompletableFuture.runAsync(() -> {
//使用线程池替代CompletableFuture.runAsync + Thread.sleep
getBusinessThreadPool().execute(() -> {
try {
//缩短出牌延时
int ot = new Random().nextInt(2);
Thread.sleep(ot * 1000+1000);
client.send("611", params, response -> {
System.out.println("出牌成功: " + response.returnCode);
});
} catch (Exception e) {
System.out.println("Thread error");
//Thread.currentThread().interrupt();
}
});

106
robots/majiang/robot_mj_hz/src/main/java/robot/mj/thread/ThreadPoolConfig.java Normal file
View File

@ -0,0 +1,106 @@
package robot.mj.thread;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 线程池配置类
*/
public class ThreadPoolConfig {
// 优化后的线程池配置：针对机器人场景优化
private static final ExecutorService BUSINESS_THREAD_POOL =
new ThreadPoolExecutor(
10, // 核心线程数 - 减少核心线程数
50, // 最大线程数 - 降低最大线程数
30, // 空闲线程存活时间 - 缩短存活时间
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(10000), // 增大队列容量，减少拒绝任务
new ThreadFactory() {
private final AtomicInteger threadNumber = new AtomicInteger(1);
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r, "RobotBusinessThread-" + threadNumber.getAndIncrement());
t.setDaemon(true);
t.setPriority(Thread.NORM_PRIORITY - 2); // 进一步降低线程优先级
return t;
}
},
new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：由调用线程执行
);
// 添加定时任务线程池，专门处理延迟操作
private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE =
Executors.newScheduledThreadPool(4, new ThreadFactory() {
private final AtomicInteger threadNumber = new AtomicInteger(1);
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r, "RobotScheduledThread-" + threadNumber.getAndIncrement());
t.setDaemon(true);
t.setPriority(Thread.NORM_PRIORITY - 2);
return t;
}
});
public static ExecutorService getBusinessThreadPool() {
return BUSINESS_THREAD_POOL;
}
public static ScheduledExecutorService getScheduledExecutorService() {
return SCHEDULED_EXECUTOR_SERVICE;
}
/**
* 执行延迟任务，替代Thread.sleep
*/
public static void scheduleDelay(Runnable task, long delay, TimeUnit unit) {
System.out.println("提交延迟任务: 延迟" + delay + " " + unit + ", 当前时间: " + System.currentTimeMillis());
SCHEDULED_EXECUTOR_SERVICE.schedule(() -> {
try {
System.out.println("执行延迟任务开始: 当前时间: " + System.currentTimeMillis());
task.run();
System.out.println("执行延迟任务完成: 当前时间: " + System.currentTimeMillis());
} catch (Exception e) {
System.err.println("延迟任务执行异常: " + e.getMessage());
}
}, delay, unit);
}
/**
* 执行周期性任务
*/
public static void scheduleAtFixedRate(Runnable task, long initialDelay, long period, TimeUnit unit) {
SCHEDULED_EXECUTOR_SERVICE.scheduleAtFixedRate(task, initialDelay, period, unit);
}
/**
* 优雅关闭线程池，释放资源
*/
public static void shutdown() {
System.out.println("开始关闭线程池...");
// 关闭定时任务线程池
SCHEDULED_EXECUTOR_SERVICE.shutdown();
try {
if (!SCHEDULED_EXECUTOR_SERVICE.awaitTermination(5, TimeUnit.SECONDS)) {
SCHEDULED_EXECUTOR_SERVICE.shutdownNow();
}
} catch (InterruptedException e) {
SCHEDULED_EXECUTOR_SERVICE.shutdownNow();
Thread.currentThread().interrupt();
}
// 关闭业务线程池
BUSINESS_THREAD_POOL.shutdown();
try {
if (!BUSINESS_THREAD_POOL.awaitTermination(10, TimeUnit.SECONDS)) {
BUSINESS_THREAD_POOL.shutdownNow();
}
} catch (InterruptedException e) {
BUSINESS_THREAD_POOL.shutdownNow();
Thread.currentThread().interrupt();
}
System.out.println("线程池关闭完成");
}
}