springboot整合netty框架實現(xiàn)站內(nèi)信

更新時間：2022年12月23日 11:52:32 作者：起風(fēng)哥

Netty 是一個基于NIO的客戶、服務(wù)器端編程框架，使用Netty 可以確保你快速和簡單的開發(fā)出一個網(wǎng)絡(luò)應(yīng)用，這篇文章主要介紹了springboot整合netty框架的方式小結(jié),需要的朋友可以參考下

代碼用到的組件介紹

ChannelInitializer 見名知意，就是channel 初始化器，當(dāng)每個客戶端創(chuàng)建連接時這里面的代碼都會執(zhí)行一遍。由于，連接建立之后，這個channel就會常駐內(nèi)存，所以這里就有個值得思考的問題：

問題：哪些實例可以聲明成單例，或者交給spring管理？因為如果每個連接都創(chuàng)建這么一大堆對象，可以想像1萬個連接，這里會多占用多少內(nèi)存出來？

這個問題也不難回答，沒有中間態(tài)，線程安全的類是可以聲明成單例的，所以我們順著這個方向大概就可以知道哪些是可以作為單例進(jìn)行聲明得。授人以魚不如授人以漁。

SimpleChannelInboundHandler 這個類是個入站消息處理類，它對資源得釋放傳遞等做了抽取，同時提供了個channelRead0抽象方法給子類實現(xiàn)，并且將消息進(jìn)行泛型化，讓你可以更專注于你得業(yè)務(wù)邏輯處理。可以看它得父類，我們可以知道，它定義了很多時機(jī)得切入點(diǎn)，比如添加后操作，注冊后操作，異常后處理，或者某些事件后處理。我們可以利用這些不同得時機(jī)做一些定制化得處理。

HttpServerCodec 這個東西沒什么好說了，它很復(fù)雜，但是也就是個http協(xié)議得編解碼器，這個不介紹了。

ChunkedWriteHandler 因為netty下是io多路復(fù)用得，所以你一定不會想讓你得一個http請求被分割成多次被處理，這樣會出問題，所以當(dāng)你得消息過大時，使用這個類處理器就可以讓你得大數(shù)據(jù)量請求可以被一次異步進(jìn)行處理

HttpObjectAggregator 由于HttpServerCodec無法處理post請求中得body參數(shù)，所以還得使用這個編解碼器進(jìn)行處理。

WebSocketServerProtocolHandler 它也是繼承至入站處理器，應(yīng)該說功能核SimpleChannelInboundHandler類似，但是為什么又要做這個區(qū)分呢？原因也不復(fù)雜，看它得說明我們可以知道，入站消息被分為了控制幀和普通消息兩種類型，控制幀說得是比如客戶端發(fā)起關(guān)閉連接，心跳請求等等得處理在這個類被處理了，所以如果需要自定義得心跳處理，可以繼承這個類。而文本類或者其它二進(jìn)制類型得入站消息就可以繼承至SimpleChannelInboundHandler處理。

websocket連接過程

Websocket一開始的握手需要借助HTTP的GET請求完成。
TCP連接成功后，瀏覽器通過HTTP協(xié)議向服務(wù)器傳送WebSocket支持的版本號等信息。
服務(wù)器收到客戶端的握手請求后，同樣采用HTTP協(xié)議返回數(shù)據(jù)。
當(dāng)收到了連接成功的消息后，通過TCP通道進(jìn)行傳輸通信。

請求報文

Sec-WebSocket-Version: 13
Sec-WebSocket-Key: fHXEZ1icd2ZsBWB8+GqoXg==
Connection: Upgrade
Upgrade: websocket
Host: localhost:9090

Upgrade：websocket / Connection: Upgrade ：參數(shù)值表明這是 WebSocket類型請求（這個是Websocket的核心，告訴Apache、Nginx等服務(wù)器，發(fā)起的是Websocket協(xié)議）。
Sec-WebSocket-Key ：是一個 Base64編碼的值，是由瀏覽器隨機(jī)生成的，提供基本的防護(hù)，防止惡意或者無意的連接。
Sec_WebSocket-Protocol ：是一個用戶定義的字符串，用來區(qū)分同URL下，不同的服務(wù)所需要的協(xié)議。可以不傳
Sec-WebSocket-Version ：表示 WebSocket 的版本，默認(rèn)13

