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

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

服務端實現(xiàn),

? 可以選擇讓netty服務端伴隨著springboot啟動，即通過注解的形式，讓netty Server變成一個bean,當加載這個bean時默認啟動，但這種方式當啟動nettyServer后會阻斷springboot的啟動，即加載到nettyServer這個bean之后，后面的bean不能繼續(xù)加載，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如何實現(xiàn)業(yè)務的分發(fā)

? 當netty接收到消息后將直接分發(fā)出去，交給業(yè)務層處理，而不用每次接收到消息都要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);
    }

}

業(yè)務的分發(fā)

? 在handler接收到消息后，如何將消息分發(fā)到業(yè)務層，成了最大的問題。這也將是本文的精華所在，通過注解的形式實現(xiàn)業(yè)務的分發(fā)，在handler中收到消息后，調(diào)用TcpDispacher的messageRecived方法去處理，messageRecived方法將從客戶端傳過來的數(shù)據(jù)中解析出messageCode，根據(jù)messageCode找到這個code所對應的實體業(yè)務bean。

? 所有的實體業(yè)務bean都統(tǒng)一實現(xiàn)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());
            }
        }
    }


}

業(yè)務bean

@Component
@Biz(value = "10003")
public class LinsenceBizService implements BaseBusinessCourse {
    @Override
    public void doBiz(ChannelHandlerContext context, String message) {
        System.out.println("業(yè)務層收到的數(shù)據(jù):" + message);
        context.writeAndFlush("{test:\"test\"}\r\n");
    }

}

Biz注解

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

    String value();

}

TcpDispacher中messageCode和業(yè)務bean的注入

? 這里通過監(jiān)聽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";
    }
}

客戶端

? 下篇再講吧，同時交代如何在發(fā)布消息的地方實時獲取到服務端返回的內(nèi)容，通過回調(diào)實現(xiàn)。以及channel連接池的設計。