仿淘宝开放平台之消息服务——客户端处理链条设计与实现

2022-01-15 09:30:15 阅读：133 来源： 互联网

标签：开放平台消息处理器 new msg 链条 public channel 客户端

使用netty框架实现websocket客户端。

启动框架

相比服务端标准模式而言，客户端启动类有比较多需要注意的地方，关键的逻辑有两个：

一是客户端需要处理自动重连，这里实际是两种情况，一种是客户端刚启动的时候，尝试去连接服务端，如不成功，则休眠5秒后再次重试；另外一种是出现异常时，包括原先建立连接、正常通信的情况下因为各种原因导致通道失效、心跳异常、服务端退出等，都会自动尝试重连，这样可以确保出现问题时无需系统管理员手工干预，自动重连来恢复运行。
二是连接成功后，要发起WebSocket的握手操作，将http协议升级为websocket协议，关键在于自实现的WebSocketClientHandshakerHandler处理器。

@Slf4j
@Component
public class MessageClient {


    @Autowired
    private MessageClientGlobalHolder config;

    @Autowired
    private MessageClientChannelInitializer messageClientChannelInitializer;



    /**
     * 启动客户端方法
     */
    public void start() {

        EventLoopGroup workerGroup = new NioEventLoopGroup(1);
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(workerGroup);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.handler(messageClientChannelInitializer);


            //客户端与服务端连接的通道，final修饰表示只会有一个
            ChannelFuture channelFuture = bootstrap.connect(config.getHost(), config.getPort());
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (!future.isSuccess()) {
                        //未成功
                        log.error("连接失败", future.cause());
                        //执行重连
                        reconnect();
                    } else {
                        log.info("连接成功");
                        Channel channel = future.channel();
                        //将channel保存到全局变量
                        config.setChannel(channel);
                        //发起握手
                        WebSocketClientHandshakerHandler handler = (WebSocketClientHandshakerHandler) channel.pipeline().get("hookedHandler");
                        handler.handshake(config.getChannel());
                    }
                }
            });
            //等待服务器端关闭
            channelFuture.channel().closeFuture().sync();


        } catch (Exception e) {
            log.error("消息客户端启动失败:{}" + e.getMessage(), e);
            //执行重连
            reconnect();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

    /**
     * 重连
     */
    public void reconnect() {
        try {
            Thread.sleep(5000);
            //执行重连
            log.info("消息客户端进行重连");
            start();

        } catch (InterruptedException e) {
            log.error("消息客户端重连过程中线程休眠失败", e);
        }

    }

   
}

处理器配置

消息处理器的装配与实现关键实现


/**
 * 初始化通道
 *
 * @author wqliu
 * @date 2021-2-5 15:12
 */
@Slf4j
@Component
public class MessageClientChannelInitializer extends ChannelInitializer<SocketChannel> {


    @Autowired
    private MessageClientGlobalHolder config;

    @Autowired
    private Environment environment;

    /**
     * 生产运行模式
     */
    private final String PRD_MODE="prd";

    /**
     * 初始化channel
     */
    @Override
    public void initChannel(SocketChannel socketChannel) throws Exception {



        //获取通道链路
        ChannelPipeline pipeline = socketChannel.pipeline();

        //仅在生产模式下加载ssl过滤器
        String mode=environment.getProperty("spring.profiles.active");
        if(PRD_MODE.equals(mode)){
            //ssl
            SSLContext sslContext = createSslContext();
            SSLEngine engine = sslContext.createSSLEngine();
            engine.setNeedClientAuth(false);
            engine.setUseClientMode(false);
            pipeline.addLast(new SslHandler(engine));
        }


        //HTTP 编解码
        pipeline.addLast(new HttpClientCodec());

        // 聚合为单个 FullHttpRequest 或者 FullHttpResponse
        pipeline.addLast(new HttpObjectAggregator(64 * 1024));

        /**
         * 注意，因WebSocketClientHandshakerHandler继承自SimpleChannelInboundHandler，会自动释放消息
         * 对于收到服务端的pong消息，默认情况下不会往通道后续的处理器传递，所以若放到WebSocketClientHandshakerHandler之后，
         * 则会产生读空闲，导致心跳超时失效。
         */
        // 添加读写通道空闲处理器，当空闲满足设置时，会触发userEventTrigger，由下个处理器获取到
        pipeline.addLast(new IdleStateHandler(config.getReadIdleTimeOut(), 0,
                0, TimeUnit.SECONDS));

        //心跳超时处理
        pipeline.addLast(new HeartbeatTimeoutHandler());


        //处理web socket协议与握手
        pipeline.addLast("hookedHandler", new WebSocketClientHandshakerHandler());


        //心跳发送
        pipeline.addLast(new HeartbeatRequestHandler(config.getHeartbeatRate()));

        //将文本按消息类型转换为请求消息或响应消息
        pipeline.addLast(new MessageTypeDecodeHandler());

        //请求消息业务逻辑处理器
        pipeline.addLast(new RequestMessageBusinessHandler());

        //响应消息业务逻辑处理器
        pipeline.addLast(new ResponseMessageBusinessHandler());

        //编码为TextWebSocketFrame
        pipeline.addLast(new TextWebSocketFrameEncodeHandler());

        //json序列化
        pipeline.addLast(new JsonEncodeHandler());


    }

