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

Posted 2022-12-28 dgutfly

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

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

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

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

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

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

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

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

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

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

  1 package com.company.server;
  2 
  3 import io.netty.buffer.ByteBuf;
  4 import io.netty.buffer.Unpooled;
  5 import io.netty.channel.ChannelFuture;
  6 import io.netty.channel.ChannelFutureListener;
  7 import io.netty.channel.ChannelHandlerContext;
  8 import io.netty.channel.ChannelPromise;
  9 import io.netty.channel.SimpleChannelInboundHandler;
 10 import io.netty.handler.codec.http.DefaultFullHttpResponse;
 11 import io.netty.handler.codec.http.FullHttpRequest;
 12 import io.netty.handler.codec.http.FullHttpResponse;
 13 import io.netty.handler.codec.http.HttpHeaders;
 14 import io.netty.handler.codec.http.HttpResponseStatus;
 15 import io.netty.handler.codec.http.HttpVersion;
 16 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
 17 import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
 18 import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
 19 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 20 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 21 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
 22 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
 23 import io.netty.util.CharsetUtil;
 24 
 25 import org.apache.log4j.Logger;
 26 
 27 import com.company.serviceimpl.BananaService;
 28 import com.company.util.BufToString;
 29 import com.company.util.CODE;
 30 import com.company.util.Request;
 31 import com.company.util.Response;
 32 import com.google.common.base.Strings;
 33 import com.google.gson.JsonSyntaxException;
 34 
 35 
 36 /**
 37  * WebSocket服务端Handler
 38  *
 39  */
 40 public class BananaWebSocketServerHandler extends SimpleChannelInboundHandler<Object> 
 41     private static final Logger LOG = Logger.getLogger(BananaWebSocketServerHandler.class.getName());
 42     
 43     private WebSocketServerHandshaker handshaker;
 44     private ChannelHandlerContext ctx;
 45     private String sessionId;
 46     private boolean isLog = true;
 47     
 48     public BananaWebSocketServerHandler() 
 49         super();
 50     
 51     
 52     public BananaWebSocketServerHandler(boolean isLog) 
 53         this();
 54         this.isLog = isLog;
 55     
 56 
 57     //netty 5的覆写函数，netty4中用channelRead0代替
 58     public void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception 
 59         if(this.isLog) 
 60             System.out.print("channel MessageReceived = = " + ctx.name());
 61         
 62         if (msg instanceof FullHttpRequest)  // 传统的HTTP接入
 63             FullHttpRequest mymsg = (FullHttpRequest) msg;
 64             System.out.println(" with http request : " + BufToString.convertByteBufToString(mymsg.content()));
 65             handleHttpRequest(ctx, mymsg);
 66          else if (msg instanceof WebSocketFrame)  // WebSocket接入
 67             WebSocketFrame mymsg = (WebSocketFrame) msg;
 68             System.out.println(" with socket request : " + BufToString.convertByteBufToString(mymsg.content()));
 69             handleWebSocketFrame(ctx, mymsg);
 70         
 71     
 72     
 73     @Override
 74     public void handlerAdded(ChannelHandlerContext ctx) throws Exception
 75         System.out.println("channel handlerAdded = = " + ctx.name());
 76     
 77     
 78     @Override
 79     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception
 80         System.out.println("channel handlerRemoved = = " + ctx.name());
 81     
 82     
 83     @Override
 84     protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception 
 85         if(this.isLog) 
 86             System.out.print("channel Read0 = = " + ctx.name());
 87         
 88         if (msg instanceof FullHttpRequest)  // 传统的HTTP接入
 89             FullHttpRequest mymsg = (FullHttpRequest) msg;
 90             System.out.println(" with http request : " + BufToString.convertByteBufToString(mymsg.content()));
 91             handleHttpRequest(ctx, mymsg);
 92          else if (msg instanceof WebSocketFrame)  // WebSocket接入
 93             WebSocketFrame mymsg = (WebSocketFrame) msg;
 94             System.out.println(" with socket request : " + BufToString.convertByteBufToString(mymsg.content()));
 95             handleWebSocketFrame(ctx, mymsg);
 96         
 97     
 98     
 99     @Override
