【netty】java如何作为websocket客户端对服务端发起请求

分手后的思念是犯贱 2024-03-25 13:38 122阅读 0赞

#### 文章目录 ####

*  前言
 *  代码

> 是的 本文介绍java如何作为客户端 发起websocket请求  
> 博主不做标题党 不会服务端客户端分不清就写个标题 乱写文章

## 前言 ##

为什么会使用java作为websocket客户端？ 虽说websocket协议 本意是web与服务端之间的通讯协议，那假设有一天 我们的供应商 或者是甲方大爷 只提供了websocket接口呢？  
如果直接让前端去对接，再把数据传给后端，那从前端对接到后端入库的步骤，万一出现数据丢失呢？ 总之把数据处理放在后端，是相对可靠的， 我们可以借助netty来实现websocket客户端功能

长链接有两点值得注意的，**一是心跳机制 二是重连机制**  
如果不发送心跳包，可能过会儿连接就断开了；  
重新机制就比较好理解了，不管是服务端还是客户端的断开，作为客户端都需要能够重连

## 代码 ##

博主对长链接并不是特别熟练，但是什么代码是能正式用的，什么代码上不了生产只是个写着玩的demo 还是一眼能分辨出来的，代码主要参考 `git@github.com:yimiancheng/netty-study.git` ，写的很不错，代码优化空间很少了， 线程池博主是推荐用new ThreadPoolExecutor()创建，避免OOM的问题，除此之外应该是准生产环境级别代码了。

maven依赖

<dependency>
                   <groupId>io.netty</groupId>
                   <artifactId>netty-all</artifactId>
                   <version>4.1.35.Final</version>
               </dependency>

websocket客户端handle类 主要处理接收的消息、事件等，  
事件触发会进入userEventTriggered方法  
建立连接会进入channelActive方法  
接收消息会进入channelRead0方法  
出现异常会进入exceptionCaught方法  
断开连接会进入channelInactive方法（本文中未重写该方法， 在channelInactive方法中重连也是没问题的）

