dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/PerformanceClientTest.java - dubbo - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.dubbo.remoting;

 import org.apache.dubbo.common.logger.Logger;
 import org.apache.dubbo.common.logger.LoggerFactory;
 import org.apache.dubbo.remoting.exchange.ExchangeClient;
 import org.apache.dubbo.remoting.exchange.Exchangers;
 import org.apache.dubbo.remoting.exchange.support.ExchangeHandlerAdapter;

 import org.junit.jupiter.api.Test;

 import java.text.DecimalFormat;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;

 import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT;
 import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
 import static org.apache.dubbo.remoting.Constants.CONNECTIONS_KEY;

 /**
  * PerformanceClientTest
  * <p>
  * mvn clean test -Dtest=*PerformanceClientTest -Dserver=10.20.153.187:9911
  */
 public class PerformanceClientTest  {

     private static final Logger logger = LoggerFactory.getLogger(PerformanceClientTest.class);

     @Test
     @SuppressWarnings("unchecked")
     public void testClient() throws Throwable {
         // read server info from property
         if (PerformanceUtils.getProperty("server", null) == null) {
             logger.warn("Please set -Dserver=127.0.0.1:9911");
             return;
         }
         final String server = System.getProperty("server", "127.0.0.1:9911");
         final String transporter = PerformanceUtils.getProperty(Constants.TRANSPORTER_KEY, Constants.DEFAULT_TRANSPORTER);
         final String serialization = PerformanceUtils.getProperty(Constants.SERIALIZATION_KEY, Constants.DEFAULT_REMOTING_SERIALIZATION);
         final int timeout = PerformanceUtils.getIntProperty(TIMEOUT_KEY, DEFAULT_TIMEOUT);
         final int length = PerformanceUtils.getIntProperty("length", 1024);
         final int connections = PerformanceUtils.getIntProperty(CONNECTIONS_KEY, 1);
         final int concurrent = PerformanceUtils.getIntProperty("concurrent", 100);
         int r = PerformanceUtils.getIntProperty("runs", 10000);
         final int runs = r > 0 ? r : Integer.MAX_VALUE;
         final String onerror = PerformanceUtils.getProperty("onerror", "continue");

         final String url = "exchange://" + server + "?transporter=" + transporter + "&serialization=" + serialization + "&timeout=" + timeout;
         // Create clients and build connections
         final ExchangeClient[] exchangeClients = new ExchangeClient[connections];
         for (int i = 0; i < connections; i++) {
             //exchangeClients[i] = Exchangers.connect(url,handler);
             exchangeClients[i] = Exchangers.connect(url);
         }

         List<String> serverEnvironment = (List<String>) exchangeClients[0].request("environment").get();
         List<String> serverScene = (List<String>) exchangeClients[0].request("scene").get();

         // Create some data for test
         StringBuilder buf = new StringBuilder(length);
         for (int i = 0; i < length; i++) {
             buf.append('A');
         }
         final String data = buf.toString();

         // counters
         final AtomicLong count = new AtomicLong();
         final AtomicLong error = new AtomicLong();
         final AtomicLong time = new AtomicLong();
         final AtomicLong all = new AtomicLong();

         // Start multiple threads
         final CountDownLatch latch = new CountDownLatch(concurrent);
         for (int i = 0; i < concurrent; i++) {
             new Thread(new Runnable() {
                 public void run() {
                     try {
                         AtomicInteger index = new AtomicInteger();
                         long init = System.currentTimeMillis();
                         for (int i = 0; i < runs; i++) {
                             try {
                                 count.incrementAndGet();
                                 ExchangeClient client = exchangeClients[index.getAndIncrement() % connections];
                                 long start = System.currentTimeMillis();
                                 String result = (String) client.request(data).get();
                                 long end = System.currentTimeMillis();
                                 if (!data.equals(result)) {
                                     throw new IllegalStateException("Invalid result " + result);
                                 }
                                 time.addAndGet(end - start);
                             } catch (Exception e) {
                                 error.incrementAndGet();
                                 e.printStackTrace();
                                 if ("exit".equals(onerror)) {
                                     System.exit(-1);
                                 } else if ("break".equals(onerror)) {
                                     break;
                                 } else if ("sleep".equals(onerror)) {
                                     try {
                                         Thread.sleep(30000);
                                     } catch (InterruptedException e1) {
                                     }
                                 }
                             }
                         }
                         all.addAndGet(System.currentTimeMillis() - init);
                     } finally {
                         latch.countDown();
                     }
                 }
             }).start();
         }

         // Output, tps is not for accuracy, but it reflects the situation to a certain extent.
         new Thread(new Runnable() {
             public void run() {
                 try {
                     SimpleDateFormat dateFormat = new SimpleDateFormat("HH:mm:ss");
                     long lastCount = count.get();
                     long sleepTime = 2000;
                     long elapsd = sleepTime / 1000;
                     boolean bfirst = true;
                     while (latch.getCount() > 0) {
                         long c = count.get() - lastCount;
                         if (!bfirst)// The first time is inaccurate.
                             System.out.println("[" + dateFormat.format(new Date()) + "] count: " + count.get() + ", error: " + error.get() + ",tps:" + (c / elapsd));

                         bfirst = false;
                         lastCount = count.get();
                         Thread.sleep(sleepTime);
                     }
                 } catch (Exception e) {
                     e.printStackTrace();
                 }
             }
         }).start();

         latch.await();

         for (ExchangeClient client : exchangeClients) {
             if (client.isConnected()) {
                 client.close();
             }
         }

         long total = count.get();
         long failed = error.get();
         long succeeded = total - failed;
         long elapsed = time.get();
         long allElapsed = all.get();
         long clientElapsed = allElapsed - elapsed;
         long art = 0;
         long qps = 0;
         long throughput = 0;
         if (elapsed > 0) {
             art = elapsed / succeeded;
             qps = concurrent * succeeded * 1000 / elapsed;
             throughput = concurrent * succeeded * length * 2 * 1000 / elapsed;
         }

         PerformanceUtils.printBorder();
         PerformanceUtils.printHeader("Dubbo Remoting Performance Test Report");
         PerformanceUtils.printBorder();
         PerformanceUtils.printHeader("Test Environment");
         PerformanceUtils.printSeparator();
         for (String item : serverEnvironment) {
             PerformanceUtils.printBody("Server " + item);
         }
         PerformanceUtils.printSeparator();
         List<String> clientEnvironment = PerformanceUtils.getEnvironment();
         for (String item : clientEnvironment) {
             PerformanceUtils.printBody("Client " + item);
         }
         PerformanceUtils.printSeparator();
         PerformanceUtils.printHeader("Test Scene");
         PerformanceUtils.printSeparator();
         for (String item : serverScene) {
             PerformanceUtils.printBody("Server " + item);
         }
         PerformanceUtils.printBody("Client Transporter: " + transporter);
         PerformanceUtils.printBody("Serialization: " + serialization);
         PerformanceUtils.printBody("Response Timeout: " + timeout + " ms");
         PerformanceUtils.printBody("Data Length: " + length + " bytes");
         PerformanceUtils.printBody("Client Shared Connections: " + connections);
         PerformanceUtils.printBody("Client Concurrent Threads: " + concurrent);
         PerformanceUtils.printBody("Run Times Per Thread: " + runs);
         PerformanceUtils.printSeparator();
         PerformanceUtils.printHeader("Test Result");
         PerformanceUtils.printSeparator();
         PerformanceUtils.printBody("Succeeded Requests: " + DecimalFormat.getIntegerInstance().format(succeeded));
         PerformanceUtils.printBody("Failed Requests: " + failed);
         PerformanceUtils.printBody("Client Elapsed Time: " + clientElapsed + " ms");
         PerformanceUtils.printBody("Average Response Time: " + art + " ms");
         PerformanceUtils.printBody("Requests Per Second: " + qps + "/s");
         PerformanceUtils.printBody("Throughput Per Second: " + DecimalFormat.getIntegerInstance().format(throughput) + " bytes/s");
         PerformanceUtils.printBorder();
     }

     static class PeformanceTestHandler extends ExchangeHandlerAdapter {

         @Override
         public void connected(Channel channel) throws RemotingException {
             System.out.println("connected event,chanel;" + channel);
         }

         @Override
         public void disconnected(Channel channel) throws RemotingException {
             System.out.println("disconnected event,chanel;" + channel);
         }
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.dubbo.remoting;

	import org.apache.dubbo.common.logger.Logger;
	import org.apache.dubbo.common.logger.LoggerFactory;
	import org.apache.dubbo.remoting.exchange.ExchangeClient;
	import org.apache.dubbo.remoting.exchange.Exchangers;
	import org.apache.dubbo.remoting.exchange.support.ExchangeHandlerAdapter;

	import org.junit.jupiter.api.Test;

	import java.text.DecimalFormat;
	import java.text.SimpleDateFormat;
	import java.util.Date;
	import java.util.List;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.atomic.AtomicLong;

	import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT;
	import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
	import static org.apache.dubbo.remoting.Constants.CONNECTIONS_KEY;

	/**
	* PerformanceClientTest
	* <p>
	* mvn clean test -Dtest=*PerformanceClientTest -Dserver=10.20.153.187:9911
	*/
	public class PerformanceClientTest {

	private static final Logger logger = LoggerFactory.getLogger(PerformanceClientTest.class);

	@Test
	@SuppressWarnings("unchecked")
	public void testClient() throws Throwable {
	// read server info from property
	if (PerformanceUtils.getProperty("server", null) == null) {
	logger.warn("Please set -Dserver=127.0.0.1:9911");
	return;
	}
	final String server = System.getProperty("server", "127.0.0.1:9911");
	final String transporter = PerformanceUtils.getProperty(Constants.TRANSPORTER_KEY, Constants.DEFAULT_TRANSPORTER);
	final String serialization = PerformanceUtils.getProperty(Constants.SERIALIZATION_KEY, Constants.DEFAULT_REMOTING_SERIALIZATION);
	final int timeout = PerformanceUtils.getIntProperty(TIMEOUT_KEY, DEFAULT_TIMEOUT);
	final int length = PerformanceUtils.getIntProperty("length", 1024);
	final int connections = PerformanceUtils.getIntProperty(CONNECTIONS_KEY, 1);
	final int concurrent = PerformanceUtils.getIntProperty("concurrent", 100);
	int r = PerformanceUtils.getIntProperty("runs", 10000);
	final int runs = r > 0 ? r : Integer.MAX_VALUE;
	final String onerror = PerformanceUtils.getProperty("onerror", "continue");

	final String url = "exchange://" + server + "?transporter=" + transporter + "&serialization=" + serialization + "&timeout=" + timeout;
	// Create clients and build connections
	final ExchangeClient[] exchangeClients = new ExchangeClient[connections];
	for (int i = 0; i < connections; i++) {
	//exchangeClients[i] = Exchangers.connect(url,handler);
	exchangeClients[i] = Exchangers.connect(url);
	}

	List<String> serverEnvironment = (List<String>) exchangeClients[0].request("environment").get();
	List<String> serverScene = (List<String>) exchangeClients[0].request("scene").get();

	// Create some data for test
	StringBuilder buf = new StringBuilder(length);
	for (int i = 0; i < length; i++) {
	buf.append('A');
	}
	final String data = buf.toString();

	// counters
	final AtomicLong count = new AtomicLong();
	final AtomicLong error = new AtomicLong();
	final AtomicLong time = new AtomicLong();
	final AtomicLong all = new AtomicLong();

	// Start multiple threads
	final CountDownLatch latch = new CountDownLatch(concurrent);
	for (int i = 0; i < concurrent; i++) {
	new Thread(new Runnable() {
	public void run() {
	try {
	AtomicInteger index = new AtomicInteger();
	long init = System.currentTimeMillis();
	for (int i = 0; i < runs; i++) {
	try {
	count.incrementAndGet();
	ExchangeClient client = exchangeClients[index.getAndIncrement() % connections];
	long start = System.currentTimeMillis();
	String result = (String) client.request(data).get();
	long end = System.currentTimeMillis();
	if (!data.equals(result)) {
	throw new IllegalStateException("Invalid result " + result);
	}
	time.addAndGet(end - start);
	} catch (Exception e) {
	error.incrementAndGet();
	e.printStackTrace();
	if ("exit".equals(onerror)) {
	System.exit(-1);
	} else if ("break".equals(onerror)) {
	break;
	} else if ("sleep".equals(onerror)) {
	try {
	Thread.sleep(30000);
	} catch (InterruptedException e1) {
	}
	}
	}
	}
	all.addAndGet(System.currentTimeMillis() - init);
	} finally {
	latch.countDown();
	}
	}
	}).start();
	}

	// Output, tps is not for accuracy, but it reflects the situation to a certain extent.
	new Thread(new Runnable() {
	public void run() {
	try {
	SimpleDateFormat dateFormat = new SimpleDateFormat("HH:mm:ss");
	long lastCount = count.get();
	long sleepTime = 2000;
	long elapsd = sleepTime / 1000;
	boolean bfirst = true;
	while (latch.getCount() > 0) {
	long c = count.get() - lastCount;
	if (!bfirst)// The first time is inaccurate.
	System.out.println("[" + dateFormat.format(new Date()) + "] count: " + count.get() + ", error: " + error.get() + ",tps:" + (c / elapsd));

	bfirst = false;
	lastCount = count.get();
	Thread.sleep(sleepTime);
	}
	} catch (Exception e) {
	e.printStackTrace();
	}
	}
	}).start();

	latch.await();

	for (ExchangeClient client : exchangeClients) {
	if (client.isConnected()) {
	client.close();
	}
	}

	long total = count.get();
	long failed = error.get();
	long succeeded = total - failed;
	long elapsed = time.get();
	long allElapsed = all.get();
	long clientElapsed = allElapsed - elapsed;
	long art = 0;
	long qps = 0;
	long throughput = 0;
	if (elapsed > 0) {
	art = elapsed / succeeded;
	qps = concurrent * succeeded * 1000 / elapsed;
	throughput = concurrent * succeeded * length * 2 * 1000 / elapsed;
	}

	PerformanceUtils.printBorder();
	PerformanceUtils.printHeader("Dubbo Remoting Performance Test Report");
	PerformanceUtils.printBorder();
	PerformanceUtils.printHeader("Test Environment");
	PerformanceUtils.printSeparator();
	for (String item : serverEnvironment) {
	PerformanceUtils.printBody("Server " + item);
	}
	PerformanceUtils.printSeparator();
	List<String> clientEnvironment = PerformanceUtils.getEnvironment();
	for (String item : clientEnvironment) {
	PerformanceUtils.printBody("Client " + item);
	}
	PerformanceUtils.printSeparator();
	PerformanceUtils.printHeader("Test Scene");
	PerformanceUtils.printSeparator();
	for (String item : serverScene) {
	PerformanceUtils.printBody("Server " + item);
	}
	PerformanceUtils.printBody("Client Transporter: " + transporter);
	PerformanceUtils.printBody("Serialization: " + serialization);
	PerformanceUtils.printBody("Response Timeout: " + timeout + " ms");
	PerformanceUtils.printBody("Data Length: " + length + " bytes");
	PerformanceUtils.printBody("Client Shared Connections: " + connections);
	PerformanceUtils.printBody("Client Concurrent Threads: " + concurrent);
	PerformanceUtils.printBody("Run Times Per Thread: " + runs);
	PerformanceUtils.printSeparator();
	PerformanceUtils.printHeader("Test Result");
	PerformanceUtils.printSeparator();
	PerformanceUtils.printBody("Succeeded Requests: " + DecimalFormat.getIntegerInstance().format(succeeded));
	PerformanceUtils.printBody("Failed Requests: " + failed);
	PerformanceUtils.printBody("Client Elapsed Time: " + clientElapsed + " ms");
	PerformanceUtils.printBody("Average Response Time: " + art + " ms");
	PerformanceUtils.printBody("Requests Per Second: " + qps + "/s");
	PerformanceUtils.printBody("Throughput Per Second: " + DecimalFormat.getIntegerInstance().format(throughput) + " bytes/s");
	PerformanceUtils.printBorder();
	}

	static class PeformanceTestHandler extends ExchangeHandlerAdapter {

	@Override
	public void connected(Channel channel) throws RemotingException {
	System.out.println("connected event,chanel;" + channel);
	}

	@Override
	public void disconnected(Channel channel) throws RemotingException {
	System.out.println("disconnected event,chanel;" + channel);
	}
	}
	}