    /**
     * 创建ssl上下文对象
     * @param type
     * @param path
     * @param password
     * @return
     * @throws Exception
     */
    public SSLContext createSslContext() throws Exception {

        //读取配置信息
        String path=environment.getProperty("server.ssl.key-store");
        log.info("证书地址:{}",path);
        String password=environment.getProperty("server.ssl.key-store-password");
        String type=environment.getProperty("server.ssl.key-store-type");

        //构建证书上下文对象
        KeyStore ks = KeyStore.getInstance(type);
        path=path.replace("classpath:","");
        log.info("处理后的证书地址:{}",path);
        ClassPathResource resource = new ClassPathResource(path);
        InputStream ksInputStream = resource.getInputStream();
        ks.load(ksInputStream, password.toCharArray());
        //KeyManagerFactory充当基于密钥内容源的密钥管理器的工厂。
        KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
        kmf.init(ks, password.toCharArray());
        //SSLContext的实例表示安全套接字协议的实现，它充当用于安全套接字工厂或 SSLEngine 的工厂。
        SSLContext sslContext = SSLContext.getInstance("TLS");
        sslContext.init(kmf.getKeyManagers(), null, null);
        return sslContext;
    }


}

处理器清单

一共涉及到12个消息处理器，其中4个是netty内置的，只是进行了参数配置，其他8个是自己实现的，用于处理逻辑和数据的，依次如下：

序号	处理器类型	职责	实现	说明
1	SslHandler	处理可靠安全连接	内置	仅在生产环境，需要进行ssl加解密
2	HttpClientCodec	HTTP 编解码	内置	对Http请求进行解码与编码
3	HttpObjectAggregator	聚合HTTP 请求或响应	内置	将http请求或响应聚合为一个完整对象
4	IdleStateHandler	空闲监测	内置	监测空闲状态，触发后续超时处理
5	HeartbeatTimeoutHandler	心跳超时处理	自定义	心跳超时执行关闭连接，触发重连
6	WebSocketClientHandshakerHandler	WebSocket专用处理	自定义	处理WebSocket的握手以及Ping、Pong、Close消息
7	HeartbeatRequestHandler	发送心跳请求	自定义	客户端向服务端定时发送心跳
8	MessageTypeDecodeHandler	文本反序列化成消息对象	自定义	将文本按消息类型转换为请求消息或响应消息
9	RequestMessageBusinessHandler	处理请求消息	自定义	请求消息业务逻辑处理器
10	ResponseMessageBusinessHandler	处理响应消息	自定义	响应消息业务逻辑处理器
11	TextWebSocketFrameEncodeHandler	JSON格式转文本帧	自定义	将json格式字符串编码为TextWebSocketFrame
12	JsonEncodeHandler	对象序列化为JSON字符串	自定义	将对象序列化为json格式字符串

自定义的8个处理器中，最后两个11和12是出站处理器，注意实际执行顺序是先12后11，也就是，业务逻辑处理器9或10的处理结果是一个对象，先由出站处理器12将其序列化为json字符串，然后再由出站处理器11将其包装为一个WebSocket协议约定的文本帧TextWebSocketFrame。

自定义处理器

心跳超时处理