100     public void channelInactive(ChannelHandlerContext ctx) 
101         if(this.isLog) 
102             System.out.println("channel Inactive = = " + ctx.name());
103         
104         try 
105             this.close(ctx, null);
106          catch (Exception e) 
107             e.printStackTrace();
108         
109     
110     
111     @Override
112     public void channelUnregistered(ChannelHandlerContext ctx) 
113         if(this.isLog) 
114             System.out.println("channel Unregistered = = " + ctx.name());
115         
116     
117 
118     @Override
119     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception 
120         ctx.flush();
121         System.out.println("channel Flush = = " + ctx.name());
122     
123     
124     @Override
125     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception 
126 
127         ctx.close();
128         if(this.isLog) 
129             System.err.println("channel exceptionCaught = = " + ctx.name());
130             cause.printStackTrace();
131         
132         BananaService.logout(sessionId); // 注销
133         BananaService.notifyDownline(sessionId); // 通知有人下线
134     
135 
136     //netty 5的覆写函数，netty4中用channelInactive代替
137     public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception 
138         if(this.isLog) 
139             System.out.println("channel close = = " + ctx.name());
140         
141         BananaService.logout(sessionId); // 注销
142         BananaService.notifyDownline(sessionId); // 通知有人下线
143         ctx.close();
144     
145 
146     /**
147      * 处理Http请求，完成WebSocket握手<br/>
148      * 注意：WebSocket连接第一次请求使用的是Http
149      * @param ctx
150      * @param request
151      * @throws Exception
152      */
153     private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception 
154         // 如果HTTP解码失败，返回HHTP异常
155         if (!request.getDecoderResult().isSuccess() || (!"websocket".equals(request.headers().get("Upgrade")))) 
156             sendHttpResponse(ctx, request, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
157             return;
158         
159 
160         // 正常WebSocket的Http连接请求，构造握手响应返回
161         WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://" + request.headers().get(HttpHeaders.Names.HOST), null, false);
162         handshaker = wsFactory.newHandshaker(request);
163         if (handshaker == null)  // 无法处理的websocket版本
164             WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
165          else  // 向客户端发送websocket握手,完成握手
166             handshaker.handshake(ctx.channel(), request);
167             // 记录管道处理上下文，便于服务器推送数据到客户端
168             this.ctx = ctx;
169         
170     
171 
172     /**
173      * 处理Socket请求
174      * @param ctx
175      * @param frame
176      * @throws Exception 
177      */
178     private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception 
179         // 判断是否是关闭链路的指令
180         if (frame instanceof CloseWebSocketFrame) 
181             handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
182             return;
183         
184         // 判断是否是Ping消息
185         if (frame instanceof PingWebSocketFrame) 
186             ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
187             return;
188         
189         // 当前只支持文本消息，不支持二进制消息
190         if (!(frame instanceof TextWebSocketFrame)) 
191             throw new UnsupportedOperationException("当前只支持文本消息，不支持二进制消息");
192         
193         
194         // 处理来自客户端的WebSocket请求
195         try 
196             /*
197             if(this.isLog) 
198                 System.out.println("handleWebSocketFrame-=-=-" + ((TextWebSocketFrame)frame).text());
199             
200             */
201             Request request = Request.create(((TextWebSocketFrame)frame).text());
202             Response response = new Response();
203             response.setServiceId(request.getServiceId());
204             if (CODE.online.code.intValue() == request.getServiceId())  // 客户端注册
205                 String requestId = request.getRequestId();
206                 if (Strings.isNullOrEmpty(requestId)) 
207                     response.setIsSucc(false).setMessage("requestId不能为空");
208                     return;
209                  else if (Strings.isNullOrEmpty(request.getName())) 
210                     response.setIsSucc(false).setMessage("name不能为空");
211                     return;
212                  else if (BananaService.bananaWatchMap.containsKey(requestId)) 
213                     response.setIsSucc(false).setMessage("您已经注册了，不能重复注册");
214                     return;
215                 
216                 if (!BananaService.register(requestId, new BananaService(ctx, request.getName()))) 
217                     response.setIsSucc(false).setMessage("注册失败");
218                  else 
219                     response.setIsSucc(true).setMessage("注册成功");
220                     
221                     BananaService.bananaWatchMap.forEach((reqId, callBack) -> 
222                         response.getHadOnline().put(reqId, ((BananaService)callBack).getName()); // 将已经上线的人员返回
223                         
224                         if (!reqId.equals(requestId)) 
225                             Request serviceRequest = new Request();
226                             serviceRequest.setServiceId(CODE.online.code);
227                             serviceRequest.setRequestId(requestId);
228                             serviceRequest.setName(request.getName());
229                             try 
230                                 callBack.send(serviceRequest); // 通知有人上线
231                              catch (Exception e) 
232                                 LOG.warn("回调发送消息给客户端异常", e);
233                             
234                         
235                     );
236                 
237                 sendWebSocket(response.toJson());
238                 this.sessionId = requestId; // 记录会话id，当页面刷新或浏览器关闭时，注销掉此链路
239              else if (CODE.send_message.code.intValue() == request.getServiceId())  // 客户端发送消息到聊天群
240                 String requestId = request.getRequestId();
241                 if (Strings.isNullOrEmpty(requestId)) 
242                     response.setIsSucc(false).setMessage("requestId不能为空");
243                  else if (Strings.isNullOrEmpty(request.getName())) 
244                     response.setIsSucc(false).setMessage("name不能为空");
245                  else if (Strings.isNullOrEmpty(request.getMessage())) 
246                     response.setIsSucc(false).setMessage("message不能为空");
247                  else 
248                     response.setIsSucc(true).setMessage("发送消息成功");
249                     
250                     BananaService.bananaWatchMap.forEach((reqId, callBack) ->  // 将消息发送到所有机器
251                         Request serviceRequest = new Request();
252                         serviceRequest.setServiceId(CODE.receive_message.code);
253                         serviceRequest.setRequestId(requestId);
254                         serviceRequest.setName(request.getName());
255                         serviceRequest.setMessage(request.getMessage());
256                         try 
257                             callBack.send(serviceRequest);
258                          catch (Exception e) 
259                             LOG.warn("回调发送消息给客户端异常", e);
260                         
261                     );
262                 
263                 sendWebSocket(response.toJson());
264              else if (CODE.downline.code.intValue() == request.getServiceId())  // 客户端下线
265                 String requestId = request.getRequestId();
266                 if (Strings.isNullOrEmpty(requestId)) 
267                     sendWebSocket(response.setIsSucc(false).setMessage("requestId不能为空").toJson());
268                  else 
269                     BananaService.logout(requestId);
270                     response.setIsSucc(true).setMessage("下线成功");
271                     
272                     BananaService.notifyDownline(requestId); // 通知有人下线
273                     
274                     sendWebSocket(response.toJson());
275                 
276                 
277              else 
278                 sendWebSocket(response.setIsSucc(false).setMessage("未知请求").toJson());
279             
280          catch (JsonSyntaxException e1) 
281             LOG.warn("Json解析异常", e1);
282             System.err.println("Json解析异常");
283             e1.printStackTrace();
284          catch (Exception e2) 
285             LOG.error("处理Socket请求异常", e2);
286             System.err.println("处理Socket请求异常");
287             e2.printStackTrace();
288         
289     
290 
291     /**
292      * Http返回
293      * @param ctx
294      * @param request
295      * @param response
296      */
297     private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest request, FullHttpResponse response) 
298         // 返回应答给客户端
299         if (response.getStatus().code() != 200) 
300             ByteBuf buf = Unpooled.copiedBuffer(response.getStatus().toString(), CharsetUtil.UTF_8);
301             response.content().writeBytes(buf);
302             buf.release();
303             HttpHeaders.setContentLength(response, response.content().readableBytes());
304         
305 
306         // 如果是非Keep-Alive，关闭连接
307         ChannelFuture f = ctx.channel().writeAndFlush(response);
308         if (!HttpHeaders.isKeepAlive(request) || response.getStatus().code() != 200) 
309             f.addListener(ChannelFutureListener.CLOSE);
310         
311     
312     
313     /**
314      * WebSocket返回
315      * @param ctx
316      * @param req
317      * @param res
318      */
319     public void sendWebSocket(String msg) throws Exception 
320         if (this.handshaker == null || this.ctx == null || this.ctx.isRemoved()) 
321             throw new Exception("尚未握手成功，无法向客户端发送WebSocket消息");
322         
323         this.ctx.channel().write(new TextWebSocketFrame(msg));
324         this.ctx.flush();
325     
326 
327 

View Code

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

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

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

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

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

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