Skip to content

Commit 3c9ad2c

Browse files
bxq2011hustbxq2011hust
authored
Merge pull request #345 from FISCO-BCOS/release-2.0.5
Async can initialize the thread pool by constructor
2 parents 255e90b + 7691b87 commit 3c9ad2c

File tree

4 files changed

+167
-8
lines changed

4 files changed

+167
-8
lines changed

‎src/main/java/org/fisco/bcos/channel/client/Service.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -913,7 +913,7 @@ public void onReceiveTransactionMessage(ChannelHandlerContext ctx, BcosMessage m
913913

914914
public String newSeq() {
915915
String seq = UUID.randomUUID().toString().replaceAll("-", "");
916-
logger.info("New Seq" + seq);
916+
logger.debug("New Seq" + seq);
917917
return seq;
918918
}
919919

‎src/main/java/org/fisco/bcos/web3j/protocol/channel/ChannelEthereumService.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,7 @@ public <T extends Response> T send(Request request, Class<T> responseType) throw
5555
channelService.sendEthereumMessage(
5656
bcosRequest, request.getTransactionSucCallback());
5757
}
58-
logger.info(
58+
logger.debug(
5959
"bcos request, seq:{}, method:{}", bcosRequest.getMessageID(), request.getMethod());
6060
logger.debug(
6161
"bcos request:{} {}",

‎src/main/java/org/fisco/bcos/web3j/utils/Async.java

+28-6
Original file line numberDiff line numberDiff line change
@@ -4,15 +4,23 @@
44

55
import java.util.concurrent.*;
66

7+
import org.slf4j.Logger;
8+
import org.slf4j.LoggerFactory;
9+
710
/** Async task facilitation. */
811
public class Async {
9-
private static final ExecutorService executor = Executors.newFixedThreadPool(web3AsyncPoolSize);
10-
11-
static {
12-
Runtime.getRuntime().addShutdownHook(new Thread(() -> shutdown(executor)));
13-
}
12+
13+
static Logger logger = LoggerFactory.getLogger(Async.class);
14+
15+
private static Executor executor;
1416

15-
public static <T> CompletableFuture<T> run(Callable<T> callable) {
17+
public static <T> CompletableFuture<T> run(Callable<T> callable) {
18+
19+
if (null == executor) {
20+
logger.info(" default set setExeutor , pool size is {}", web3AsyncPoolSize);
21+
setExeutor(Executors.newFixedThreadPool(web3AsyncPoolSize), true);
22+
}
23+
1624
CompletableFuture<T> result = new CompletableFuture<>();
1725
CompletableFuture.runAsync(
1826
() -> {
@@ -67,4 +75,18 @@ private static void shutdown(ExecutorService executorService) {
6775
Thread.currentThread().interrupt();
6876
}
6977
}
78+
79+
public static synchronized void setExeutor(Executor pool, boolean setIfNull) {
80+
if(null == Async.executor && setIfNull) {
81+
Async.executor = pool;
82+
logger.info(" set setExeutor because executor null, executor is {}", pool.toString());
83+
} else if(!setIfNull) {
84+
Async.executor = pool;
85+
logger.info(" set setExeutor even executor already exist, executor is {}", pool.toString());
86+
}
87+
}
88+
89+
public Async(Executor pool) {
90+
setExeutor(pool, true);
91+
}
7092
}

‎src/test/java/org/fisco/bcos/channel/test/contract/PerformanceOkDSync.java

+137
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,137 @@
1+
package org.fisco.bcos.channel.test.contract;
2+
3+
import com.google.common.util.concurrent.RateLimiter;
4+
5+
import java.math.BigInteger;
6+
import java.util.Random;
7+
import java.util.concurrent.Executors;
8+
import java.util.concurrent.ScheduledExecutorService;
9+
import java.util.concurrent.atomic.AtomicInteger;
10+
import org.fisco.bcos.channel.client.Service;
11+
import org.fisco.bcos.web3j.crypto.Credentials;
12+
import org.fisco.bcos.web3j.protocol.Web3j;
13+
import org.fisco.bcos.web3j.protocol.channel.ChannelEthereumService;
14+
import org.fisco.bcos.web3j.protocol.core.methods.response.TransactionReceipt;
15+
import org.fisco.bcos.web3j.utils.Async;
16+
import org.slf4j.Logger;
17+
import org.slf4j.LoggerFactory;
18+
import org.springframework.context.ApplicationContext;
19+
import org.springframework.context.support.ClassPathXmlApplicationContext;
20+
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
21+
22+
public class PerformanceOkDSync {
23+
private static Logger logger = LoggerFactory.getLogger(PerformanceOkDSync.class);
24+
private static AtomicInteger sended = new AtomicInteger(0);
25+
26+
public static void main(String[] args) throws Exception {
27+
try {
28+
String groupId = args[3];
29+
ApplicationContext context =
30+
new ClassPathXmlApplicationContext("classpath:applicationContext.xml");
31+
Service service = context.getBean(Service.class);
32+
service.setGroupId(Integer.parseInt(groupId));
33+
service.run();
34+
35+
System.out.println("Start test...");
36+
System.out.println(
37+
"===================================================================");
38+
39+
ChannelEthereumService channelEthereumService = new ChannelEthereumService();
40+
channelEthereumService.setChannelService(service);
41+
42+
if(args.length > 4) {
43+
Integer threadPoolSize = Integer.parseInt(args[4]);
44+
Async async = new Async(Executors.newFixedThreadPool(threadPoolSize));
45+
System.out.println(" === thread pool size = " + threadPoolSize);
46+
}
47+
48+
ScheduledExecutorService scheduledExecutorService =
49+
Executors.newScheduledThreadPool(500);
50+
Web3j web3 =
51+
Web3j.build(
52+
channelEthereumService,
53+
15 * 100,
54+
scheduledExecutorService,
55+
Integer.parseInt(groupId));
56+
57+
Credentials credentials =
58+
Credentials.create(
59+
"b83261efa42895c38c6c2364ca878f43e77f3cddbc922bf57d0d48070f79feb6");
60+
61+
BigInteger gasPrice = new BigInteger("30000000");
62+
BigInteger gasLimit = new BigInteger("30000000");
63+
64+
String command = args[0];
65+
Integer count = 0;
66+
Integer qps = 0;
67+
68+
switch (command) {
69+
case "trans":
70+
count = Integer.parseInt(args[1]);
71+
qps = Integer.parseInt(args[2]);
72+
73+
break;
74+
default:
75+
System.out.println("Args: <trans> <Total> <QPS> <GroupID> <ThreadPoolSize>");
76+
}
77+
78+
ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
79+
threadPool.setCorePoolSize(200);
80+
threadPool.setMaxPoolSize(500);
81+
threadPool.setQueueCapacity(count);
82+
83+
threadPool.initialize();
84+
85+
System.out.println("Deploying contract...");
86+
OkD ok = OkD.deploy(web3, credentials, gasPrice, gasLimit).send();
87+
88+
PerformanceCollector collector = new PerformanceCollector();
89+
collector.setTotal(count);
90+
91+
RateLimiter limiter = RateLimiter.create(qps);
92+
Integer area = count / 10;
93+
final Integer total = count;
94+
95+
Random random = new Random(System.currentTimeMillis());
96+
97+
System.out.println("Start test，total：" + count);
98+
for (Integer i = 0; i < count; ++i) {
99+
threadPool.execute(
100+
new Runnable() {
101+
@Override
102+
public void run() {
103+
limiter.acquire();
104+
PerformanceOkCallback callback = new PerformanceOkCallback();
105+
callback.setCollector(collector);
106+
try {
107+
TransactionReceipt receipt = ok.trans(
108+
String.valueOf(random.nextLong()),
109+
new BigInteger("1")).sendAsync().get();
110+
callback.onResponse(receipt);
111+
} catch (Exception e) {
112+
TransactionReceipt receipt = new TransactionReceipt();
113+
receipt.setStatus("-1");
114+
115+
callback.onResponse(receipt);
116+
logger.error("Error sending:", e);
117+
}
118+
119+
int current = sended.incrementAndGet();
120+
121+
if (current >= area && ((current % area) == 0)) {
122+
System.out.println(
123+
"Already sended: "
124+
+ current
125+
+ "/"
126+
+ total
127+
+ " transactions");
128+
}
129+
}
130+
});
131+
}
132+
} catch (Exception e) {
133+
e.printStackTrace();
134+
System.exit(-1);
135+
}
136+
}
137+
}

0 commit comments

Comments
 (0)