SpringBoot+Netty+WebSocket,服务端、JS页面、客户端互发消息（附项目源码）

SpringBoot+Netty+WebSocket,服务端、JS页面、客户端互发消息（附项目源码）

Netty简介

    Netty是基于Java NIO client-server的网络应用框架，使用Netty可以快速开发网络应用，例如服务器和客户端协议。Netty提供了一种新的方式来开发网络应用程序，这种新的方式使它很容易使用和具有很强的扩展性。Netty的内部实现是很复杂的，但是Netty提供了简单易用的API从网络处理代码中解耦业务逻辑。Netty是完全基于NIO实现的，所以整个Netty都是异步的。

项目源码：百度网盘 请输入提取码    提取码: 4e7u

1.服务端

步骤一：pom文件引入

  
            io.netty
            netty-all
            4.1.33.Final
        

        
            cn.hutool
            hutool-all
            5.2.3
        


        
            com.alibaba
            fastjson
            1.2.76
        

步骤二：NettyConfig

package com.example.nettybackstage.config;

import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.util.concurrent.ConcurrentHashMap;

public class NettyConfig {
    
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    
    private static ConcurrentHashMap userChannelMap = new ConcurrentHashMap<>();

    private NettyConfig() {}

    
    public static ChannelGroup getChannelGroup() {
        return channelGroup;
    }

    
    public static ConcurrentHashMap getUserChannelMap(){
        return userChannelMap;
    }
}

步骤三：NettyServer

package com.example.nettybackstage.config;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.stream.ChunkedWriteHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.InetSocketAddress;



@Component
public class NettyServer{
    private static final Logger log = LoggerFactory.getLogger(NettyServer.class);

    
    @Value("${webSocket.protocol}")
    private String webSocketProtocol;

    
    @Value("${webSocket.netty.port}")
    private int port;

    
    @Value("${webSocket.netty.path}")
    private String webSocketPath;

