NettyClinet.java 8.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172
  1. package com.persagy.ztkencryptdecodedata.netty;
  2. import com.alibaba.fastjson.JSONObject;
  3. import com.persagy.ztkencryptdecodedata.entity.ChattingUser;
  4. import io.netty.bootstrap.Bootstrap;
  5. import io.netty.channel.*;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.nio.NioSocketChannel;
  8. import io.netty.handler.codec.string.StringDecoder;
  9. import io.netty.handler.codec.string.StringEncoder;
  10. import io.netty.util.CharsetUtil;
  11. import org.slf4j.Logger;
  12. import org.slf4j.LoggerFactory;
  13. import org.springframework.stereotype.Component;
  14. import javax.annotation.PostConstruct;
  15. import java.text.ParseException;
  16. import java.text.SimpleDateFormat;
  17. import java.util.Date;
  18. import java.util.concurrent.*;
  19. @Component
  20. public class NettyClinet implements NettyClientInterface {
  21. private static Logger logger = LoggerFactory.getLogger(NettyClinet.class.getClass());
  22. // private static BlockingQueue messageTcpSendQueue = new LinkedBlockingQueue<>(1024 * 1024);
  23. BlockingQueue<String> messageTcpSendQueue = new LinkedBlockingQueue<>(1024 * 1024);
  24. @PostConstruct
  25. public void init() {
  26. setMessageTcpSendQueue(messageTcpSendQueue);
  27. }
  28. static class GlobeTime {
  29. static public String globetime ="0";
  30. static public Object ackstatue;
  31. static public int globecount =0;
  32. public static void BuildAckstatue(Object ackstatue1){
  33. ackstatue=ackstatue1;
  34. }
  35. public static void BuildValue(String currentTime){
  36. globetime=currentTime;
  37. }
  38. public static int BuilValue(int count){
  39. globecount=count;
  40. return globecount;
  41. }
  42. }
  43. /*
  44. * 将时间转换为时间戳
  45. */
  46. public static String dateToStamp(String s) throws ParseException {
  47. String res;
  48. SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss:SSS");//格式化
  49. Date date = sdf.parse(s);
  50. long ts = date.getTime();
  51. res = String.valueOf(ts);
  52. return res;
  53. }
  54. public void setMessageTcpSendQueue(BlockingQueue<String> messageTcpSendQueue){
  55. int poolSize = 1;
  56. ExecutorService executorService = new ThreadPoolExecutor(poolSize,poolSize, 0L, TimeUnit.MILLISECONDS,
  57. new LinkedBlockingQueue<>());
  58. executorService.submit(() -> {
  59. while (true) {
  60. try {
  61. while (!messageTcpSendQueue.isEmpty()) {
  62. /// Thread.sleep(3000);
  63. Bootstrap bootstrap = new Bootstrap();
  64. EventLoopGroup group = new NioEventLoopGroup();
  65. try {
  66. bootstrap.group(group).channel(NioSocketChannel.class);
  67. bootstrap.handler(new ChannelInitializer<Channel>() {
  68. @Override
  69. protected void initChannel(Channel ch) throws Exception {
  70. ChannelPipeline pipeline = ch.pipeline();
  71. // pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
  72. // pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
  73. pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
  74. pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
  75. pipeline.addLast("handler", new TcpClientHandler());
  76. //pipeline.addLast("log", new LoggingHandler(LogLevel.INFO));
  77. }
  78. });
  79. logger.info("count1>>>>>>" + GlobeTime.BuilValue(GlobeTime.globecount + 1));
  80. Object messageObject = messageTcpSendQueue.peek();
  81. if (null == messageObject) {
  82. // System.out.println(">>>> queue has no data");
  83. logger.info(">>> Tcp queue size: " + messageTcpSendQueue.size());
  84. Thread.sleep(1000);
  85. } else {
  86. JSONObject object = JSONObject.parseObject(messageObject.toString());
  87. bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
  88. // Start the client.
  89. String currentTime = dateToStamp(object.getJSONObject("content").getString("time"));
  90. if (!(GlobeTime.globetime).equals(currentTime)) {
  91. ChannelFuture future = bootstrap.connect(object.getJSONObject("content").getString("targetAddress"), Integer.valueOf(object.getJSONObject("content").getString("port"))).sync();
  92. try {
  93. logger.info(">>> Tcp queue size: " + messageTcpSendQueue.size());
  94. logger.info("连接建立!" + future.channel().id());
  95. future.channel().writeAndFlush(object.getJSONObject("content").getString("msg")).sync();
  96. logger.info("sent——to——tcpserver>>>" + object.getJSONObject("content").getString("msg"));
  97. TimeUnit.SECONDS.sleep(1);
  98. if (future.isDone()) {
  99. while (true) {
  100. ChattingUser info = new ChattingUser();
  101. if ("ACK!".equals(GlobeTime.ackstatue)) {
  102. messageTcpSendQueue.poll();
  103. info.setAck("false");
  104. GlobeTime.BuildValue(currentTime);
  105. GlobeTime.BuildAckstatue(null);
  106. GlobeTime.BuilValue(0);
  107. break;
  108. } else {
  109. continue;
  110. }
  111. }
  112. }
  113. future.channel().closeFuture().sync();
  114. // System.out.println("-----------------");
  115. } catch (Exception e) {
  116. e.printStackTrace();
  117. logger.info(">>> 连接未建立>>>Tcp queue size: " + messageTcpSendQueue.size());
  118. logger.info(object.getJSONObject("content").getString("targetAddress") + ":" + Integer.valueOf(object.getJSONObject("content").getString("port")) + "连接未建立!");
  119. logger.error("--- tcp_send_error:" + e);
  120. } finally {
  121. future.channel().closeFuture().sync();
  122. }
  123. }
  124. }
  125. // logger.info(">>> kafka message: " + messageObject.toString());
  126. Thread.sleep(1000 * 5);
  127. } catch (InterruptedException e) {
  128. logger.error("--- tcp_send_error:" + e);
  129. e.printStackTrace();
  130. } finally {
  131. group.shutdownGracefully();
  132. }
  133. }}catch (Exception e){
  134. e.printStackTrace();}
  135. }
  136. });
  137. executorService.shutdown();
  138. }
  139. }