【MINA】用mina做业务服之间的通信，实现业务负载均衡思路
学习mina目的还是搭建通信架构，学完mina我们了解了如何实现客户端和服务端，也就是一个正常channel我们是知道怎么建立的
但是问题是，我们应用环境通信分为两种
1.前后端通信
其实这个比较好实现，提供一个mina server端，供前端语言通过socket建连接就行，这个通信就算是ok了，编解码等通信解析的细节这里不讲了
以前的游戏服务端架构业务多用短连接，聊天用长连接，聊天的部分其实就是上面表述的情况
现在是长连接的天下，聊天依旧是长连接，业务也做成长连接，实现了真正意义上的长连接游戏架构，这其实就表述了一种当下典型架构，
就是后端提供两个开放的通信端口【即两个mina server】，供前端的socket连接，一个负责聊天，登录，注册，另一个负责其他业务，这样就实现了协议通信的负载均衡
2.后端的业务服通信【这是本文的重点】
那么后端的业务就不需要负载均衡吗？比如job，异步更新db，活动副本等
当然也是需要的，怎么做那，先拿1中的做个解释
mainserevr[聊天，登录，注册]---nodeserver[其他业务]
这两个mina sever端已经建立起来了，但是两个server之间还不能通信，我们有两个选择，要么在mainserevr上起个mina client去连nodeserver，要么在nodeserver
上起个mina client去连mainserevr，思路肯定是这样的，一旦这个通道建立了，其实互为server和client的，会有一个iosession被通道持有，只要有这个iosession，
就可以主动write，当然对于通道的另一端可以response，也可以通过取得iosession来主动写
实现方式，我们在nodeserevr上提供一个mainserverClient这样一个spring的bean去连接mainserver，这样在nodeserver上就可以向mainserevr发消息了
3.带着这个思路设计一下
我把游戏中的业务分为
public static final String SERVER_TYPE_NODE_STR = "nodeserver";// game node public static final String SERVER_TYPE_MAIN_STR = "mainserver";// 主server public static final String SERVER_TYPE_JOB_STR = "jobserver";// job server public static final String SERVER_TYPE_ASYNCDB_STR = "asyncdbserver";// 异步DB public static final String SERVER_TYPE_ACTIVE_STR = "activityserver";// 活动 public static final String SERVER_TYPE_OTHER_STR = "other";// 其他 public static final String SERVER_TYPE_GM_STR = "GM";//管理端
每次启动一种server时，首先启动一次mina serevr，然后启动多个mina client去连接其他的mina server，
比如启动nodeserevr 服务端，然后启动多个client分别连接mainserevr，jobserevr等的服务端，这样我就可以
在nodeserver上给其他业务serevr发请求了，具体启动哪些client看需要
搞一个启动server类型的方法
public static ClassPathXmlApplicationContext start(String serverTypeStr) { try { //关闭连接池的钩子线程 ProxoolFacade.disableShutdownHook(); //spring 的核心配置文件 String xmlFile = "applicationContext.xml"; .... log.info("启动 {} server................", serverTypeName); // 设置到系统环境变量 System.setProperty(NodeSessionMgr.SERVER_TYPE_KEY, serverType + ""); System.setProperty(NodeSessionMgr.SERVER_TYPE_NAME_KEY, serverTypeName); // final ClassPathXmlApplicationContext parent = new // ClassPathXmlApplicationContext( // xmlFile); String fileName = null; //这是把spring的住配置文件拆分了一部分内容出来，目前是只加载本server需要的bean if (serverType == NodeSessionMgr.SERVER_TYPE_NODE) { fileName = "wolf/app_nodeserver.xml"; } else { fileName = "wolf/app_server.xml"; } //手动启动spring final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( new String[] { xmlFile, fileName }); if (context != null) { ServiceLocator.getInstance().setApplicationContext(context); } // 启动socket server final WolfServer server = (WolfServer) ServiceLocator .getSpringBean("wolf_server"); server.setServerType(serverType); //这个调用就是我们熟悉的启动mina server端 server.start(); //这个动用做两件事，选区需要的serevr类型建立mina client连接 startClient(server); //钩子线程用来监听应用停止，为了做停止时的后续处理 Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { public void run() { _shutdown(); } }, "shutdownHookThread")); //为了支持web，springMVC，内置一个web server if (NodeSessionMgr.SERVER_TYPE_MAIN_STR .equalsIgnoreCase(serverTypeStr)) { JettyServer jettyServer = (JettyServer) ServiceLocator .getSpringBean("jettyServer"); jettyServer.start(); } log.info("start {} end................", serverTypeName); return context; } catch (Exception e) { e.printStackTrace(); shutdown(); } finally { } return null; }
在看下startClient(server);
private static void startClient(WolfServer server) { // asyncdbServer只会被连接，不会主动连接其他server // 这部分目的是过滤那些不需要主动连比人的serevr，比武我这里的异步db，和活动服 if (server.getServerType() == NodeSessionMgr.SERVER_TYPE_ASYNCDB || server.getServerType() == NodeSessionMgr.SERVER_TYPE_ACTIVE) { return; } // 发送game Server ip port到mainserver Map<String, Object> params = new HashMap<String, Object>(); params.put("nodeServerIp", server.getIp()); params.put("nodeServerPort", server.getPort()); params.put("serverType", server.getServerType()); //我需要mainserevr的client，就弄个bean在本服 final IWolfClientService mainServerClient = (IWolfClientService) ServiceLocator .getSpringBean("mainServerClient"); //这个位置其实就是mina的client连server端 mainServerClient.init(); Object localAddress = mainServerClient.registerNode(params); //同上，需要jobserevr的client final IWolfClientService jobServerClient = (IWolfClientService) ServiceLocator .getSpringBean("jobServerClient"); if (jobServerClient != null) { jobServerClient.init(); Map<String, Object> params1 = new HashMap<String, Object>(); params1.putAll(params); jobServerClient.registerNode(params1); } // } ..... }
再看下WolfClientService.init()
public void init() { if (start) return; if (wolfClient == null) { log.error("wolf client is null"); return; } //mina 的client 连接 mina server wolfClient.start(); if (wolfClient.isConnected()) start = true; }
再看下wolfclient.start()
/** * 连接一个服务器，并指定处理接收到的消息的处理方法 * */ public void start() { // this.context.put("resultMgr", this.resultMgr); logger.info(com.youxigu.dynasty2.i18n.MarkupMessages .getString("WolfClient_9"), processorNum); logger.info(com.youxigu.dynasty2.i18n.MarkupMessages .getString("WolfClient_0"), corePoolSize); logger.info(com.youxigu.dynasty2.i18n.MarkupMessages .getString("WolfClient_4"), maxPoolSize); if (this.serverIp == null || this.serverIp.equals("")) { logger.error(clientName + "没有配置serverIp,不启动........."); return; } String threadPrefix = clientName + "[" + this.serverIp + ":" + this.serverPort + "]"; // exector = Executors.newCachedThreadPool(new // NamingThreadFactory(threadPrefix)); processor = new SimpleIoProcessorPool<NioSession>(NioProcessor.class, processorNum); // connector = new NioSocketConnector((Executor) exector, processor); connector = new NioSocketConnector(processor); // connector.getSessionConfig().setReuseAddress(true); DefaultIoFilterChainBuilder chain = connector.getFilterChain(); if (useLogFilter == 2) { chain.addLast("logging", new LoggingFilter()); } // codec filter要放在ExecutorFilter前，因为读写同一个socket connection的socket // buf不能并发（事实上主要是读，写操作mina已经封装成一个write Queue) chain.addLast("codec", new ProtocolCodecFilter(codecFactory)); // 设置编码过滤器 // 添加心跳过滤器,客户端只接受服务端的心跳请求，不发送心跳请求 // connector.getSessionConfig().setReaderIdleTime(readIdleTimeOut); // 这里的KeepAliveFilter必须在codec之后，因为KeepAliveMessageFactoryImpl返回的是Object,如果KeepAliveMessageFactoryImpl返回的是IOBuffer,则可以在codec之前 // KeepAliveFilter到底在ExecutorFilter之前好还是之后好，我也不确定 KeepAliveFilter filter = new KeepAliveFilter( new KeepAliveMessageFactoryImpl(keepAliveRequestInterval <= 0), IdleStatus.READER_IDLE, new RequestTimeoutCloseHandler(), keepAliveRequestInterval <= 0 ? 600 : keepAliveRequestInterval, 30); chain.addLast("ping", filter); // 添加执行线程池 executor = new UnorderedThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, new NamingThreadFactory( threadPrefix)); // 这里是预先启动corePoolSize个处理线程 executor.prestartAllCoreThreads(); chain.addLast("exec", new ExecutorFilter(executor, IoEventType.EXCEPTION_CAUGHT, IoEventType.MESSAGE_RECEIVED, IoEventType.SESSION_CLOSED, IoEventType.SESSION_IDLE, IoEventType.SESSION_OPENED)); if (useWriteThreadPool) { executorWrite = new UnorderedThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, new NamingThreadFactory(threadPrefix + "write")); executorWrite.prestartAllCoreThreads(); chain.addLast("execWrite", new ExecutorFilter(executorWrite, IoEventType.WRITE, IoEventType.MESSAGE_SENT)); } // ,logger.isDebugEnabled() ? new // LoggingIoEventQueueHandler("execWrite") : nulls // 配置handler的 logger,在codec之后，打印的是decode前或者encode后的消息的log // 可以配置在ExecutorFilter之后：是为了在工作线程中打印log,不是在NioProcessor中打印 if (useLogFilter == 1) { chain.addLast("logging", new LoggingFilter()); } connector.setHandler(handler); connector.getSessionConfig().setReuseAddress(true); connector.getSessionConfig().setTcpNoDelay(tcpNoDelay); logger.info(com.youxigu.dynasty2.i18n.MarkupMessages .getString("WolfClient_1") + serverIp + ":" + serverPort); ConnectFuture cf = null; long start = System.currentTimeMillis(); while (true) { //这地很关键，是个无线循环，每10秒连接一次，直到可以和服务端建立连接，否则一支循环下去 cf = connector.connect(serverAddress);// 建立连接 cf.awaitUninterruptibly(10000L); if (!cf.isConnected()) { if ((System.currentTimeMillis() - start) > timeout) { throw new RuntimeException( com.youxigu.dynasty2.i18n.MarkupMessages .getString("WolfClient_5") + serverIp + ":" + serverPort); } if (cf.getException() != null) { logger.error(com.youxigu.dynasty2.i18n.MarkupMessages .getString("WolfClient_6"), serverIp + ":" + serverPort, cf.getException().getMessage()); } try { Thread.sleep(10000); } catch (Exception e) { } continue; } //这就是终极目标了，我们的目的就是在serevr的客户端的bean里，可以拿到这个iosession this.setSession(cf.getSession()); logger.info(com.youxigu.dynasty2.i18n.MarkupMessages .getString("WolfClient_10") + serverIp + ":" + serverPort); shutDown = false; if (handler instanceof WolfMessageChain) { WolfMessageChain wmc = WolfMessageChain.class.cast(handler); wmc.init(context); } break; } }
这样后端的业务通信网就可以轻松的建立起来，之后想怎么通信就看你的了