import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.SocketClient;
    import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.websocket.SendMsg;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelPromise;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
    import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
    import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler;
    import io.netty.handler.codec.http.websocketx.WebSocketFrame;
    import io.netty.handler.timeout.IdleStateEvent;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * WebSocketClientFrameHandler
     *
     * @date 2019/8/20 16:42
     */
    public class WebSocketClientFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
        
        private static final Logger LOG = LoggerFactory.getLogger(WebSocketClientFrameHandler.class);
        private SocketClient socketClient;
        private ChannelPromise channelPromise;
    
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        
            LOG.info("客户端接收到事件 " + (evt.getClass()) + " | " + evt.toString());
    
            if(WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE.equals(evt)) {
        
                LOG.info(ctx.channel().id().asShortText() + " 握手完成！");
                socketClient.CHANNEL_IS_READY.set(true);
                channelPromise.setSuccess();
                // SendMsg.send(ctx.channel(),"客户端握手完成消息 -》服务器时间: " + System.currentTimeMillis());
            }
            else if(evt instanceof IdleStateEvent){
        
                //ctx.channel().writeAndFlush(new PingWebSocketFrame());
                IdleStateEvent evtIdle = (IdleStateEvent) evt;
                switch(evtIdle.state()) {
        
                    case WRITER_IDLE:
                        // SendMsg.send(ctx.channel(),"客户端 ping 消息 -》服务器时间: " + System.currentTimeMillis());
                        ctx.channel().writeAndFlush(new PingWebSocketFrame());
                    case READER_IDLE:
                        // SendMsg.send(ctx.channel(),"客户端 ping 消息 -》服务器时间: " + System.currentTimeMillis());
                        ctx.channel().writeAndFlush(new PingWebSocketFrame());
                    default:
                        break;
                }
            }
    
            super.userEventTriggered(ctx, evt);
        }
    
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        
            Channel channel = ctx.channel();
            LOG.info("打开连接 handlerAdded SUCCESS. | name = " +channel.id().asShortText());
            super.handlerAdded(ctx);
            channelPromise = ctx.newPromise();
        }
    
        /**
         * Channel 已经被注册到了EventLoop
         */
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        
            Channel channel = ctx.channel();
            LOG.info("注册成功 channelRegistered SUCCESS. | name = " +channel.id().asShortText());
            super.channelRegistered(ctx);
        }
    
        /**
         * Channel 处于活动状态（已经连接到它的远程节点）。它现在可以接收和发送数据了
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
            Channel channel = ctx.channel();
            LOG.info("活动状态 channelActive SUCCESS. | name = " +channel.id().asShortText());
            super.channelActive(ctx);
        }
    
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, WebSocketFrame webSocketFrame) throws
            Exception {
        
            if(webSocketFrame instanceof TextWebSocketFrame) {
        
                // String message = textWebSocketFrame.content().toString(Charset.forName("utf-8"));
                String message = ((TextWebSocketFrame) webSocketFrame).text();
                LOG.info("客户端接收到消息：" + message);
            }
            else {
        
                LOG.info("接收到消息类型" + (webSocketFrame.getClass().getName()));
            }
    
            SendMsg.sendPong(channelHandlerContext.channel());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        
            LOG.error("消息处理失败: " + cause.getMessage(), cause);
            ctx.close();
        }
    
        /**
         * Channel 已经被创建，但还未注册到EventLoop
         * 连接断开
         */
        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        
            Channel channel = ctx.channel();
            LOG.info("连接断开 channelUnregistered SUCCESS. | name = " +channel.id().asShortText());
            super.channelUnregistered(ctx);
            channelPromise = null;
        }
    
        public SocketClient getSocketClient() {
        
            return socketClient;
        }
    
        public void setSocketClient(SocketClient socketClient) {
        
            this.socketClient = socketClient;
        }
    
        public ChannelPromise getChannelPromise() {
        
            return channelPromise;
        }
    }

心跳任务类

import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.SocketClient;
    import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.websocket.SendMsg;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.TimerTask;
    
    /**
     * HeartBeatTimerTask
     *
     * @date 2019/9/2 16:24
     */
    public class HeartBeatTimerTask extends TimerTask {
        
        private static final Logger LOG = LoggerFactory.getLogger(ReconnectTimerTask.class);
        private SocketClient socketClient;
    
        public HeartBeatTimerTask(SocketClient socketClient) {
        
            this.socketClient = socketClient;
        }
    
        @Override
        public void run() {
        
            if(socketClient != null && socketClient.isValid()) {
        
                SendMsg.send(socketClient.getChannel(), "客户端心跳消息 => " + System.currentTimeMillis());
            }
        }
    }

重连任务类

import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.SocketClient;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.TimerTask;
    
    /**
     * ReconnectTimerTask
     *
     * @date 2019/9/2 16:03
     */
    public class ReconnectTimerTask extends TimerTask {
        
        private static final Logger LOG = LoggerFactory.getLogger(ReconnectTimerTask.class);
        private SocketClient socketClient;
    
        public ReconnectTimerTask(SocketClient socketClient) {
        
            this.socketClient = socketClient;
        }
    
        @Override
        public void run() {
        
            if(socketClient != null && !socketClient.isValid()) {
        
                LOG.info("=== 客户端重连 " + System.currentTimeMillis());
                socketClient.connect();
            }
        }
    }

自定义线程工厂类，主要是给线程重命名 便于维护、调试