響應(yīng)報文

upgrade: websocket
connection: upgrade
sec-websocket-accept: yvrH9uLtxFSIDyS2ZwrnPKuiPvs=

首先，101 狀態(tài)碼表示服務(wù)器已經(jīng)理解了客戶端的請求，并將通過 Upgrade 消息頭通知客戶端采用不同的協(xié)議來完成這個請求；
然后，Sec-WebSocket-Accept 這個則是經(jīng)過服務(wù)器確認(rèn)，并且加密過后的 Sec-WebSocket-Key；
最后，Sec-WebSocket-Protocol 則是表示最終使用的協(xié)議，可以沒有

websocket的握手請求的消息類型是FullHttpRequest，所以我們可以定義一個channelHander專門處理握手請求得一些定制化操作，比如認(rèn)證操作，認(rèn)證通過后，將用戶未讀消息數(shù)帶回去。并將用戶和對應(yīng)得channel信息進(jìn)行映射保存起來，后續(xù)通過mq推送得消息要發(fā)給誰獲取channel進(jìn)行推送消息。

由于是站內(nèi)信形式得，所以我們可以屏蔽客戶端主動向服務(wù)端發(fā)起的消息，空處理就可以了。如果需要處理再頂一個ChannelHandler 消息類型為 TextWebSocketFrame的SimpleChannelInboundHandler <TextWebSocketFrame>在channelRead0方法中去處理即可。所以我們這里面主要的兩段邏輯很簡單就是第一個做認(rèn)證，并保存對應(yīng)的用戶和channel關(guān)系，第二個，從mq訂閱消息，將消息發(fā)送給對應(yīng)用戶的channel。但是這里面也有一些值得思考的問題。

問題：

1、怎么防止一個用戶使用一個token對服務(wù)器無限個連接？

答：channel中存了個AttributeMap我們可以將對應(yīng)的屬性設(shè)置給channel，每個channel連接進(jìn)來的時候我們先判斷下對應(yīng)的token是不是已經(jīng)連接過了即可。如果已經(jīng)連接過了直接返回不讓連接了。

2、需求上用戶賬號如果可以同一時間多地登入，或者多端登入，如何處理？

答：我們可以通過存儲一個map <userId,List<channel>>的結(jié)構(gòu)，這樣就支持一個賬號多端登入都能收到消息了。

3、用戶不在線，消息如何持久化。

答：我們作為一個消息分發(fā)器，為了高性能，所以我們不做連接數(shù)據(jù)庫的操作，所以我們可以選擇的時機(jī)是在客戶端將消息發(fā)送到mq前將消息先持久化起來，這樣作為消息分發(fā)的服務(wù)端，就可以做到只管分發(fā)，不管存儲，用戶沒在線，就直接丟棄。

4、消息服務(wù)多節(jié)點(diǎn)，會產(chǎn)生什么問題？如何解決？

答：如果服務(wù)多節(jié)點(diǎn)，就會產(chǎn)生一個問題，就是客戶端連接進(jìn)來了只會連接在一個節(jié)點(diǎn)上，那么此時，哪個節(jié)點(diǎn)拿到mq消息就成了問題，所以此處我們可以使用廣播的形式將消息廣播給所有節(jié)點(diǎn)，

在節(jié)點(diǎn)上判斷如果這個用戶消息在我這里我就推給他，不在我這里我就直接丟棄就好了。

5、在微服務(wù)下，消息服務(wù)并非只有處理站內(nèi)信，那么在springboot下我們開了兩個端口，這兩個端口該如何暴露給網(wǎng)關(guān)？