这个处理器实际是配合netty内置的空闲检测处理器IdleStateHandler使用的，只有满足IdleStateHandler中设置的触发条件，才会触发本处理器中的userEventTriggered方法，执行自定义的逻辑操作，这里是主动关闭连接。

客户端每隔固定时间频率向服务器端发送心跳，WebSocket协议约定的PingWebSocketFrame，服务端收到后马上会回复PongWebSocketFrame，如通道失效或服务端无响应情况下，就会触发客户端读空闲。


/**
 * 心跳超时处理器
 * @author wqliu
 * @date 2021-10-2 14:25
 **/
@Slf4j
public class HeartbeatTimeoutHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state().equals(IdleState.READER_IDLE)) {
                log.info("读空闲");
                //关闭连接
                ctx.channel().close();
            }
        } else {
            //非空闲事件，传递到下个处理器
            super.userEventTriggered(ctx, evt);
        }

    }


}

WebSocket专用处理

这是很关键的一个处理器，自身也比较复杂。

主要实现是借助netty提供的一个WebSocketClientHandshaker类，在初始化时设置websocket服务端连接信息，然后在客户端启动时，调用该类的发起握手方法handshake，服务器端收到该握手请求后，会进行后续处理，响应一个协议升级，将http协议升级为WebSocket协议。

同时需要注意的是，这里还有一个我们自定义的操作，即在握手成功，协议升级后，客户端发出一个登录服务端的请求消息。

/**
 * 处理web socket握手
 *
 * @author wqliu
 * @date 2021-9-28 16:33
 **/
@Slf4j
@Data
public class WebSocketClientHandshakerHandler extends SimpleChannelInboundHandler<Object> {
    /**
     * 握手
     */
    private WebSocketClientHandshaker handshaker;
    /**
     * 握手 异步处理
     */
    private ChannelPromise handshakeFuture;

    public WebSocketClientHandshakerHandler() {
        //初始化握手处理者
        MessageClientGlobalHolder config = SpringUtil.getBean(MessageClientGlobalHolder.class);
        URI webSocketUri = null;
        try {
            webSocketUri = new URI(config.getWebSocketUrl());
        } catch (URISyntaxException e) {
            log.error("解析远程服务器地址出错", e);
        }
        WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(webSocketUri, WebSocketVersion.V13, (String) null, true, new DefaultHttpHeaders());
        this.setHandshaker(handshaker);


    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        // log.info("收到消息：{}",msg.toString());
        Channel ch = ctx.channel();
        FullHttpResponse response;
        //进行握手操作
        if (!this.handshaker.isHandshakeComplete()) {
            try {
                response = (FullHttpResponse) msg;
                //握手协议返回，设置结束握手
                this.handshaker.finishHandshake(ch, response);
                //设置成功
                this.handshakeFuture.setSuccess();

            } catch (WebSocketHandshakeException var7) {
                 //已握手成功并将http协议升级为了WebSocket协议，不应再收到Http消息，发生这种情况则抛出异常
                FullHttpResponse res = (FullHttpResponse) msg;
                String errorMsg = String.format("握手失败,status:%s,reason:%s", res.status(), res.content().toString(CharsetUtil.UTF_8));
                this.handshakeFuture.setFailure(new Exception(errorMsg));
            }
        } else if (msg instanceof FullHttpResponse) {
            response = (FullHttpResponse) msg;
            throw new IllegalStateException("Unexpected FullHttpResponse (getStatus=" + response.status() + ", content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
        } else if (msg instanceof CloseWebSocketFrame) {
            log.info("收到关闭信息");

        } else if (msg instanceof TextWebSocketFrame) {
            log.info("收到文本帧，往下传递");
            ReferenceCountUtil.retain(msg);
            ctx.fireChannelRead(msg);
        }

    }


    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {

        //设置握手成功后，发起登录请求
        this.handshakeFuture = ctx.newPromise();
        ChannelFuture handshakeFuture = this.handshakeFuture;
        handshakeFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    //发送登录请求
                    log.info("握手成功");
                    Login login = new Login();
                    login.sendMessage();

                } else {
                    //握手失败
                    log.error("握手失败", future.cause());
                }
            }
        });


    }

    /**
     * 发起握手
     */
    public void handshake(Channel channel) {
        this.getHandshaker().handshake(channel);
    }


}

