Fun论设计模式之7：中介者模式（Mediator Pattern）

　　意图：用一个中介对象来封装一系列的对象交互，中介者使各对象不需要显式地相互引用，从而使其耦合松散，而且可以独立地改变它们之间的交互。

　　主要解决：对象与对象之间存在大量的关联关系，这样势必会导致系统的结构变得很复杂，同时若一个对象发生改变，我们也需要跟踪与之相关联的对象，同时做出相应的处理。

　　何时使用：多个类相互耦合，形成了网状结构。

　　如何解决：将上述网状结构分离为星型结构。

　　关键代码：对象 Colleague 之间的通信封装到一个类中单独处理。

　　应用实例： 1、中国加入 WTO 之前是各个国家相互贸易，结构复杂，现在是各个国家通过 WTO 来互相贸易。 2、机场调度系统。 3、MVC 框架，其中C（控制器）就是 M（模型）和 V（视图）的中介者。

　　中介者就是管理不同对象的互相通信的对象。把其他对象不会用或者容易出错的操作移交中介者是比较正确的选择。

　　例如下文的聊天室，就是必须使用中介者模式解决的。

　　之前在项目中设计了基于netty的websocket聊天室，在leonzm的websocket_demo项目的基础上完善的系统，就是使用了中介者模式。

　　这里的“Colleague”实际上是web客户端的websocket请求，因为web客户端确实如意图所说，没有干预到服务器处理聊天的过程。为什么服务器内部重写的ChannelInboundHandler不是呢？在程序执行过程中，这个handler几乎是原原本本把数据传递过来的，但是里面也有对数据的处理及返回，实质上是和ServerBootstrap一起的，处理web客户端请求并返回结果的“中介者”。

package com.company.server;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.CharsetUtil;

import org.apache.log4j.Logger;

import com.company.serviceimpl.BananaService;
import com.company.util.BufToString;
import com.company.util.CODE;
import com.company.util.Request;
import com.company.util.Response;
import com.google.common.base.Strings;
import com.google.gson.JsonSyntaxException;


/**
 * WebSocket服务端Handler
 *
 */