答：我的做法是將兩個端口作為兩個不同的服務(wù)注冊給網(wǎng)關(guān)即可。我這邊用nacos，詳情可以查看后續(xù)的代碼

創(chuàng)建springboot項目，引入maven包

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

創(chuàng)建WebSocketChannelInitializer

常量

public interface ServerConst {
    String SERVICE_NAME = "netty-notice";
    int DEFAULT_PORT=9090;
}

@Slf4j
public class WebSocketChannelInitializer extends ChannelInitializer<SocketChannel> {
    private FullHttpRequestHandler paramsHandler;
    private TextWebSocketFrameHandler textHandler;
    public WebSocketChannelInitializer(FullHttpRequestHandler paramsHandler,
        TextWebSocketFrameHandler textHandler){
        this.paramsHandler=paramsHandler;
        this.textHandler=textHandler;
    }
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        log.info("連接初始化");
        //http編解碼
        ch.pipeline().addLast(new HttpServerCodec());
        //大數(shù)據(jù)量讀寫
        ch.pipeline().addLast(new ChunkedWriteHandler());
        //http消息聚合
        ch.pipeline().addLast(new HttpObjectAggregator(65536));
        //連接升級處理
        ch.pipeline().addLast(paramsHandler);
        //協(xié)議配置
        ch.pipeline().addLast(new WebSocketServerProtocolHandler("/"+ServerConst.SERVICE_NAME+"/ws"));
        //ch.pipeline().addLast(new IdleStateHandler(30, 60, 120));
        //消息處理，業(yè)務(wù)邏輯
        ch.pipeline().addLast(textHandler);
    }
}

定義TextWebSocketFrameHandler，這個可以聲明為bean

@Slf4j
@Component
@ChannelHandler.Sharable
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        AttributeKey<String> utKey = AttributeKey.valueOf(UserChannelContext.USER_TOKEN);
        Attribute<String> ut = ctx.channel().attr(utKey);
        UserChannelContext.remove(ut.get(), ctx.channel());
        ctx.channel().close();
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //正常流程下不存在問題，但是無法處理硬件層面問題導(dǎo)致連接斷開等,連接斷開時移除channel
        AttributeKey<String> utKey = AttributeKey.valueOf(UserChannelContext.USER_TOKEN);
        Attribute<String> ut = ctx.channel().attr(utKey);
        UserChannelContext.remove(ut.get(),ctx.channel());
    }
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext,
        TextWebSocketFrame textWebSocketFrame) throws Exception {
        //todo 如果有長耗時業(yè)務(wù)邏輯處理，建議將數(shù)據(jù)打包到另一個線程處理？
    }
}

自定義控制幀處理CustomWebSocketServerProtocolHandler

/**
 * @Description todo 控制幀處理，關(guān)閉幀，ping幀，pong幀，暫時未處理
 * @Author 姚仲杰#80998699
 * @Date 2022/12/6 11:33
 */
public class CustomWebSocketServerProtocolHandler extends WebSocketServerProtocolHandler {
    public CustomWebSocketServerProtocolHandler(String websocketPath) {
        super(websocketPath);
    }
    public CustomWebSocketServerProtocolHandler(String websocketPath, boolean checkStartsWith) {
        super(websocketPath, checkStartsWith);
    }
    public CustomWebSocketServerProtocolHandler(String websocketPath, String subprotocols) {
        super(websocketPath, subprotocols);
    }
    public CustomWebSocketServerProtocolHandler(String websocketPath, String subprotocols,
        boolean allowExtensions) {
        super(websocketPath, subprotocols, allowExtensions);
    }
    public CustomWebSocketServerProtocolHandler(String websocketPath, String subprotocols,
        boolean allowExtensions, int maxFrameSize) {
        super(websocketPath, subprotocols, allowExtensions, maxFrameSize);
    }
    public CustomWebSocketServerProtocolHandler(String websocketPath, String subprotocols,
        boolean allowExtensions, int maxFrameSize, boolean allowMaskMismatch) {
        super(websocketPath, subprotocols, allowExtensions, maxFrameSize, allowMaskMismatch);
    }
    public CustomWebSocketServerProtocolHandler(String websocketPath, String subprotocols,
        boolean allowExtensions, int maxFrameSize, boolean allowMaskMismatch,
        boolean checkStartsWith) {
        super(websocketPath, subprotocols, allowExtensions, maxFrameSize, allowMaskMismatch,
            checkStartsWith);
    }
    public CustomWebSocketServerProtocolHandler(String websocketPath, String subprotocols,
        boolean allowExtensions, int maxFrameSize, boolean allowMaskMismatch,
        boolean checkStartsWith,
        boolean dropPongFrames) {
        super(websocketPath, subprotocols, allowExtensions, maxFrameSize, allowMaskMismatch,
            checkStartsWith, dropPongFrames);
    }
    @Override
    protected void decode(ChannelHandlerContext ctx, WebSocketFrame frame, List<Object> out)
        throws Exception {
        super.decode(ctx, frame, out);
    }
}

