关于netty结合springboot的一些高级用法

netty和springboot的整合方式，netty采用的是4.0.25版本

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

服务端实现,

可以选择让netty服务端伴随着springboot启动，即通过注解的形式，让netty Server变成一个bean,当加载这个bean时默认启动，但这种方式当启动nettyServer后会阻断springboot的启动，即加载到nettyServer这个bean之后，后面的bean不能继续加载，springboot项目的启动也将阻塞在这。第二种方式通过手动启动，比较友好，不会阻塞springboot的启动。

package com.zt.apply.server;

import com.zt.apply.handler.DispacherHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

//@Component
public class Server {


//    @PostConstruct
    public static void run() {

        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();

        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workerGroup);
        serverBootstrap.channel(NioServerSocketChannel.class);
        serverBootstrap.childHandler(new ChanelInit());
        serverBootstrap = serverBootstrap.option(ChannelOption.SO_BACKLOG, 128);
        serverBootstrap = serverBootstrap.option(ChannelOption.SO_KEEPALIVE, true);

        /***
         * 绑定端口并启动去接收进来的连接
         */
        ChannelFuture f = null;
        try {
            f = serverBootstrap.bind(6910).sync();

            System.err.println("启动成功");
            /**
             * 这里会一直等待，直到socket被关闭
             */
            f.channel().closeFuture().sync();
            System.err.println("Start succss!");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            System.err.println("关闭完成!");
        }
    }

    static class ChanelInit extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            socketChannel.pipeline().addLast(new StringEncoder());
            socketChannel.pipeline().addLast(new StringDecoder());
            socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
            socketChannel.pipeline().addLast(new DispacherHandler());
        }

    }
}

手动方式启动:

public class NettyApplyApplication {

    public static void main(String[] args) {
        SpringApplication.run(NettyApplyApplication.class, args);
        Server.run();
    }
}

netty如何实现业务的分发

当netty接收到消息后将直接分发出去，交给业务层处理，而不用每次接收到消息都要handler自己去处理

package com.zt.apply.handler;
import com.zt.apply.dispacher.TcpDispacher;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;


public class DispacherHandler extends SimpleChannelInboundHandler<String> {

    private TcpDispacher tcpDispacher = TcpDispacher.getInstance();

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        tcpDispacher.messageRecived(channelHandlerContext, s);
    }

}

业务的分发

在handler接收到消息后，如何将消息分发到业务层，成了最大的问题。这也将是本文的精华所在，通过注解的形式实现业务的分发，在handler中收到消息后，调用TcpDispacher的messageRecived方法去处理，messageRecived方法将从客户端传过来的数据中解析出messageCode，根据messageCode找到这个code所对应的实体业务bean。

所有的实体业务bean都统一实现BaseBusinessCourse接口。

package com.zt.apply.dispacher;

import com.alibaba.fastjson.JSONObject;
import com.zt.apply.base.BaseBusinessCourse;
import io.netty.channel.ChannelHandlerContext;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: zt
 * @Date: 2019/1/6 15:50
 */
@Component
public class TcpDispacher {

    private static TcpDispacher instance = new TcpDispacher();

    private TcpDispacher() {

    }

    public static TcpDispacher getInstance() {
        return instance;
    }


    private static Map<String, Object> coursesTable = new ConcurrentHashMap<>();

    /**
     * 消息流转处理
     *
     * @param channelHandlerContext
     * @param s
     */
    public void messageRecived(ChannelHandlerContext channelHandlerContext, String s) {
        System.err.println("收到的消息:" + s);
        JSONObject jsonObject = JSONObject.parseObject(s);
        String code = jsonObject.getString("messageCode");
        BaseBusinessCourse baseBusinessCourse = (BaseBusinessCourse) coursesTable.get(code);
        baseBusinessCourse.doBiz(channelHandlerContext, s);
    }


    public void setCourses(Map<String, Object> courseMap) {
        System.err.println("设置map的值");
        if (courseMap != null && courseMap.size() > 0) {
            for (Map.Entry<String, Object> entry : courseMap.entrySet()) {
                coursesTable.put(entry.getKey(), entry.getValue());
            }
        }
    }


}

业务bean

@Component
@Biz(value = "10003")
public class LinsenceBizService implements BaseBusinessCourse {
    @Override
    public void doBiz(ChannelHandlerContext context, String message) {
        System.out.println("业务层收到的数据:" + message);
        context.writeAndFlush("{test:\"test\"}\r\n");
    }

}

Biz注解

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
@Documented
public @interface Biz {

    String value();

}

TcpDispacher中messageCode和业务bean的注入

这里通过监听ApplicationStartedEvent事件，在项目启动完成后获取到所有被Biz注解过的bean,并获取到注解中的value值，存入map,注入到TcpDispacher中。

public class ContextRefreshedListener implements ApplicationListener<ApplicationStartedEvent> {

    @Override
    public void onApplicationEvent(ApplicationStartedEvent applicationStartedEvent) {

        Map<String, Object> map = new HashMap<>();

        Map<String, Object> bizMap = applicationStartedEvent.getApplicationContext().getBeansWithAnnotation(Biz.class);
        for (Map.Entry<String, Object> entry : bizMap.entrySet()) {
            Object object = entry.getValue();
            Class c = object.getClass();
            System.err.println(c + "===>");
            Annotation[] annotations = c.getDeclaredAnnotations();

            for (Annotation annotation : annotations) {
                if (annotation.annotationType().equals(Biz.class)) {
                    Biz biz = (Biz) annotation;
                    map.put(biz.value(), object);
                }
            }
        }


        TcpDispacher tcpDispacher = (TcpDispacher) applicationStartedEvent.getApplicationContext().getBean("tcpDispacher");
        tcpDispacher.setCourses(map);

    }
}

请求体

public class BaseRequest {

    private String messageCode;

    private String content;

    public String getMessageCode() {
        return messageCode;
    }

    public void setMessageCode(String messageCode) {
        this.messageCode = messageCode;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

    public String toJson() {
        return JSON.toJSONString(this) + "\r\n";
    }
}

客户端

下篇再讲吧，同时交代如何在发布消息的地方实时获取到服务端返回的内容，通过回调实现。以及channel连接池的设计。

关于netty结合springboot的一些高级用法

猜你喜欢