作者 crossoverJie

:sparkles: Introducing new features.

@@ -9,7 +9,7 @@ Netty 实战相关 @@ -9,7 +9,7 @@ Netty 实战相关
9 9
10 * [x] [Netty(一) SpringBoot 整合长连接心跳机制](https://crossoverjie.top/2018/05/24/netty/Netty(1)TCP-Heartbeat/) 10 * [x] [Netty(一) SpringBoot 整合长连接心跳机制](https://crossoverjie.top/2018/05/24/netty/Netty(1)TCP-Heartbeat/)
11 * [x] [Netty(二) 从线程模型的角度看 Netty 为什么是高性能的?](https://crossoverjie.top/2018/07/04/netty/Netty(2)Thread-model/) 11 * [x] [Netty(二) 从线程模型的角度看 Netty 为什么是高性能的?](https://crossoverjie.top/2018/07/04/netty/Netty(2)Thread-model/)
12 -* [ ] 如何解决 TCP 的拆包、粘包? 12 +* [x] [Netty(三) 什么是 TCP 拆、粘包?如何解决?](https://crossoverjie.top/2018/08/03/netty/Netty(3)TCP-Sticky/)
13 13
14 14
15 ## 安装 15 ## 安装
  1 +package com.crossoverjie.netty.action;
  2 +
  3 +import com.crossoverjie.netty.action.common.pojo.CustomProtocol;
  4 +import com.crossoverjie.netty.action.init.CustomerHandleInitializer;
  5 +import io.netty.bootstrap.Bootstrap;
  6 +import io.netty.channel.ChannelFuture;
  7 +import io.netty.channel.EventLoopGroup;
  8 +import io.netty.channel.nio.NioEventLoopGroup;
  9 +import io.netty.channel.socket.SocketChannel;
  10 +import io.netty.channel.socket.nio.NioSocketChannel;
  11 +import org.slf4j.Logger;
  12 +import org.slf4j.LoggerFactory;
  13 +import org.springframework.beans.factory.annotation.Value;
  14 +import org.springframework.stereotype.Component;
  15 +
  16 +import javax.annotation.PostConstruct;
  17 +
  18 +/**
  19 + * Function:
  20 + *
  21 + * @author crossoverJie
  22 + * Date: 22/05/2018 14:19
  23 + * @since JDK 1.8
  24 + */
  25 +@Component
  26 +public class HeartbeatClient {
  27 +
  28 + private final static Logger LOGGER = LoggerFactory.getLogger(HeartbeatClient.class);
  29 +
  30 + private EventLoopGroup group = new NioEventLoopGroup();
  31 +
  32 +
  33 + @Value("${netty.server.port}")
  34 + private int nettyPort;
  35 +
  36 + @Value("${netty.server.host}")
  37 + private String host;
  38 +
  39 + private SocketChannel channel ;
  40 +
  41 + @PostConstruct
  42 + public void start() throws InterruptedException {
  43 + Bootstrap bootstrap = new Bootstrap();
  44 + bootstrap.group(group)
  45 + .channel(NioSocketChannel.class)
  46 + .handler(new CustomerHandleInitializer())
  47 + ;
  48 +
  49 + ChannelFuture future = bootstrap.connect(host, nettyPort).sync();
  50 + if (future.isSuccess()) {
  51 + LOGGER.info("启动 Netty 成功");
  52 + }
  53 + channel = (SocketChannel) future.channel();
  54 + }
  55 +
  56 + /**
  57 + * 发送消息
  58 + * @param customProtocol
  59 + */
  60 + public void sendMsg(CustomProtocol customProtocol){
  61 + channel.writeAndFlush(customProtocol) ;
  62 + }
  63 +}
  1 +package com.crossoverjie.netty.action;
  2 +
  3 +import org.slf4j.Logger;
  4 +import org.slf4j.LoggerFactory;
  5 +import org.springframework.boot.CommandLineRunner;
  6 +import org.springframework.boot.SpringApplication;
  7 +import org.springframework.boot.autoconfigure.SpringBootApplication;
  8 +
  9 +/**
  10 + * @author crossoverJie
  11 + */
  12 +@SpringBootApplication
  13 +public class HeartbeatClientApplication implements CommandLineRunner{
  14 +
  15 + private final static Logger LOGGER = LoggerFactory.getLogger(HeartbeatClientApplication.class);
  16 +
  17 +
  18 +
  19 + public static void main(String[] args) {
  20 + SpringApplication.run(HeartbeatClientApplication.class, args);
  21 + LOGGER.info("启动 Client 成功");
  22 + }
  23 +
  24 + @Override
  25 + public void run(String... args) throws Exception {
  26 + }
  27 +}
  1 +package com.crossoverjie.netty.action.config;
  2 +
  3 +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
  4 +import org.springframework.context.annotation.Bean;
  5 +import org.springframework.context.annotation.Configuration;
  6 +import springfox.documentation.builders.ApiInfoBuilder;
  7 +import springfox.documentation.builders.PathSelectors;
  8 +import springfox.documentation.builders.RequestHandlerSelectors;
  9 +import springfox.documentation.service.ApiInfo;
  10 +import springfox.documentation.spi.DocumentationType;
  11 +import springfox.documentation.spring.web.plugins.Docket;
  12 +import springfox.documentation.swagger2.annotations.EnableSwagger2;
  13 +
  14 +
  15 +@Configuration
  16 +@EnableSwagger2
  17 +/** 是否打开swagger **/
  18 +@ConditionalOnExpression("'${swagger.enable}' == 'true'")
  19 +public class SwaggerConfig {
  20 +
  21 +
  22 + @Bean
  23 + public Docket createRestApi() {
  24 + return new Docket(DocumentationType.SWAGGER_2)
  25 + .apiInfo(apiInfo())
  26 + .select()
  27 + .apis(RequestHandlerSelectors.basePackage("com.crossoverjie.netty.action.client.controller"))
  28 + .paths(PathSelectors.any())
  29 + .build();
  30 + }
  31 +
  32 + private ApiInfo apiInfo() {
  33 + return new ApiInfoBuilder()
  34 + .title("sbc order api")
  35 + .description("sbc order api")
  36 + .termsOfServiceUrl("http://crossoverJie.top")
  37 + .contact("crossoverJie")
  38 + .version("1.0.0")
  39 + .build();
  40 + }
  41 +
  42 +}
  1 +package com.crossoverjie.netty.action.controller;
  2 +
  3 +import com.crossoverjie.netty.action.HeartbeatClient;
  4 +import com.crossoverjie.netty.action.vo.req.SendMsgReqVO;
  5 +import com.crossoverjie.netty.action.vo.res.SendMsgResVO;
  6 +import com.crossoverjie.netty.action.common.enums.StatusEnum;
  7 +import com.crossoverjie.netty.action.common.pojo.CustomProtocol;
  8 +import com.crossoverjie.netty.action.common.res.BaseResponse;
  9 +import io.swagger.annotations.ApiOperation;
  10 +import org.springframework.beans.factory.annotation.Autowired;
  11 +import org.springframework.stereotype.Controller;
  12 +import org.springframework.web.bind.annotation.RequestBody;
  13 +import org.springframework.web.bind.annotation.RequestMapping;
  14 +import org.springframework.web.bind.annotation.ResponseBody;
  15 +
  16 +/**
  17 + * Function:
  18 + *
  19 + * @author crossoverJie
  20 + * Date: 22/05/2018 14:46
  21 + * @since JDK 1.8
  22 + */
  23 +@Controller
  24 +@RequestMapping("/")
  25 +public class IndexController {
  26 +
  27 + @Autowired
  28 + private HeartbeatClient heartbeatClient ;
  29 +
  30 + /**
  31 + * 向服务端发消息
  32 + * @param sendMsgReqVO
  33 + * @return
  34 + */
  35 + @ApiOperation("发送消息")
  36 + @RequestMapping("sendMsg")
  37 + @ResponseBody
  38 + public BaseResponse<SendMsgResVO> sendMsg(@RequestBody SendMsgReqVO sendMsgReqVO){
  39 + BaseResponse<SendMsgResVO> res = new BaseResponse();
  40 + heartbeatClient.sendMsg(new CustomProtocol(122,sendMsgReqVO.getMsg())) ;
  41 +
  42 + SendMsgResVO sendMsgResVO = new SendMsgResVO() ;
  43 + sendMsgResVO.setMsg("OK") ;
  44 + res.setCode(StatusEnum.SUCCESS.getCode()) ;
  45 + res.setMessage(StatusEnum.SUCCESS.getMessage()) ;
  46 + res.setDataBody(sendMsgResVO) ;
  47 + return res ;
  48 + }
  49 +}
  1 +package com.crossoverjie.netty.action.encode;
  2 +
  3 +import com.crossoverjie.netty.action.common.pojo.CustomProtocol;
  4 +import io.netty.buffer.ByteBuf;
  5 +import io.netty.channel.ChannelHandlerContext;
  6 +import io.netty.handler.codec.MessageToByteEncoder;
  7 +
  8 +/**
  9 + * Function:编码
  10 + *
  11 + * @author crossoverJie
  12 + * Date: 17/05/2018 19:07
  13 + * @since JDK 1.8
  14 + */
  15 +public class HeartbeatEncode extends MessageToByteEncoder<CustomProtocol> {
  16 + @Override
  17 + protected void encode(ChannelHandlerContext ctx, CustomProtocol msg, ByteBuf out) throws Exception {
  18 +
  19 + out.writeLong(msg.getHeader()) ;
  20 + out.writeBytes(msg.getContent().getBytes()) ;
  21 +
  22 + }
  23 +}
  1 +package com.crossoverjie.netty.action.handle;
  2 +
  3 +import com.crossoverjie.netty.action.common.pojo.CustomProtocol;
  4 +import io.netty.buffer.ByteBuf;
  5 +import io.netty.buffer.Unpooled;
  6 +import io.netty.channel.ChannelHandlerContext;
  7 +import io.netty.channel.SimpleChannelInboundHandler;
  8 +import io.netty.handler.timeout.IdleState;
  9 +import io.netty.handler.timeout.IdleStateEvent;
  10 +import io.netty.util.CharsetUtil;
  11 +import org.slf4j.Logger;
  12 +import org.slf4j.LoggerFactory;
  13 +
  14 +/**
  15 + * Function:
  16 + *
  17 + * @author crossoverJie
  18 + * Date: 16/02/2018 18:09
  19 + * @since JDK 1.8
  20 + */
  21 +public class EchoClientHandle extends SimpleChannelInboundHandler<ByteBuf> {
  22 +
  23 + private final static Logger LOGGER = LoggerFactory.getLogger(EchoClientHandle.class);
  24 +
  25 +
  26 + @Override
  27 + public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  28 +
  29 + if (evt instanceof IdleStateEvent){
  30 + IdleStateEvent idleStateEvent = (IdleStateEvent) evt ;
  31 +
  32 + if (idleStateEvent.state() == IdleState.WRITER_IDLE){
  33 + LOGGER.info("已经 10 秒没有发送信息!");
  34 + //向客户端发送消息
  35 + CustomProtocol customProtocol = new CustomProtocol(45678L,"ping") ;
  36 + ctx.writeAndFlush(Unpooled.copiedBuffer(customProtocol.toString(), CharsetUtil.UTF_8)) ;
  37 + }
  38 +
  39 +
  40 + }
  41 +
  42 + super.userEventTriggered(ctx, evt);
  43 + }
  44 +
  45 + @Override
  46 + public void channelActive(ChannelHandlerContext ctx) throws Exception {
  47 +
  48 + //客户端和服务端建立连接时调用
  49 + LOGGER.info("已经建立了联系。。");
  50 + }
  51 +
  52 + @Override
  53 + protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf in) throws Exception {
  54 +
  55 + //从服务端收到消息时被调用
  56 + LOGGER.info("客户端收到消息={}",in.toString(CharsetUtil.UTF_8)) ;
  57 +
  58 + }
  59 +
  60 + @Override
  61 + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  62 + //异常时断开连接
  63 + cause.printStackTrace() ;
  64 + ctx.close() ;
  65 + }
  66 +}
  1 +package com.crossoverjie.netty.action.init;
  2 +
  3 +import com.crossoverjie.netty.action.encode.HeartbeatEncode;
  4 +import com.crossoverjie.netty.action.handle.EchoClientHandle;
  5 +import io.netty.channel.Channel;
  6 +import io.netty.channel.ChannelInitializer;
  7 +import io.netty.handler.timeout.IdleStateHandler;
  8 +
  9 +/**
  10 + * Function:
  11 + *
  12 + * @author crossoverJie
  13 + * Date: 23/02/2018 22:47
  14 + * @since JDK 1.8
  15 + */
  16 +public class CustomerHandleInitializer extends ChannelInitializer<Channel> {
  17 + @Override
  18 + protected void initChannel(Channel ch) throws Exception {
  19 + ch.pipeline()
  20 + //10 秒没发送消息
  21 + .addLast(new IdleStateHandler(0, 10, 0))
  22 + .addLast(new HeartbeatEncode())
  23 + .addLast(new EchoClientHandle())
  24 + ;
  25 + }
  26 +}
  1 +package com.crossoverjie.netty.action.vo.req;
  2 +
  3 +import com.crossoverjie.netty.action.common.req.BaseRequest;
  4 +import io.swagger.annotations.ApiModelProperty;
  5 +
  6 +import javax.validation.constraints.NotNull;
  7 +
  8 +/**
  9 + * Function:
  10 + *
  11 + * @author crossoverJie
  12 + * Date: 2018/05/21 15:56
  13 + * @since JDK 1.8
  14 + */
  15 +public class SendMsgReqVO extends BaseRequest {
  16 +
  17 + @NotNull(message = "msg 不能为空")
  18 + @ApiModelProperty(required = true, value = "msg", example = "hello")
  19 + private String msg ;
  20 +
  21 + public String getMsg() {
  22 + return msg;
  23 + }
  24 +
  25 + public void setMsg(String msg) {
  26 + this.msg = msg;
  27 + }
  28 +
  29 +}
  1 +package com.crossoverjie.netty.action.vo.res;
  2 +
  3 +/**
  4 + * Function:
  5 + *
  6 + * @author crossoverJie
  7 + * Date: 2017/6/26 15:43
  8 + * @since JDK 1.8
  9 + */
  10 +public class SendMsgResVO {
  11 + private String msg ;
  12 +
  13 + public String getMsg() {
  14 + return msg;
  15 + }
  16 +
  17 + public void setMsg(String msg) {
  18 + this.msg = msg;
  19 + }
  20 +}