定義第一個請求的處理器，也就是握手連接升級等等，我們可以在這里做認(rèn)證等等的處理，這個也可以做為bean

@Slf4j
@Component
@ChannelHandler.Sharable
public class FullHttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
    @Autowired
    private RedisClient redisClient;
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
        String uri = request.uri();
        log.info("連接請求uri：{}",uri);
        Map<CharSequence, CharSequence> queryMap = UrlBuilder.ofHttp(uri).getQuery().getQueryMap();
        String ut = (String) queryMap.get("ut");
        //todo 此處進(jìn)行認(rèn)證操作
        if (!StringUtils.isEmpty(ut)){
            UserInfo userInfo = redisClient.get(String.format(CommonCacheConst.USER_UT_KEY, ut),UserInfo.class);
            if (userInfo!=null) {
                //認(rèn)證通過將channel緩存起來，便于服務(wù)端推送消息
                //todo 推送有多少未讀消息
                //一個ut只能建立一個連接，避免連接被占滿
                if (UserChannelContext.isConnected(ut)){
                    log.info("ut={}未認(rèn)證，連接失?。。。?,ut);
                    FullHttpResponse response = new DefaultFullHttpResponse(
                        HTTP_1_1, HttpResponseStatus.FORBIDDEN, Unpooled.wrappedBuffer("多次連接".getBytes()));
                    ctx.channel().writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
                    return;
                }
                String userCode = userInfo.getUserCode();
                AttributeKey<String> userIdKey = AttributeKey.valueOf(UserChannelContext.USER_KEY);
                ctx.channel().attr(userIdKey).setIfAbsent(userCode);
                AttributeKey<String> userTokenKey = AttributeKey.valueOf(UserChannelContext.USER_TOKEN);
                ctx.channel().attr(userTokenKey).setIfAbsent(ut);
                log.info("用戶{}連接成功?。?！",userCode);
                UserChannelContext.put(userCode,ut, ctx.channel());
            }else{
                log.info("ut={}未認(rèn)證，連接失?。。?！",ut);
                FullHttpResponse response = new DefaultFullHttpResponse(
                    HTTP_1_1, HttpResponseStatus.UNAUTHORIZED, Unpooled.wrappedBuffer("未認(rèn)證".getBytes()));
                ctx.channel().writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
                return;
            }
        }else{
            log.info("連接參數(shù)不正確ut不存在");
            FullHttpResponse response = new DefaultFullHttpResponse(
                HTTP_1_1, HttpResponseStatus.BAD_REQUEST, Unpooled.wrappedBuffer("參數(shù)不正確".getBytes()));
            ctx.channel().writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
            return;
        }
        request.setUri(URLUtil.getPath(uri));
        ctx.fireChannelRead(request.retain());
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

在定義個用戶上下文

public class UserChannelContext {
    public final static String USER_KEY="userCode";
    public final static String USER_TOKEN="ut";
    private static ConcurrentHashMap<String, List<Channel>> userChannelMap = new ConcurrentHashMap<>();
    private static ConcurrentHashMap<String, String> utConnectMap = new ConcurrentHashMap<>();
    public static boolean isConnected(String ut){
        return utConnectMap.containsKey(ut);
    }
    public static synchronized void put(String userCode,String ut, Channel channel) {
        utConnectMap.put(ut,ut);
        List<Channel> channels = get(userCode);
        if (channels!=null){
            channels.add(channel);
        }else{
            List<Channel> list = new ArrayList<>();
            list.add(channel);
            userChannelMap.put(userCode, list);
        }
    }
    public static List<Channel> get(String userCode) {
        return userChannelMap.get(userCode);
    }
    public static synchronized void remove(String ut,Channel channel){
        utConnectMap.remove(ut);
        AttributeKey<String> userCodeKey = AttributeKey.valueOf(USER_KEY);
        if (channel.hasAttr(userCodeKey)) {
            Attribute<String> userCode = channel.attr(userCodeKey);
            if (userCode!=null&&!StringUtils.isEmpty(userCode.get())){
                List<Channel> channels = userChannelMap.get(userCode.get());
                for (Channel cn : channels) {
                    if (cn.equals(channel)){
                        channels.remove(cn);
                        break;
                    }
                }
            }
        }
    }
}

將這個端口也注冊為一個服務(wù)給nacos注冊中心

@Component
public class NacosServiceRegister implements ApplicationContextAware, InitializingBean {
    private ApplicationContext context;
    private NacosRegistration nacosRegistration;
    @Autowired
    private NacosServiceRegistry registry;
    @Value("${netty.server.port:9090}")
    private int port;
    @Autowired
    NacosDiscoveryProperties properties;
    @Override
    public void afterPropertiesSet() throws Exception {
        NacosDiscoveryProperties nacosDiscoveryProperties = new NacosDiscoveryProperties();
        BeanUtils.copyProperties(properties, nacosDiscoveryProperties);
        nacosDiscoveryProperties.setService(ServerConst.SERVICE_NAME);
        nacosDiscoveryProperties.setPort(this.port);
        NacosRegistration nacosRegistration = new NacosRegistration(nacosDiscoveryProperties,
            context);
        this.nacosRegistration = nacosRegistration;
    }
    public void register() {
        this.registry.register(this.nacosRegistration);
    }
    public void deregister() {
        this.registry.deregister(this.nacosRegistration);
    }
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.context = applicationContext;
    }
}