    @Autowired
    private WebSocketHandler webSocketHandler;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workGroup;


    
    private void start() throws InterruptedException {
        //数据量上来时设置线程池
        
        bossGroup = new NioEventLoopGroup();
        workGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        // bossGroup辅助客户端的tcp连接请求, workGroup负责与客户端之前的读写 *** 作
        bootstrap.group(bossGroup,workGroup);
        // 设置NIO类型的channel
        bootstrap.channel(NioServerSocketChannel.class);
        // 设置监听端口
        bootstrap.localAddress(new InetSocketAddress(port));
        // 连接到达时会创建一个通道
        bootstrap.childHandler(new ChannelInitializer() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                // 流水线管理通道中的处理程序（Handler），用来处理业务
                // webSocket协议本身是基于http协议的，所以这边也要使用http编解码器
                ch.pipeline().addLast(new HttpServerCodec());
                ch.pipeline().addLast(new ObjectEncoder());
                // 以块的方式来写的处理器
                ch.pipeline().addLast(new ChunkedWriteHandler());
        
                ch.pipeline().addLast(new HttpObjectAggregator(8192));
        
                ch.pipeline().addLast(new WebSocketServerProtocolHandler(webSocketPath, webSocketProtocol, true, 65536 * 10));
                // 自定义的handler，处理业务逻辑
                ch.pipeline().addLast(webSocketHandler);

            }
        });
        // 配置完成，开始绑定server，通过调用sync同步方法阻塞直到绑定成功
        ChannelFuture channelFuture = bootstrap.bind().sync();
        log.info("Server started and listen on:{}",channelFuture.channel().localAddress());
        // 对关闭通道进行监听
        channelFuture.channel().closeFuture().sync();
    }

    
    @PreDestroy
    public void destroy() throws InterruptedException {
        if(bossGroup != null){
            bossGroup.shutdownGracefully().sync();
        }
        if(workGroup != null){
            workGroup.shutdownGracefully().sync();
        }
    }
    @PostConstruct()
    public void init() {
        //需要开启一个新的线程来执行netty server 服务器
        new Thread(() -> {
            try {
                start();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }

}

步骤四：WebSocketHandler

package com.example.nettybackstage.config;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketframe;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.util.HashMap;
import java.util.Map;



@Component
@ChannelHandler.Sharable
public class WebSocketHandler extends SimpleChannelInboundHandler {

    private static final Logger log = LoggerFactory.getLogger(WebSocketHandler.class);
    //标识
    private static final String CHANNEL_TYPE = "netty_msg";

    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        log.info("handlerAdded 被调用:{}",ctx.channel().id().asLongText());
        // 添加到channelGroup 通道组
        NettyConfig.getChannelGroup().add(ctx.channel());
    }

    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msgObj) throws Exception {
        // 1. 获取客户端发送的消息
        try {
            TextWebSocketframe msg = (TextWebSocketframe) msgObj;
            if (!ObjectUtils.isEmpty(msg.text())){
                Map map = new HashMap<>();
                map = com.alibaba.fastjson.JSONObject.parseObject(msg.text(),Map.class);
                String typeId = String.valueOf(map.get("typeId"));
                switch (typeId){
                    //netty_msg_first类型ID:首次建立连接管道信息
                    case "netty_msg_first":
                        log.info("(类型ID:netty_msg_first)服务器收到消息连接成功：{}",msg.text());
                        // 获取CHANNEL_TYPE,关联channel(自定义发送类型)
                        NettyConfig.getUserChannelMap().put(CHANNEL_TYPE,ctx.channel());
                        // 将用type作为自定义属性加入到channel中，方便随时channel中获取type
                        AttributeKey key = AttributeKey.valueOf("type");
                        ctx.channel().attr(key).setIfAbsent(CHANNEL_TYPE);
                        break;

                        //netty_web_20220204类型ID:接收前端发送的消息类型
                    case "netty_web_20220204":
                        log.info("(类型ID:netty_web_20220204)服务器收到前端发送消息：{}",msg.text());

                        break;

                    //netty_web_20220204类型ID:接收客户端发送的消息类型
                    case "netty_client_20220204":
                        log.info("(类型ID:netty_client_20220204)服务器收到前端发送消息：{}",msg.text());

                        break;
                }
            }

        }catch (Exception e){
            log.error(e.getMessage());
            e.printStackTrace();
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketframe textWebSocketframe) throws Exception {

    }


    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        log.info("handlerRemoved 被调用:{}",ctx.channel().id().asLongText());
        // 删除通道
        NettyConfig.getChannelGroup().remove(ctx.channel());
        removeUserId(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("异常：{}",cause.getMessage());
        // 删除通道
        NettyConfig.getChannelGroup().remove(ctx.channel());
        removeUserId(ctx);
        ctx.close();
    }

    
    private void removeUserId(ChannelHandlerContext ctx){
        AttributeKey key = AttributeKey.valueOf("type");
        String userId = ctx.channel().attr(key).get();
        NettyConfig.getUserChannelMap().remove(userId);
    }
}

步骤五：MessageDto

package com.example.nettybackstage.dto;

import java.io.Serializable;
import java.util.Map;

public class MessageDto implements Serializable {

    public MessageDto(){}

    public MessageDto(String nettyId,String type,String status,String msg,Map objectMap,String typeId,String date){
        this.nettyId = nettyId;
        this.type = type;
        this.status = status;
        this.msg = msg;
        this.objectMap = objectMap;
        this.typeId = typeId;
        this.date = date;
    }

    private String nettyId;

    private String type;

    private String status;

    private String msg;

    private String typeId;

    private String date;

    public String getDate() {
        return date;
    }

    public void setDate(String date) {
        this.date = date;
    }

    public String getTypeId() {
        return typeId;
    }

    public void setTypeId(String typeId) {
        this.typeId = typeId;
    }

    private Map objectMap;

    public String getNettyId() {
        return nettyId;
    }

    public void setNettyId(String nettyId) {
        this.nettyId = nettyId;
    }

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    public String getStatus() {
        return status;
    }

    public void setStatus(String status) {
        this.status = status;
    }

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }

    public Map getObjectMap() {
        return objectMap;
    }

    public void setObjectMap(Map objectMap) {
        this.objectMap = objectMap;
    }

}

步骤六：PushService

package com.example.nettybackstage.service;

import com.example.nettybackstage.dto.MessageDto;

public interface PushService {
    
    public void sendMsgToOne(String type, MessageDto msg);

    
    public void sendMsgToAll(MessageDto msg);
}

步骤七：PushServiceImpl

package com.example.nettybackstage.serviceimpl;

import com.alibaba.fastjson.JSONObject;
import com.example.nettybackstage.config.NettyConfig;
import com.example.nettybackstage.config.WebSocketHandler;
import com.example.nettybackstage.dto.MessageDto;
import com.example.nettybackstage.service.PushService;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketframe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.concurrent.ConcurrentHashMap;


@Service
public class PushServiceImpl implements PushService {

    private static final Logger log = LoggerFactory.getLogger(PushServiceImpl.class);

