-
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连接池的设计。