給服務(wù)設(shè)置啟動類

@Slf4j
@Component
public class NettyRunner implements ApplicationRunner {
    @Value("${netty.server.port:9090}")
    private int port;
    private FullHttpRequestHandler paramsHandler;
    private TextWebSocketFrameHandler textHandler;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    @Autowired
    NacosServiceRegister nacosServiceRegister;
    @Override
    public void run(ApplicationArguments args) throws Exception {
        this.start();
        nacosServiceRegister.register();
    }
    //養(yǎng)成好習(xí)慣，標(biāo)注下作為bean示例化的構(gòu)造函數(shù)，當(dāng)然也可以不寫
    @Autowired
    public NettyRunner(FullHttpRequestHandler paramsHandler,
        TextWebSocketFrameHandler textHandler) {
        this.paramsHandler=paramsHandler;
        this.textHandler=textHandler;
    }
    public void start() throws Exception {
        this.bossGroup = new NioEventLoopGroup(1);
        this.workerGroup = new NioEventLoopGroup();
        ServerBootstrap sb = new ServerBootstrap();
        //tcp連接隊列長度
        sb.option(ChannelOption.SO_BACKLOG, 1024);
        //設(shè)置線程池，連接線程池和工作線程池
        sb.group(bossGroup,workerGroup)
            //這里怎么判斷使用epoll還是kqueue？
            .channel(NioServerSocketChannel.class)
            //服務(wù)地址于端口號設(shè)置
            .localAddress(this.port)
            //channel初始化操作
            .childHandler(new WebSocketChannelInitializer(paramsHandler,textHandler));
        sb.bind().sync();
        log.info("Netty started on port(s):{}", this.port);
    }
    @PreDestroy
    private void destroy() throws InterruptedException {
        if (ObjectUtil.isNotNull(this.bossGroup)) {
            this.bossGroup.shutdownGracefully().sync();
        }
        if (ObjectUtil.isNotNull(this.workerGroup)) {
            this.workerGroup.shutdownGracefully().sync();
        }
        nacosServiceRegister.deregister();
    }
}