public class BananaWebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
    private static final Logger LOG = Logger.getLogger(BananaWebSocketServerHandler.class.getName());
    
    private WebSocketServerHandshaker handshaker;
    private ChannelHandlerContext ctx;
    private String sessionId;
    private boolean isLog = true;
    
    public BananaWebSocketServerHandler() {
        super();
    }
    
    public BananaWebSocketServerHandler(boolean isLog) {
        this();
        this.isLog = isLog;
    }

    //netty 5的覆写函数，netty4中用channelRead0代替
    public void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(this.isLog) {
            System.out.print("channel MessageReceived = = " + ctx.name());
        }
        if (msg instanceof FullHttpRequest) { // 传统的HTTP接入
            FullHttpRequest mymsg = (FullHttpRequest) msg;
            System.out.println(" with http request : " + BufToString.convertByteBufToString(mymsg.content()));
            handleHttpRequest(ctx, mymsg);
        } else if (msg instanceof WebSocketFrame) { // WebSocket接入
            WebSocketFrame mymsg = (WebSocketFrame) msg;
            System.out.println(" with socket request : " + BufToString.convertByteBufToString(mymsg.content()));
            handleWebSocketFrame(ctx, mymsg);
        }
    }
    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception{
        System.out.println("channel handlerAdded = = " + ctx.name());
    }
    
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception{
        System.out.println("channel handlerRemoved = = " + ctx.name());
    }
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(this.isLog) {
            System.out.print("channel Read0 = = " + ctx.name());
        }
        if (msg instanceof FullHttpRequest) { // 传统的HTTP接入
            FullHttpRequest mymsg = (FullHttpRequest) msg;
            System.out.println(" with http request : " + BufToString.convertByteBufToString(mymsg.content()));
            handleHttpRequest(ctx, mymsg);
        } else if (msg instanceof WebSocketFrame) { // WebSocket接入
            WebSocketFrame mymsg = (WebSocketFrame) msg;
            System.out.println(" with socket request : " + BufToString.convertByteBufToString(mymsg.content()));
            handleWebSocketFrame(ctx, mymsg);
        }
    }
    
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        if(this.isLog) {
            System.out.println("channel Inactive = = " + ctx.name());
        }
        try {
            this.close(ctx, null);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) {
        if(this.isLog) {
            System.out.println("channel Unregistered = = " + ctx.name());
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
        System.out.println("channel Flush = = " + ctx.name());
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        ctx.close();
        if(this.isLog) {
            System.err.println("channel exceptionCaught = = " + ctx.name());
            cause.printStackTrace();
        }
        BananaService.logout(sessionId); // 注销
        BananaService.notifyDownline(sessionId); // 通知有人下线
    }

    //netty 5的覆写函数，netty4中用channelInactive代替
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        if(this.isLog) {
            System.out.println("channel close = = " + ctx.name());
        }
        BananaService.logout(sessionId); // 注销
        BananaService.notifyDownline(sessionId); // 通知有人下线
        ctx.close();
    }

    /**
     * 处理Http请求，完成WebSocket握手<br/>
     * 注意：WebSocket连接第一次请求使用的是Http
     * @param ctx
     * @param request
     * @throws Exception
     */
    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
        // 如果HTTP解码失败，返回HHTP异常
        if (!request.getDecoderResult().isSuccess() || (!"websocket".equals(request.headers().get("Upgrade")))) {
            sendHttpResponse(ctx, request, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }

        // 正常WebSocket的Http连接请求，构造握手响应返回
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://" + request.headers().get(HttpHeaders.Names.HOST), null, false);
        handshaker = wsFactory.newHandshaker(request);
        if (handshaker == null) { // 无法处理的websocket版本
            WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
        } else { // 向客户端发送websocket握手,完成握手
            handshaker.handshake(ctx.channel(), request);
            // 记录管道处理上下文，便于服务器推送数据到客户端
            this.ctx = ctx;
        }
    }

    /**
     * 处理Socket请求
     * @param ctx
     * @param frame
     * @throws Exception 
     */
    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {
        // 判断是否是关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        // 判断是否是Ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        // 当前只支持文本消息，不支持二进制消息
        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException("当前只支持文本消息，不支持二进制消息");
        }
        
        // 处理来自客户端的WebSocket请求
        try {
            /*
            if(this.isLog) {
                System.out.println("handleWebSocketFrame-=-=-" + ((TextWebSocketFrame)frame).text());
            }
            */
            Request request = Request.create(((TextWebSocketFrame)frame).text());
            Response response = new Response();
            response.setServiceId(request.getServiceId());
            if (CODE.online.code.intValue() == request.getServiceId()) { // 客户端注册
                String requestId = request.getRequestId();
                if (Strings.isNullOrEmpty(requestId)) {
                    response.setIsSucc(false).setMessage("requestId不能为空");
                    return;
                } else if (Strings.isNullOrEmpty(request.getName())) {
                    response.setIsSucc(false).setMessage("name不能为空");
                    return;
                } else if (BananaService.bananaWatchMap.containsKey(requestId)) {
                    response.setIsSucc(false).setMessage("您已经注册了，不能重复注册");
                    return;
                }
                if (!BananaService.register(requestId, new BananaService(ctx, request.getName()))) {
                    response.setIsSucc(false).setMessage("注册失败");
                } else {
                    response.setIsSucc(true).setMessage("注册成功");
                    
                    BananaService.bananaWatchMap.forEach((reqId, callBack) -> {
                        response.getHadOnline().put(reqId, ((BananaService)callBack).getName()); // 将已经上线的人员返回
                        
                        if (!reqId.equals(requestId)) {
                            Request serviceRequest = new Request();
                            serviceRequest.setServiceId(CODE.online.code);
                            serviceRequest.setRequestId(requestId);
                            serviceRequest.setName(request.getName());
                            try {
                                callBack.send(serviceRequest); // 通知有人上线
                            } catch (Exception e) {
                                LOG.warn("回调发送消息给客户端异常", e);
                            }
                        }
                    });
                }
                sendWebSocket(response.toJson());
                this.sessionId = requestId; // 记录会话id，当页面刷新或浏览器关闭时，注销掉此链路
            } else if (CODE.send_message.code.intValue() == request.getServiceId()) { // 客户端发送消息到聊天群
                String requestId = request.getRequestId();
                if (Strings.isNullOrEmpty(requestId)) {
                    response.setIsSucc(false).setMessage("requestId不能为空");
                } else if (Strings.isNullOrEmpty(request.getName())) {
                    response.setIsSucc(false).setMessage("name不能为空");
                } else if (Strings.isNullOrEmpty(request.getMessage())) {
                    response.setIsSucc(false).setMessage("message不能为空");
                } else {
                    response.setIsSucc(true).setMessage("发送消息成功");
                    
                    BananaService.bananaWatchMap.forEach((reqId, callBack) -> { // 将消息发送到所有机器
                        Request serviceRequest = new Request();
                        serviceRequest.setServiceId(CODE.receive_message.code);
                        serviceRequest.setRequestId(requestId);
                        serviceRequest.setName(request.getName());
                        serviceRequest.setMessage(request.getMessage());
                        try {
                            callBack.send(serviceRequest);
                        } catch (Exception e) {
                            LOG.warn("回调发送消息给客户端异常", e);
                        }
                    });
                }
                sendWebSocket(response.toJson());
            } else if (CODE.downline.code.intValue() == request.getServiceId()) { // 客户端下线
                String requestId = request.getRequestId();
                if (Strings.isNullOrEmpty(requestId)) {
                    sendWebSocket(response.setIsSucc(false).setMessage("requestId不能为空").toJson());
                } else {
                    BananaService.logout(requestId);
                    response.setIsSucc(true).setMessage("下线成功");
                    
                    BananaService.notifyDownline(requestId); // 通知有人下线
                    
                    sendWebSocket(response.toJson());
                }
                
            } else {
                sendWebSocket(response.setIsSucc(false).setMessage("未知请求").toJson());
            }
        } catch (JsonSyntaxException e1) {
            LOG.warn("Json解析异常", e1);
            System.err.println("Json解析异常");
            e1.printStackTrace();
        } catch (Exception e2) {
            LOG.error("处理Socket请求异常", e2);
            System.err.println("处理Socket请求异常");
            e2.printStackTrace();
        }
    }

    /**
     * Http返回
     * @param ctx
     * @param request
     * @param response
     */
    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest request, FullHttpResponse response) {
        // 返回应答给客户端
        if (response.getStatus().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(response.getStatus().toString(), CharsetUtil.UTF_8);
            response.content().writeBytes(buf);
            buf.release();
            HttpHeaders.setContentLength(response, response.content().readableBytes());
        }

        // 如果是非Keep-Alive，关闭连接
        ChannelFuture f = ctx.channel().writeAndFlush(response);
        if (!HttpHeaders.isKeepAlive(request) || response.getStatus().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }
    
    /**
     * WebSocket返回
     * @param ctx
     * @param req
     * @param res
     */
    public void sendWebSocket(String msg) throws Exception {
        if (this.handshaker == null || this.ctx == null || this.ctx.isRemoved()) {
            throw new Exception("尚未握手成功，无法向客户端发送WebSocket消息");
        }
        this.ctx.channel().write(new TextWebSocketFrame(msg));
        this.ctx.flush();
    }

}

　　这里也可以看出，handler只是在channelRead0（netty5是messageReceived）执行时，接到客户端的数据，往后都是对数据的处理返回。

　　这里还有channelInactive这个函数，监听客户端连接中断事件，当此类事件发生，或者客户端主动中断，服务器需要对使用这个聊天服务的所有客户端做处理（主要解决）。

　　优点： 1、降低了类的复杂度，将一对多转化成了一对一。 2、各个类之间的解耦。 3、符合迪米特原则。

　　缺点：中介者会庞大，变得复杂难以维护。（只能在中介者内部简化函数，或者适时使用其他设计模式优化）

　　使用场景： 1、系统中对象之间存在比较复杂的引用关系，导致它们之间的依赖关系结构混乱而且难以复用该对象。 2、想通过一个中间类来封装多个类中的行为，而又不想生成太多的子类。

　　注意事项：不应当在职责混乱的时候使用。（像这个系统，如果还涉及到在线好友的增减，一定要把好友服务写成独立于聊天服务的业务，不要让聊天服务的代码充斥if...else分支，导致代码可维护性下降）