Netty WebSocket长连接

## websocket客户端在线测试工具链接
注意点：websocket客户端在线测试工具，如你要测本机服务端的长连接，切记“ws://localhost:8888/ajaxchattest” 不要写本机的IP直接写localhost即可，另端口后面要加/ajaxchattest，否则总是报连接服务断开，我就是在此处踩了坑！！！

本人使用长连接的使用场景

业务场景：由于本人公司是做物业系统的，在业主的APP程序中创建访客信息，那么需求是则在物业平台系统中能自动弹出创建该访客的信息，所以前端WEB端要和后台服务端保存长链接，当访客添加信息后后台将此信息主动推给前端并展示！

Netty WebSocket Maven依赖

        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty</artifactId>
            <version>3.10.6.Final</version>
        </dependency>

Netty WebSocket 后台 Java代码

以下代码就可以实现websocket长连接功能

webscoket初始化启动类

/**
 * <desc>
 * webscoket初始化启动类.
 * </desc>
 *
 * @createDate 2018/7/19
 */
@RestController
public class WebsocketInitializationService {

    @Autowired
    WebSocketServer webSocketServer;

    /**
     * <desc>
     *      初始化
     * </desc>
     *
     * @author Juguang.S
     * @createDate 2018/07/19
     */
    @PostConstruct
    public void polling() throws Exception {
        webSocketServer.run();
    }
}

websocket长连接服务类

参数：
PROPERTY_VISITOR_PUSH_MESSAGE_URL 为服务端所部署服务器的IP
PROPERTY_VISITOR_PUSH_MESSAGE_PORT 为服务端所部署服务器的端口

/**
 * <desc>
 * websocket长连接服务类.
 * </desc>
 *
 * @createDate 2018/7/5
 */

@Service
public class WebSocketServer {

    @Value("${" + PlatformConstants.PROPERTY_VISITOR_PUSH_MESSAGE_URL + "}")
    private  String PROPERTY_VISITOR_PUSH_MESSAGE_URL;

    @Value("${" + PlatformConstants.PROPERTY_VISITOR_PUSH_MESSAGE_PORT + "}")
    private  String PROPERTY_VISITOR_PUSH_MESSAGE_PORT;

    private static Logger LOG = Logger.getLogger(WebSocketServer.class);
    static final boolean SSL = System.getProperty("ssl") != null;
    public void run() throws Exception {
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContext.newServerContext(ssc.certificate(), ssc.privateKey());
        } else {
            sslCtx = null;
        }

        ServerBootstrap bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(
                Executors.newCachedThreadPool(), Executors.newCachedThreadPool()));

        bootstrap.setPipelineFactory(new WebSocketServerPipelineFactory(sslCtx));

        bootstrap.bind(new InetSocketAddress(PROPERTY_VISITOR_PUSH_MESSAGE_URL, Integer.parseInt(PROPERTY_VISITOR_PUSH_MESSAGE_PORT)));
        }

}

保存Channel对应类

/**
 * <desc>
 * 保存Channel对应类.
 * </desc>
 *
 * @createDate 2018/7/5
 */
public class GatewayService {

    private static Map<String, Channel> map = new ConcurrentHashMap<>();

    public static void addGatewayChannel(String id, Channel gateway_channel){
        map.put(id, gateway_channel);
    }

    public static Map<String, Channel> getChannels(){
        return map;
    }

    public static Channel getGatewayChannel(String id){
        return map.get(id);
    }

    public static void removeGatewayChannel(String id){
        map.remove(id);
    }
}

websocket中处理业务逻辑类

/**
 * <desc>
 * websocket中处理业务逻辑类.
 * </desc>
 *
 * @createDate 2018/7/5
 */
public class WebSocketServerHandler extends SimpleChannelUpstreamHandler {

    private static final String WEBSOCKET_PATH = "/websocket";

    private WebSocketServerHandshaker handshaker;

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {

        Object msg = e.getMessage();
        //传统的HTTP接入
        if (msg instanceof HttpRequest) {
            handleHttpRequest(ctx, (HttpRequest) msg);
        }//WebSocket接入
         else if (msg instanceof WebSocketFrame) {
            handleWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }
    @Override
    public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        String uuid = ctx.getChannel().getId().toString();
        GatewayService.addGatewayChannel(uuid, ctx.getChannel());
        System.out.println(GatewayService.getGatewayChannel(uuid));
    }