定義mq消息監(jiān)聽器用戶接收消息然后分發(fā)給特定用戶

常量

public interface QueueConst {
    String NOTICE_DIRECT_QUEUE = "notice_direct_queue";
    String NOTICE_DIRECT_EXCHANGE = "notice_direct_exchange";
    String NOTICE_DIRECT_BIND_KEY = "notice_direct_bind_key";
}

@Configuration
public class QueueConfiguration {
    @Bean
    public Queue noticeQueue() {
        return new Queue(QueueConst.NOTICE_DIRECT_QUEUE);
    }
    @Bean
    public DirectExchange noticeDirectExchange() {
        return new DirectExchange(QueueConst.NOTICE_DIRECT_EXCHANGE);
    }
    @Bean
    public Binding noticeDirectBinding() {
        return BindingBuilder.bind(noticeQueue()).to(noticeDirectExchange()).with(QueueConst.NOTICE_DIRECT_BIND_KEY);
    }
}

@Component
public class NoticeReceiver {
    private static ObjectMapper MAPPER = new ObjectMapper();
    @RabbitListener(queues = QueueConst.NOTICE_DIRECT_QUEUE)
    @RabbitHandler
    public void receiveTopic(Message message) throws Exception {
        String receiveMsg = new String(message.getBody());
        message.getMessageProperties().getReceivedUserId();
        ChannelMessage channelMessage=JSONUtil.toBean(receiveMsg,ChannelMessage.class);
        //todo 將消息存庫，此處采用另一個方案，直接由消息發(fā)送方進(jìn)行存儲，這里只做分發(fā)
        //save
        //todo 獲取對應(yīng)用戶的channel列表,并推送消息給用戶
        List<Channel> channels = UserChannelContext.get(channelMessage.getUserId());
        for (Channel channel : channels) {
            if (channel!=null){
                //todo 發(fā)送消息
                channel.writeAndFlush(new TextWebSocketFrame(MAPPER.writeValueAsString(channelMessage.getData())));
            }
        }
        //todo 補(bǔ)充：如果用戶不在線則直接放棄;
        //todo 補(bǔ)充：無論如何消息消費(fèi)后需要返回ack
    }
    public static void main(String[] args) {
        ChannelMessage message=new ChannelMessage();
        message.setUserId("YG0000049");
        message.setData("Hello world");
        System.out.println(JSONUtil.toJsonStr(message));
    }
}

channelMessage

@Data
@Accessors
public class ChannelMessage<T> implements Serializable {
    private String userId;
    private T data;
}

用戶登入成功后，會將ut存在對應(yīng)的redis中，所以我們在認(rèn)證的時候是去redis中直接取ut進(jìn)行比對即可，登入模塊我就不貼了

直接啟動springboot項目打開postman，進(jìn)行連接

可以看到連接成功

后臺日志

接著我們打開rabbitmq控制臺，直接發(fā)送一條信息，信息的生成實例在NoticeReceiver 中執(zhí)行main函數(shù)即可

點(diǎn)擊發(fā)布我們可以看到消息已經(jīng)到postman中

剩下的事情就是將服務(wù)部署上線配置nginx轉(zhuǎn)發(fā)規(guī)則即可

map $http_upgrade $connection_upgrade {
default keep-alive;
'websocket' upgrade;
}

server{

location /websocket/{
proxy_pass http://xxx/;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $connection_upgrade;
proxy_connect_timeout 60s;
proxy_read_timeout 7200s;
proxy_send_timeout 60s;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}