import java.util.concurrent.ThreadFactory;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class NamedThreadFactory implements ThreadFactory {
        
        static final AtomicInteger poolNumber = new AtomicInteger(1);
        final AtomicInteger threadNumber = new AtomicInteger(1);
        final ThreadGroup group;
        final String prefix;
        final boolean isDaemon;
        final int priority;
    
        public NamedThreadFactory() {
        
            this("pool");
        }
    
        public NamedThreadFactory(String prefix) {
        
            this(prefix, false, Thread.NORM_PRIORITY);
        }
    
        public NamedThreadFactory(String prefix, boolean isDaemon, int priority) {
        
            SecurityManager sm = System.getSecurityManager();
            this.group = (sm != null) ? sm.getThreadGroup() : Thread.currentThread().getThreadGroup();
            this.prefix = prefix + "-" + poolNumber.getAndIncrement() + "-thread-";
            this.isDaemon = isDaemon;
            this.priority = priority;
        }
    
        public Thread newThread(Runnable r) {
        
            Thread thread = new Thread(group, r, prefix + threadNumber.getAndIncrement(), 0);
            thread.setDaemon(isDaemon);
            thread.setPriority(priority);
            return thread;
        }
    }

消息监听类，主要用于发送消息（包括心跳包）是否成功

import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelFutureListener;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    class CustomerChannelFutureListener implements ChannelFutureListener {
        
        private static final Logger LOG = LoggerFactory.getLogger(CustomerChannelFutureListener.class);
    
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
        
                // LOG.info(JSON.toJSONString(channelFuture));
                if(channelFuture.isDone() && channelFuture.isSuccess()){
        
                   // LOG.info("send success.");
                }
                else {
        
                    channelFuture.channel().close();
                    LOG.info("send error. cause = " + channelFuture.cause());
                    channelFuture.cause().printStackTrace();
                }
            }
        }

简单封装的消息发送类

import com.alibaba.fastjson.JSON;
    import io.netty.channel.Channel;
    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 org.apache.commons.collections.MapUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import java.util.Collections;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    /**
     * SendMsg
     *
     * @date 2019/8/30 18:02
     */
    public class SendMsg {
        
        private static final Logger LOG = LoggerFactory.getLogger(SendMsg.class);
        public static ConcurrentHashMap<String, Channel> ALL_CHANNEL = new ConcurrentHashMap<String, Channel>();
    
        public static void startSendMsg() {
        
            Thread thread = new Thread(new Runnable() {
        
                public void run() {
        
                    while(true) {
        
                        sendMsgTest();
                    }
                }
            });
    
            thread.start();
        }
    
        public static void sendMsgTest() {
        
            try {
        
                Map<String, Channel> map = Collections.unmodifiableMap(ALL_CHANNEL);
                LOG.info("map size = " + map.size());
                if(MapUtils.isEmpty(map)) {
        
                    Thread.sleep(2000);
                    return;
                }
    
                for(Map.Entry<String, Channel> entry : map.entrySet()) {
        
                    LOG.info("------------- key = " + entry.getKey());
                    send(entry.getValue(), "服务端发送消息 " + entry.getKey() + " | " + System.currentTimeMillis());
                }
    
                Thread.sleep(10000);
            }
            catch(Exception ex) {
        
                ex.printStackTrace();
            }
        }
    
        public static void put(Channel channel) {
        
            ALL_CHANNEL.put(channel.id().asShortText(), channel);
        }
    
        public static void remove(Channel channel) {
        
            ALL_CHANNEL.remove(channel.id().asShortText());
        }
    
        public static void send(Channel channel, Object msg) {
        
            final String textMsg = JSON.toJSONString(msg);
    
            if(channel != null && channel.isActive()) {
        
                TextWebSocketFrame frame = new TextWebSocketFrame(textMsg);
                channel.writeAndFlush(frame)
                    .addListener(new CustomerChannelFutureListener());
            }
            else {
        
                LOG.error("消息发送失败！ textMsg = " + textMsg);
            }
        }
    
        public static void sendPing(Channel channel) {
        
            if(channel != null && channel.isActive()) {
        
                channel.writeAndFlush(new PingWebSocketFrame())
                    .addListener(new CustomerChannelFutureListener());
            }
            else {
        
                LOG.error("消息发送失败！ ping");
            }
        }
    
        public static void sendPong(Channel channel) {
        
            if(channel != null && channel.isActive()) {
        
                channel.writeAndFlush(new PongWebSocketFrame())
                    .addListener(new CustomerChannelFutureListener());
            }
            else {
        
                LOG.error("消息发送失败！ pong");
            }
        }
    
    }