发送心跳请求

心跳机制是客户端每隔固定时间频率向服务器端发送心跳，WebSocket协议约定的PingWebSocketFrame

/**
 * 心跳请求处理器
 * @author wqliu
 * @date 2021-10-2 13:24
 **/
@Slf4j
public class HeartbeatRequestHandler extends ChannelInboundHandlerAdapter {

    /**
     * 心跳发送间隔，单位秒
     */
    private int heartbeatInterval=5;

    public HeartbeatRequestHandler(int heartbeatInterval){
        this.heartbeatInterval=heartbeatInterval;
    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        EventLoop eventLoop = ctx.channel().eventLoop();
        eventLoop.scheduleWithFixedDelay(new Runnable() {
            private Channel channel;
            @Override
            public void run() {
                // log.info("发送心跳");
                PingWebSocketFrame frame=new PingWebSocketFrame();
                ChannelFuture channelFuture = channel.writeAndFlush(frame);
                channelFuture.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        // log.error(future.isSuccess()+"",future.cause());
                    }
                });
            }

            public Runnable setChannel(Channel channel){
                this.channel=channel;
                return this;
            }

        }.setChannel(ctx.channel()),15,heartbeatInterval, TimeUnit.SECONDS);

        //不调用父类方法，则其他处理器的channelActive事件不再触发
        super.channelActive(ctx);
    }
}

文本反序列化成消息对象

我将消息设计为两类，请求消息和响应消息，这里通过自己实现的一个处理器，将客户端传来的文本帧，通过消息类型属性反序列化成请求消息对象或响应消息对象，这里调用的是公用的处理器，即服务端也使用相同的处理器。


/**
 * 消息类型解码
 * @author wqliu
 * @date 2021-10-6 11:23
 **/
public class MessageTypeDecodeHandler extends MessageToMessageDecoder<TextWebSocketFrame> {
    @Override
    protected void decode(ChannelHandlerContext ctx, TextWebSocketFrame msg, List<Object> out) throws Exception {
        String message=msg.text();
        //消息解析
        JSONObject jsonObject = JSONObject.parseObject(message);
        String messageType = jsonObject.getString("messageType");
        if (messageType.equals(MessageType.REQUEST.name())) {
            RequestMessage requestMessage = JSON.parseObject(message, RequestMessage.class);
            out.add(requestMessage);

        }else if (messageType.equals(MessageType.RESPONSE.name())) {

            ResponseMessage responseMessage = JSON.parseObject(message, ResponseMessage.class);
            out.add(responseMessage);
        }
    }

}

请求/响应消息处理器

上一步把消息内容通过解码形成了请求消息或响应消息，而这两个处理器只需加入到链条中即可，根据传入的消息类型，也就是泛型参数类型，会自动识别处理或者往下传递。

JSON格式转文本帧

这个其实也没什么好说的，就是把JSON格式字符串放到文本帧中，这里调用的是公用的处理器，即服务端也使用相同的处理器。

/**
 * 将json格式字符串编码为TextWebSocketFrame
 * @author wqliu
 * @date 2021-10-6 11:23
 **/

public class TextWebSocketFrameEncodeHandler extends MessageToMessageEncoder<String> {
    @Override
    protected void encode(ChannelHandlerContext ctx, String msg, List<Object> out) throws Exception {
        TextWebSocketFrame frame=new TextWebSocketFrame(msg);
        out.add(frame);

    }


}

对象序列化为JSON字符串

这个其实也没什么好说的，就是把对象转换为JSON格式字符串，这里调用的是公用的处理器，即服务端也使用相同的处理器。

/**
 * 将对象序列化为json格式字符串
 * @author wqliu
 * @date 2021-10-6 11:23
 **/
public class JsonEncodeHandler extends MessageToMessageEncoder<BaseMessage> {
    @Override
    protected void encode(ChannelHandlerContext ctx, BaseMessage msg, List<Object> out) throws Exception {

        if(msg instanceof BaseMessage) {
            out.add(JSONObject.toJSONString(msg));
        }else{
            out.add(msg);
        }
    }


}

标签：开放平台,消息,处理器,new,msg,链条,public,channel,客户端
来源： https://blog.csdn.net/seawaving/article/details/122498327

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9