    private void handleHttpRequest(ChannelHandlerContext ctx, HttpRequest req) {
        // Allow only GET methods.
        if (req.getMethod() != GET) {
            sendHttpResponse(ctx, req, new DefaultHttpResponse(HTTP_1_1, FORBIDDEN));
            return;
        }

        if ("/".equals(req.getUri())) {
            HttpResponse res = new DefaultHttpResponse(HTTP_1_1, OK);
            return;
        }
        if ("/favicon.ico".equals(req.getUri())) {
            HttpResponse res = new DefaultHttpResponse(HTTP_1_1, NOT_FOUND);
            sendHttpResponse(ctx, req, res);
            return;
        }

       //构造握手响应返回，本机测试
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                getWebSocketLocation(req), null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            wsFactory.sendUnsupportedWebSocketVersionResponse(ctx.getChannel());
        } else {
            handshaker.handshake(ctx.getChannel(), req).addListener(WebSocketServerHandshaker.HANDSHAKE_LISTENER);
        }
    }
    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        //判断是否关闭链路指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.getChannel(), (CloseWebSocketFrame) frame);
            return;
        }
        //判断是否是Ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.getChannel().write(new PongWebSocketFrame(frame.getBinaryData()));
            return;
        }
        //本例程仅支持文本信息，不支持二进制消息
        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(
                    String.format("%s frame types not supported", frame.getClass().getName()));
        }

       //返回应答消息
        String request = ((TextWebSocketFrame) frame).getText();
        System.err.println(String.format("Channel %s received %s", ctx.getChannel().getId(), request));
		// ctx.getChannel().write(new TextWebSocketFrame(request.toUpperCase()));
    }
    private static void sendHttpResponse(ChannelHandlerContext ctx, HttpRequest req, HttpResponse res) {
		//返回应答给客户端
        if (res.getStatus().getCode() != 200) {
            res.setContent(ChannelBuffers.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8));
            setContentLength(res, res.getContent().readableBytes());
        }
		//如果是非Keep-Alive，关闭连接
        ChannelFuture f = ctx.getChannel().write(res);
        if (!isKeepAlive(req) || res.getStatus().getCode() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
        e.getCause().printStackTrace();
        e.getChannel().close();
    }
    private static String getWebSocketLocation(HttpRequest req) {
        String location =  req.headers().get(HOST) + WEBSOCKET_PATH;
        if (WebSocketServer.SSL) {
            return "wss://" + location;
        } else {
            return "ws://" + location;
        }
    }
}

WebSocketServerPipelineFactory类

/**
 * <desc>
 * WebSocketServerPipelineFactory类.
 * </desc>
 *
 * @createDate 2018/7/5
 */
public class WebSocketServerPipelineFactory implements ChannelPipelineFactory {

    private final SslContext sslCtx;

    public WebSocketServerPipelineFactory(SslContext sslCtx) {
        this.sslCtx = sslCtx;
    }

    public ChannelPipeline getPipeline() {
        ChannelPipeline pipeline = Channels.pipeline();
        if (sslCtx != null) {
            pipeline.addLast("ssl", sslCtx.newHandler());
        }
        pipeline.addLast("decoder", new HttpRequestDecoder());
        pipeline.addLast("aggregator", new HttpChunkAggregator(65536));
        pipeline.addLast("encoder", new HttpResponseEncoder());
        pipeline.addLast("handler", new WebSocketServerHandler());
        return pipeline;
    }
}

推送业务逻辑触发类

/**
     * <desc>
     *     App用户添加访客后主动向前端推送消息
     * </desc>
     * @return
     * @createDate 2018/07/13
     */
    @Override
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRED)
    public void visitorPushMessage() throws Exception {
        System.out.println("task is beginning...");
        LOG.info("【定时任务】定时刷新访客表获取未推送的APP新增访客信息task is beginning...");
        try{
            Map<String, Channel> map = GatewayService.getChannels();
            Iterator<String> it = map.keySet().iterator();
            List<String> passId = new ArrayList<>();
            while (it.hasNext()) {
                String key = it.next();
                Channel obj = map.get(key);
                System.out.println("channel id is: " + key);
                LOG.info("【定时任务】定时刷新访客表获取未推送的APP新增访客信息channel id is: " + key);
                List<PropertyVisitorInfoByPushVO> infoByPushVOSList = iVisitorPushMessageDao.getVisitorInfoByPush();
                if(!obj.isOpen() || !obj.isConnected()){
                    GatewayService.removeGatewayChannel(key);
                }

                if(infoByPushVOSList!=null && infoByPushVOSList.size()>0){
                    String strUTF8 = URLDecoder.decode(JSON.toJSONString(infoByPushVOSList), "UTF-8");
                    System.out.println("strUTF8:"+strUTF8);
                    if(obj.isOpen()){
                        obj.write(new TextWebSocketFrame(strUTF8));

                        for(PropertyVisitorInfoByPushVO pushVO : infoByPushVOSList){
                            passId.add(pushVO.getPassId());
                        }
                    }else{
                        GatewayService.removeGatewayChannel(key);
                    }
                }
            }
            if(passId.size()>0){
                iVisitorPushMessageDao.updatePushStatus(passId.toArray(new String[passId.size()]));
                LOG.info("【定时任务】定时刷新访客表获取未推送的APP新增访客信息传送给客户端修改状态已完毕");
                System.err.println("传送给客户端修改状态已完毕");
            }
        }catch(Exception e){throw e;}
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

最后

以上就是稳重翅膀为你收集整理的Netty WebSocket长连接Netty WebSocket长连接的全部内容，希望文章能够帮你解决Netty WebSocket长连接Netty WebSocket长连接所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。