代码部分片段。
TransactionListener transactionListener = new TransactionListenerImpl();
TransactionMQProducer producer = new TransactionMQProducer("transaction_producer");
producer.setNamesrvAddr("192.168.52.128:9876");
ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue(2000), new ThreadFactory() {
@OverRide
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("client-transaction-msg-check-thread");
return thread;
}
});
producer.setExecutorService(executorService);
producer.setTransactionListener(transactionListener);
producer.start();
String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 3; i++) {
try {
Message msg =
new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
("shenxinquan " + i*100).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.sendMessageInTransaction(msg, "params");
System.out.printf("%s%n", sendResult);
// Thread.sleep(100);
} catch (MQClientException | UnsupportedEncodingException e) {
e.printStackTrace();
}
}
for (int i = 0; i <60; i++) {
Thread.sleep(1000);
}
producer.shutdown();
描述：当producer 设置自定义线程池(executorService)的时候 producer 调用shutdown 方法的时候会报如下错误
Exception in thread "main" java.lang.NullPointerException
at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.destroyTransactionEnv(DefaultMQProducerImpl.java:135)
at org.apache.rocketmq.client.producer.TransactionMQProducer.shutdown(TransactionMQProducer.java:50)
at com.mmall.concurrency.rocketmq.TransactionProducer.main(TransactionProducer.java:89)
主要是 DefaultMQProducerImpl 类 方法
public void destroyTransactionEnv() {
this.checkExecutor.shutdown();
this.checkRequestQueue.clear();
}
里面的this.checkRequestQueue.clear(); 报错
分析主要原因是 DefaultMQProducerImpl 里面的initTransactionEnv()方法 当有ExecutorService线程池的时候 this.checkRequestQueue 没初始化导致的。
代码片段如下
public void initTransactionEnv() {
TransactionMQProducer producer = (TransactionMQProducer) this.defaultMQProducer;
if (producer.getExecutorService() != null) {
this.checkExecutor = producer.getExecutorService();
} else {
this.checkRequestQueue = new LinkedBlockingQueue(2000);
this.checkExecutor = new ThreadPoolExecutor(
1,
1,
1000 * 60,
TimeUnit.MILLISECONDS,
this.checkRequestQueue);
}
}
代码部分片段。
TransactionListener transactionListener = new TransactionListenerImpl();
TransactionMQProducer producer = new TransactionMQProducer("transaction_producer");
producer.setNamesrvAddr("192.168.52.128:9876");
ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue(2000), new ThreadFactory() {
@OverRide
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("client-transaction-msg-check-thread");
return thread;
}
});
producer.setExecutorService(executorService);
producer.setTransactionListener(transactionListener);
producer.start();
String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 3; i++) {
try {
Message msg =
new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
("shenxinquan " + i*100).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.sendMessageInTransaction(msg, "params");
System.out.printf("%s%n", sendResult);
// Thread.sleep(100);
} catch (MQClientException | UnsupportedEncodingException e) {
e.printStackTrace();
}
}
for (int i = 0; i <60; i++) {
Thread.sleep(1000);
}
producer.shutdown();
描述：当producer 设置自定义线程池(executorService)的时候 producer 调用shutdown 方法的时候会报如下错误
Exception in thread "main" java.lang.NullPointerException
at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.destroyTransactionEnv(DefaultMQProducerImpl.java:135)
at org.apache.rocketmq.client.producer.TransactionMQProducer.shutdown(TransactionMQProducer.java:50)
at com.mmall.concurrency.rocketmq.TransactionProducer.main(TransactionProducer.java:89)
主要是 DefaultMQProducerImpl 类 方法
public void destroyTransactionEnv() {
this.checkExecutor.shutdown();
this.checkRequestQueue.clear();
}
里面的this.checkRequestQueue.clear(); 报错
分析主要原因是 DefaultMQProducerImpl 里面的initTransactionEnv()方法 当有ExecutorService线程池的时候 this.checkRequestQueue 没初始化导致的。
代码片段如下
public void initTransactionEnv() {
TransactionMQProducer producer = (TransactionMQProducer) this.defaultMQProducer;
if (producer.getExecutorService() != null) {
this.checkExecutor = producer.getExecutorService();
} else {
this.checkRequestQueue = new LinkedBlockingQueue(2000);
this.checkExecutor = new ThreadPoolExecutor(
1,
1,
1000 * 60,
TimeUnit.MILLISECONDS,
this.checkRequestQueue);
}
}