@@ -0,0 +1,2340 @@
+package com.ctrip.hermes.producer;%0A%0Aimport java.util.Random;%0Aimport java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Executors;%0Aimport java.util.concurrent.atomic.AtomicLong;%0A%0Aimport org.unidal.lookup.ComponentTestCase;%0A%0Aimport com.ctrip.hermes.core.result.SendResult;%0Aimport com.ctrip.hermes.metrics.HttpMetricsServer;%0Aimport com.ctrip.hermes.producer.api.Producer;%0Aimport com.google.common.util.concurrent.FutureCallback;%0Aimport com.google.common.util.concurrent.Futures;%0Aimport com.google.common.util.concurrent.SettableFuture;%0A%0A/**%0A * @author Leo Liang(jhliang@ctrip.com)%0A *%0A */%0Apublic class ProducerStressTest extends ComponentTestCase %7B%0A%0A%09public static void main(String%5B%5D args) throws Exception %7B%0A%09%09ExecutorService executors = Executors.newFixedThreadPool(5);%0A%09%09// String topic = %22hello_world%22;%0A%09%09String topic = %22order_new%22;%0A%09%09int bodyLength = 1000;%0A%09%09int count = 10000000;%0A%0A%09%09HttpMetricsServer server = new HttpMetricsServer(%22localhost%22, 9999);%0A%09%09server.start();%0A%0A%09%09System.in.read();%0A%0A%09%09System.out.println(%22Producer started...%22);%0A%0A%09%09AtomicLong sending = new AtomicLong(0);%0A%09%09final AtomicLong sent = new AtomicLong(0);%0A%09%09final AtomicLong failed = new AtomicLong(0);%0A%0A%09%09String body = topic + %22--%22 + generateRandomString(bodyLength);%0A%0A%09%09Producer.getInstance().message(topic, %22prepare-msg%22, topic + %22--prepare-msg%22).sendSync();%0A%0A%09%09while (sending.get() %3C count) %7B%0A%0A%09%09%09long id = sending.incrementAndGet();%0A%09%09%09SettableFuture%3CSendResult%3E future = (SettableFuture%3CSendResult%3E) Producer.getInstance()%0A%09%09%09      .message(topic, body, body).withRefKey(Long.toString(id)).send();%0A%0A%09%09%09Futures.addCallback(future, new FutureCallback%3CSendResult%3E() %7B%0A%0A%09%09%09%09@Override%0A%09%09%09%09public void onSuccess(SendResult result) %7B%0A%09%09%09%09%09sent.incrementAndGet();%0A%09%09%09%09%7D%0A%0A%09%09%09%09@Override%0A%09%09%09%09public void onFailure(Throwable t) %7B%0A%09%09%09%09%09failed.incrementAndGet();%0A%09%09%09%09%7D%0A%09%09%09%7D, executors);%0A%0A%09%09%09long current = sending.get();%0A%0A%09%09%09if (current != 0 && current %25 10000 == 0) %7B%0A%09%09%09%09System.out.println(%22Sending: %22 + sending.get() + %22 Sent: %22 + sent.get() + %22 Failed: %22 + failed.get());%0A%09%09%09%09Thread.sleep(20 * 1000);%0A%09%09%09%7D%0A%09%09%7D%0A%09%7D%0A%0A%09private static String generateRandomString(int size) %7B%0A%09%09StringBuilder sb = new StringBuilder(size);%0A%09%09Random random = new Random();%0A%09%09for (int i = 0; i %3C size; i++) %7B%0A%09%09%09sb.append('a' + random.nextInt(25));%0A%09%09%7D%0A%0A%09%09return sb.toString();%0A%09%7D%0A%0A%7D%0A

add producer stress test