import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * javac -encoding UTF-8 C10kTestClientConcurrent.java
 *
 * https://mp.weixin.qq.com/s/VdyXDBevE48Wtr95ug_aKw
 *
 * 代码比较简单,直接使用BIO串行创建30000个连接。
 */
public class C10kTestClientConcurrent {
// static String ip = "192.168.110.35";
 static String ip = "";
 private static int port = 666;

 public static void main(String[] args) throws IOException {
 performanceTest();
// httpAsyncTest();
 }

 private static void performanceTest() throws IOException {
 ExecutorService executorService = Executors.newFixedThreadPool(16);
 Queue<SocketChannel> clients = new ConcurrentLinkedQueue<SocketChannel>();
 new Thread(new Runnable() {
 @Override
 public void run() {
 /**
 * 每秒发一次心跳
 */
 while (true){
 try {
 long start = System.currentTimeMillis();
 for (SocketChannel client : clients) {
// client.write(ByteBuffer.wrap(("1"+EchoConstant.SEPARATOR).getBytes(StandardCharsets.UTF_8)));
 client.write(ByteBuffer.wrap(("biz json data$").getBytes(StandardCharsets.UTF_8)));
 }
 System.out.println("span millis time: [{}]" + (System.currentTimeMillis() - start));
// TimeUnit.SECONDS.sleep(1);
 TimeUnit.MILLISECONDS.sleep(1000);
 System.out.println("heart Beat " + clients.size());
 } catch (IOException e) {
 e.printStackTrace();
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 }
 },"heartBeat").start();
 InetSocketAddress serverAddr = new InetSocketAddress(ip, port);
 String str = "Hello, World!";
 // 字符串到 ByteBuffer 的转换
 ByteBuffer byteBuffer = ByteBuffer.wrap(str.getBytes(StandardCharsets.UTF_8));
// IntStream.range(20000, 50000).forEach(i -> {
 AtomicInteger count = new AtomicInteger();
// IntStream.range(20000, 20200).forEach(i -> {
 long start = System.currentTimeMillis();

// for (int i = 20000; i < 30000; i++) {
 for (int i = 20000; i < 25000; i++) {
// for (int i = 20000; i < 21000; i++) {
 executorService.submit(new Runnable() {
 @Override
 public void run() {
 try {
 SocketChannel client = SocketChannel.open();
// client.bind(new InetSocketAddress(ip, i));
 long startRope = System.currentTimeMillis();
 client.connect(serverAddr);
 long endRope = System.currentTimeMillis() - startRope;
 System.out.println(endRope);
 //内容为客户端的时间
 Integer devid = count.incrementAndGet();
 String uploadMsg = getConnectMsg(devid);
 client.write(ByteBuffer.wrap(uploadMsg.getBytes(StandardCharsets.UTF_8)));
 System.out.println("client:" + count.get() + " connected " + devid);
 clients.add(client);
// TimeUnit.MILLISECONDS.sleep(2);
 } catch (Exception e) {
 System.out.println("IOException" + count.get());
 e.printStackTrace();
 }
 }
 });
 }

 long x = System.currentTimeMillis() - start;
 System.out.println(x/1000);
 System.out.println("clients.size: " + clients.size());

 //阻塞主线程
 System.in.read();
 }

 private static String getConnectMsg(Integer devid) {
 /*Message message = new Message();
 message.setIsFirstConnect("1");
 message.setDevid(devid + "");
 message.setIemi("1111111111v004");
 message.setMode(0);
 return message.toJSON() + "$";*/
 return "1$";
 }
}