    @Override
    public void sendMsgToOne(String type, MessageDto msg){
        ConcurrentHashMap userChannelMap = NettyConfig.getUserChannelMap();
        Channel channel = userChannelMap.get(type);
        String jsonMsg = JSONObject.toJSonString(msg);
        channel.writeAndFlush(new TextWebSocketframe(jsonMsg));
        log.info("pushMsgToOne消息发送成功={}",jsonMsg);
    }
    @Override
    public void sendMsgToAll(MessageDto msg){
        String jsonMsg = JSONObject.toJSonString(msg);
        NettyConfig.getChannelGroup().writeAndFlush(new TextWebSocketframe(jsonMsg));
        log.info("pushMsgToAll消息发送成功={}",msg);
    }
}

步骤八：PushController

package com.example.nettybackstage.web;

import com.example.nettybackstage.dto.MessageDto;
import com.example.nettybackstage.serviceimpl.PushServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;


@RestController
@RequestMapping("/push")
public class PushController {

    @Autowired
    private PushServiceImpl pushService;

    
    @PostMapping("/sendMsgToAll")
    public void sendMsgToAll(@RequestParam("msg") String msg){
        MessageDto messageDto = new MessageDto();
        String uuid = String.valueOf(UUID.randomUUID()).replace("-","");
        messageDto.setNettyId(uuid);
        messageDto.setMsg(msg);
        messageDto.setStatus("0000");
        messageDto.setType("netty_msg");
        messageDto.setTypeId("");
        messageDto.setDate("2021-12-05 21:19:10");
        Map map = new HashMap<>();
        map.put("id",uuid);
        map.put("name","王五");
        map.put("number","A-1257564246");
        messageDto.setObjectMap(map);
        pushService.sendMsgToAll(messageDto);
    }

    
    @PostMapping("/sendMsgToOne")
    public void sendMsgToOne(@RequestParam("name") String name,@RequestParam("msg") String msg,@RequestParam("typeId") String typeId){
        MessageDto messageDto = new MessageDto();
        String uuid = String.valueOf(UUID.randomUUID()).replace("-","");
        messageDto.setNettyId(uuid);
        messageDto.setMsg(msg);
        messageDto.setStatus("0000");
        messageDto.setType("netty_msg");
        messageDto.setTypeId(typeId);
        messageDto.setDate("2022-02-05 15:09:00");
        Map map = new HashMap<>();
        map.put("id",uuid);
        map.put("name",name);
        map.put("number","1257564246");
        messageDto.setObjectMap(map);
        pushService.sendMsgToOne(messageDto.getType(),messageDto);
    }
}

步骤十：application.properties

server.port=8081
#netty配置
#连接端口
webSocket.netty.port=58080
#协定名称(可定义)
webSocket.protocol=webSocket
#路径
webSocket.netty.path=/${webSocket.protocol}

2.页面客户端

步骤一

3.客户端

步骤一:pom

 
            org.java-websocket
            Java-WebSocket
            1.5.1
        
        
            com.alibaba
            fastjson
            1.2.76
        

步骤二：WebSocketListener

package com.example.nettyclient.listener;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft_6455;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.net.URI;
import java.util.HashMap;
import java.util.Map;


@Slf4j
@Component
public class WebSocketListener {

    @Bean
    public WebSocketClient webSocketClient() {
        try {
            WebSocketClient webSocketClient = new WebSocketClient(new URI("ws://127.0.0.1:58080/webSocket"),new Draft_6455()) {
                @Override
                public void onOpen(ServerHandshake handshakedata) {
                    log.info("[websocket] 连接成功");
                    Map map = new HashMap<>();
                    map.put("typeId","netty_msg_first");
                    String jsonMsg = JSONObject.toJSonString(map);
                    this.send(jsonMsg);
                    log.info("[websocket] 回发连接成功消息!");
                }

                @Override
                public void onMessage(String message) {
                    log.info("[websocket] 收到消息={}",message);
                    if (ObjectUtils.isEmpty(message)){
                        return;
                    }

                    Map map = JSONObject.parseObject(message,Map.class);
                    String typeId = String.valueOf(map.get("typeId"));
                    switch (typeId){
                        case "netty_client_20220204":
                            log.info("[websocket] 收到类型ID为={},消息={}",typeId,message);
                            break;
                    }

                }

                @Override
                public void onClose(int code, String reason, boolean remote) {
                    log.info("[websocket] 退出连接");
                }

                @Override
                public void onError(Exception ex) {
                    log.info("[websocket] 连接错误={}",ex.getMessage());
                }
            };
            webSocketClient.connect();
            return webSocketClient;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

}

4.服务端和页面客户端互发消息

启动服务端和页面客户端,启动后访问页面:http://127.0.0.1:8082/index.html,如下:

 

 用postman发送一条消息

页面成功接收到服务端即时发送的消息，同时页面也向服务端回发一条消息

5.服务端和客户端互相发送消息

启动服务

用postman调用服务端接口发送一条消息

 成功接收到服务端消息