123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172 |
- package com.persagy.ztkencryptdecodedata.netty;
- import com.alibaba.fastjson.JSONObject;
- import com.persagy.ztkencryptdecodedata.entity.ChattingUser;
- import io.netty.bootstrap.Bootstrap;
- import io.netty.channel.*;
- import io.netty.channel.nio.NioEventLoopGroup;
- import io.netty.channel.socket.nio.NioSocketChannel;
- import io.netty.handler.codec.string.StringDecoder;
- import io.netty.handler.codec.string.StringEncoder;
- import io.netty.util.CharsetUtil;
- import org.slf4j.Logger;
- import org.slf4j.LoggerFactory;
- import org.springframework.stereotype.Component;
- import javax.annotation.PostConstruct;
- import java.text.ParseException;
- import java.text.SimpleDateFormat;
- import java.util.Date;
- import java.util.concurrent.*;
- @Component
- public class NettyClinet implements NettyClientInterface {
- private static Logger logger = LoggerFactory.getLogger(NettyClinet.class.getClass());
- // private static BlockingQueue messageTcpSendQueue = new LinkedBlockingQueue<>(1024 * 1024);
- BlockingQueue<String> messageTcpSendQueue = new LinkedBlockingQueue<>(1024 * 1024);
- @PostConstruct
- public void init() {
- setMessageTcpSendQueue(messageTcpSendQueue);
- }
- static class GlobeTime {
- static public String globetime ="0";
- static public Object ackstatue;
- static public int globecount =0;
- public static void BuildAckstatue(Object ackstatue1){
- ackstatue=ackstatue1;
- }
- public static void BuildValue(String currentTime){
- globetime=currentTime;
- }
- public static int BuilValue(int count){
- globecount=count;
- return globecount;
- }
- }
- /*
- * 将时间转换为时间戳
- */
- public static String dateToStamp(String s) throws ParseException {
- String res;
- SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss:SSS");//格式化
- Date date = sdf.parse(s);
- long ts = date.getTime();
- res = String.valueOf(ts);
- return res;
- }
- public void setMessageTcpSendQueue(BlockingQueue<String> messageTcpSendQueue){
- int poolSize = 1;
- ExecutorService executorService = new ThreadPoolExecutor(poolSize,poolSize, 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<>());
- executorService.submit(() -> {
- while (true) {
- try {
- while (!messageTcpSendQueue.isEmpty()) {
- /// Thread.sleep(3000);
- Bootstrap bootstrap = new Bootstrap();
- EventLoopGroup group = new NioEventLoopGroup();
- try {
- bootstrap.group(group).channel(NioSocketChannel.class);
- bootstrap.handler(new ChannelInitializer<Channel>() {
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ChannelPipeline pipeline = ch.pipeline();
- // pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
- // pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
- pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
- pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
- pipeline.addLast("handler", new TcpClientHandler());
- //pipeline.addLast("log", new LoggingHandler(LogLevel.INFO));
- }
- });
- logger.info("count1>>>>>>" + GlobeTime.BuilValue(GlobeTime.globecount + 1));
- Object messageObject = messageTcpSendQueue.peek();
- if (null == messageObject) {
- // System.out.println(">>>> queue has no data");
- logger.info(">>> Tcp queue size: " + messageTcpSendQueue.size());
- Thread.sleep(1000);
- } else {
- JSONObject object = JSONObject.parseObject(messageObject.toString());
- bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
- // Start the client.
- String currentTime = dateToStamp(object.getJSONObject("content").getString("time"));
- if (!(GlobeTime.globetime).equals(currentTime)) {
- ChannelFuture future = bootstrap.connect(object.getJSONObject("content").getString("targetAddress"), Integer.valueOf(object.getJSONObject("content").getString("port"))).sync();
- try {
- logger.info(">>> Tcp queue size: " + messageTcpSendQueue.size());
- logger.info("连接建立!" + future.channel().id());
- future.channel().writeAndFlush(object.getJSONObject("content").getString("msg")).sync();
- logger.info("sent——to——tcpserver>>>" + object.getJSONObject("content").getString("msg"));
- TimeUnit.SECONDS.sleep(1);
- if (future.isDone()) {
- while (true) {
- ChattingUser info = new ChattingUser();
- if ("ACK!".equals(GlobeTime.ackstatue)) {
- messageTcpSendQueue.poll();
- info.setAck("false");
- GlobeTime.BuildValue(currentTime);
- GlobeTime.BuildAckstatue(null);
- GlobeTime.BuilValue(0);
- break;
- } else {
- continue;
- }
- }
- }
- future.channel().closeFuture().sync();
- // System.out.println("-----------------");
- } catch (Exception e) {
- e.printStackTrace();
- logger.info(">>> 连接未建立>>>Tcp queue size: " + messageTcpSendQueue.size());
- logger.info(object.getJSONObject("content").getString("targetAddress") + ":" + Integer.valueOf(object.getJSONObject("content").getString("port")) + "连接未建立!");
- logger.error("--- tcp_send_error:" + e);
- } finally {
- future.channel().closeFuture().sync();
- }
- }
- }
- // logger.info(">>> kafka message: " + messageObject.toString());
- Thread.sleep(1000 * 5);
- } catch (InterruptedException e) {
- logger.error("--- tcp_send_error:" + e);
- e.printStackTrace();
- } finally {
- group.shutdownGracefully();
- }
- }}catch (Exception e){
- e.printStackTrace();}
- }
- });
- executorService.shutdown();
- }
- }
|