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

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

服務(wù)端實(shí)現(xiàn),

? 可以選擇讓netty服務(wù)端伴隨著springboot啟動(dòng)，即通過注解的形式，讓netty Server變成一個(gè)bean,當(dāng)加載這個(gè)bean時(shí)默認(rèn)啟動(dòng)，但這種方式當(dāng)啟動(dòng)nettyServer后會(huì)阻斷springboot的啟動(dòng)郁副，即加載到nettyServer這個(gè)bean之后，后面的bean不能繼續(xù)加載，springboot項(xiàng)目的啟動(dòng)也將阻塞在這父款。第二種方式通過手動(dòng)啟動(dòng)，比較友好瞻凤，不會(huì)阻塞springboot的啟動(dòng)憨攒。

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);

        /***
         * 綁定端口并啟動(dòng)去接收進(jìn)來的連接
         */
        ChannelFuture f = null;
        try {
            f = serverBootstrap.bind(6910).sync();

            System.err.println("啟動(dòng)成功");
            /**
             * 這里會(huì)一直等待，直到socket被關(guān)閉
             */
            f.channel().closeFuture().sync();
            System.err.println("Start succss!");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            System.err.println("關(guān)閉完成!");
        }
    }

    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());
        }

    }
}

手動(dòng)方式啟動(dòng):

public class NettyApplyApplication {

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

netty如何實(shí)現(xiàn)業(yè)務(wù)的分發(fā)

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

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

? 所有的實(shí)體業(yè)務(wù)bean都統(tǒng)一實(shí)現(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<>();

    /**
     * 消息流轉(zhuǎn)處理
     *
     * @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("設(shè)置map的值");
        if (courseMap != null && courseMap.size() > 0) {
            for (Map.Entry<String, Object> entry : courseMap.entrySet()) {
                coursesTable.put(entry.getKey(), entry.getValue());
            }
        }
    }


}

業(yè)務(wù)bean

@Component
@Biz(value = "10003")
public class LinsenceBizService implements BaseBusinessCourse {
    @Override
    public void doBiz(ChannelHandlerContext context, String message) {
        System.out.println("業(yè)務(wù)層收到的數(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è)務(wù)bean的注入

? 這里通過監(jiān)聽ApplicationStartedEvent事件溜族，在項(xiàng)目啟動(dòng)完成后獲取到所有被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";
    }
}

客戶端

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