channel初始化

import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.handler.WebSocketClientFrameHandler;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.http.HttpClientCodec;
    import io.netty.handler.codec.http.HttpObjectAggregator;
    import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
    import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler;
    import io.netty.handler.timeout.IdleStateHandler;
    
    /**
     * ClientChannelInitializer
     *
     * @date 2019/8/31 16:05
     */
    public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
        
        private WebSocketClientHandshaker webSocketClientHandshaker;
        private WebSocketClientFrameHandler webSocketFrameHandler;
    
        public ClientChannelInitializer(WebSocketClientHandshaker webSocketClientHandshaker, WebSocketClientFrameHandler
            webSocketFrameHandler) {
        
            this.webSocketClientHandshaker = webSocketClientHandshaker;
            this.webSocketFrameHandler = webSocketFrameHandler;
        }
    
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
        
            ChannelPipeline pipeline = socketChannel.pipeline();
    
            pipeline.addLast(new HttpClientCodec());//Http协议编码解码器
            pipeline.addLast(new HttpObjectAggregator(65536));//聚合 HttpRequest
            pipeline.addLast(new IdleStateHandler(5, 10, 0));
            //会处理ping pong close消息
            pipeline.addLast(new WebSocketClientProtocolHandler(webSocketClientHandshaker,true));
            pipeline.addLast(webSocketFrameHandler);
        }
    }

客户端主类 : 添加工作组 , 启动websocket client , 建立连接等

import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.handler.WebSocketClientFrameHandler;
    import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.task.HeartBeatTimerTask;
    import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.task.ReconnectTimerTask;
    import com.qiuhuanhen.springroot.interfaces.websocket.websocketclient.thread.NamedThreadFactory;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.http.DefaultHttpHeaders;
    import io.netty.handler.codec.http.HttpHeaders;
    import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
    import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
    import io.netty.handler.codec.http.websocketx.WebSocketVersion;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.net.URI;
    import java.net.URISyntaxException;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.ScheduledThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicBoolean;
    
    /**
     * SocketClient
     *
     * @date 2019/8/20 17:53
     */
    public class SocketClient {
        
        private static final Logger LOG = LoggerFactory.getLogger(SocketClient.class);
    
        public static String CLIENT_VERSION = "client_version";
        private static final int DEFAULT_PORT = 80;
        /**
         * 长链重连间隔时间，单位s
         */
        public static long RECONNECT_INTERVAL = 10;
        /**
         * 长链心跳时间，单位s
         */
        public static long FETCH_PERIOD = 30;
    
        public static String host = "127.0.0.1";
        public static int port = 8585;
    
        private Channel channel = null;
        private NioEventLoopGroup nioEventLoopGroup;
    
        public AtomicBoolean CHANNEL_IS_READY = new AtomicBoolean(false);
    
        private ScheduledExecutorService RECONNECT_TIMER;
        private ScheduledExecutorService HEARTBEAT_TIMER;
        static final String URL = System.getProperty("url", "ws://127.0.0.1:8585/boot/imserver/1111");
        URI uri = new URI(URL);
    
        static {
        
            // RECONNECT_TIMER = Executors.newSingleThreadScheduledExecutor();
        }
    
    
        public SocketClient(URI uri) throws URISyntaxException {
        
            this.uri = uri;
        }
    
        private void start() {
        
            Bootstrap boot = new Bootstrap();
            nioEventLoopGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
    
            try {
        
                HttpHeaders httpHeaders = new DefaultHttpHeaders();
                httpHeaders.add(CLIENT_VERSION, 1);
    
                WebSocketClientHandshaker webSocketClientHandshaker = WebSocketClientHandshakerFactory.newHandshaker(uri,
                    WebSocketVersion.V13, null, false, httpHeaders);
    
                boot.group(nioEventLoopGroup)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .channel(NioSocketChannel.class);
    
                WebSocketClientFrameHandler webSocketFrameHandler = new WebSocketClientFrameHandler();
                webSocketFrameHandler.setSocketClient(this);
    
                ClientChannelInitializer clientChannelInitializer =
                    new ClientChannelInitializer(webSocketClientHandshaker, webSocketFrameHandler);
    
                boot.handler(new LoggingHandler(LogLevel.INFO));
                boot.handler(clientChannelInitializer);
    
                port = (uri.getPort() == -1) ? DEFAULT_PORT : uri.getPort();
                host = uri.getHost();
                channel = boot.connect(host, port).sync().channel();
                LOG.info("SocketClient has started. CHANNEL_IS_READY = " + CHANNEL_IS_READY.get());
                webSocketFrameHandler.getChannelPromise().sync();
                LOG.info("SocketClient has started full. CHANNEL_IS_READY = " + CHANNEL_IS_READY.get());
            }
            catch(Exception ex) {
        
                ex.printStackTrace();
                LOG.error("connect error. uri " + uri.toString());
            }
            finally {
        
                // nioEventLoopGroup.shutdownGracefully();
            }
        }
    
        /**
         * 客户端连接服务端
         */
        public void connect() {
        
            stop();
            start();
            startReconnect();
            doHeartBeat();
        }
    
        /**
         * 开启线程-断开重连
         */
        public void startReconnect() {
        
            RECONNECT_TIMER = new ScheduledThreadPoolExecutor(1,
                new NamedThreadFactory("reconnect-schedule-pool", Boolean.TRUE, Thread.NORM_PRIORITY));
    
            // https://www.jianshu.com/p/502f9952c09b
            RECONNECT_TIMER.scheduleAtFixedRate(new ReconnectTimerTask(this),
                RECONNECT_INTERVAL * 1000L, RECONNECT_INTERVAL * 1000L, TimeUnit.MILLISECONDS);
        }
    
        /**
         * 心跳
         */
        private void doHeartBeat() {
        
            HEARTBEAT_TIMER = new ScheduledThreadPoolExecutor(1,
                new NamedThreadFactory("heartbeat-schedule-pool", Boolean.TRUE, Thread.NORM_PRIORITY));
    
            // https://www.jianshu.com/p/502f9952c09b
            HEARTBEAT_TIMER.scheduleAtFixedRate(new HeartBeatTimerTask(this),
                FETCH_PERIOD * 1000L, FETCH_PERIOD * 1000L, TimeUnit.MILLISECONDS);
        }
    
        /**
         * 客户端停止
         */
        public void stop() {
        
            try {
        
                if(nioEventLoopGroup != null) {
        
                    nioEventLoopGroup.shutdownGracefully();
                }
    
                if(channel != null) {
        
                    channel.close();
                }
    
                if(RECONNECT_TIMER != null) {
        
                    RECONNECT_TIMER.shutdown();
                    RECONNECT_TIMER = null;
                }
    
                if(HEARTBEAT_TIMER != null) {
        
                    HEARTBEAT_TIMER.shutdown();
                    HEARTBEAT_TIMER = null;
                }
            }
            catch(Exception ex) {
        
                //do nothing.
            }
        }
    
    
        public boolean isValid() {
        
            if (channel != null && channel.isActive()) {
        
                return true;
            } else {
        
                return false;
            }
        }
    
        public Channel getChannel() {
        
            return channel;
        }
    
        public static void main(String[] args) {
        
            try {
        
                new SocketClient(new URI(URL)).connect();
            } catch (URISyntaxException e) {
        
                e.printStackTrace();
            }
        }
    }

重连成功截图：  
![在这里插入图片描述][67029ec64f4a4c8aa80ac0ef4e6aac99.png]

[67029ec64f4a4c8aa80ac0ef4e6aac99.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/1408632edc924720a54eedfa2883efbb.png