Skip to content

R
Rocketmq

小五666\n哈哈 / Rocketmq
与 Fork 源项目一致

Fork自 Apache RocketMQ / Rocketmq

通知 1
Star 0
Fork 0
R
Rocketmq
提交 ebc0ede8 编写于 作者: Q qqeasonchen
浏览文件
操作

optimization exception declare of request

上级 a8c9fe68
隐藏空白更改
内联 并排
Showing with 61 addition and 67 deletion
client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
浏览文件 @ ebc0ede8
......@@ -1377,32 +1377,28 @@ public class DefaultMQProducerImpl implements MQProducerInner {
}
}
public void request(Message msg, final RequestCallback requestCallback, long timeout) throws RemotingException {
public void request(Message msg, final RequestCallback requestCallback, long timeout)
throws RemotingException, InterruptedException, MQClientException, MQBrokerException {
long beginTimestamp = System.currentTimeMillis();
prepareSendRequest(msg, timeout);
final String requestUniqId = msg.getProperty(MessageConst.PROPERTY_REQUEST_UNIQ_ID);
try {
final RequestResponseFuture requestResponseFuture = new RequestResponseFuture(requestUniqId, timeout, requestCallback);
RequestFutureTable.getRequestFutureTable().put(requestUniqId, requestResponseFuture);
final RequestResponseFuture requestResponseFuture = new RequestResponseFuture(requestUniqId, timeout, requestCallback);
RequestFutureTable.getRequestFutureTable().put(requestUniqId, requestResponseFuture);
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendDefaultImpl(msg, CommunicationMode.ASYNC, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendReqeustOk(true);
}
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendDefaultImpl(msg, CommunicationMode.ASYNC, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendReqeustOk(true);
}
@Override
public void onException(Throwable e) {
requestResponseFuture.setCause(e);
requestFail(requestUniqId);
}
}, timeout - cost);
} catch (Exception ex) {
log.warn("send request message to <{}> failed.", msg.getTopic(), ex);
throw new RemotingSendRequestException(msg.getTopic(), ex);
}
@Override
public void onException(Throwable e) {
requestResponseFuture.setCause(e);
requestFail(requestUniqId);
}
}, timeout - cost);
}
public Message request(final Message msg, final MessageQueueSelector selector, final Object arg,
......@@ -1447,32 +1443,29 @@ public class DefaultMQProducerImpl implements MQProducerInner {
}
public void request(final Message msg, final MessageQueueSelector selector, final Object arg,
final RequestCallback requestCallback, final long timeout) throws RemotingException {
final RequestCallback requestCallback, final long timeout)
throws RemotingException, InterruptedException, MQClientException, MQBrokerException {
long beginTimestamp = System.currentTimeMillis();
prepareSendRequest(msg, timeout);
final String requestUniqId = msg.getProperty(MessageConst.PROPERTY_REQUEST_UNIQ_ID);
try {
final RequestResponseFuture requestResponseFuture = new RequestResponseFuture(requestUniqId, timeout, requestCallback);
RequestFutureTable.getRequestFutureTable().put(requestUniqId, requestResponseFuture);
final RequestResponseFuture requestResponseFuture = new RequestResponseFuture(requestUniqId, timeout, requestCallback);
RequestFutureTable.getRequestFutureTable().put(requestUniqId, requestResponseFuture);
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendReqeustOk(true);
}
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendSelectImpl(msg, selector, arg, CommunicationMode.ASYNC, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendReqeustOk(true);
}
@Override
public void onException(Throwable e) {
requestResponseFuture.setCause(e);
requestFail(requestUniqId);
}
}, timeout - cost);
@Override
public void onException(Throwable e) {
requestResponseFuture.setCause(e);
requestFail(requestUniqId);
}
}, timeout - cost);
} catch (Exception ex) {
log.warn("send request message to <{}> failed.", msg.getTopic(), ex);
throw new RemotingSendRequestException(msg.getTopic(), ex);
}
}
public Message request(final Message msg, final MessageQueue mq, final long timeout)
......@@ -1516,32 +1509,27 @@ public class DefaultMQProducerImpl implements MQProducerInner {
}
public void request(final Message msg, final MessageQueue mq, final RequestCallback requestCallback, long timeout)
throws RemotingException {
throws RemotingException, InterruptedException, MQClientException, MQBrokerException {
long beginTimestamp = System.currentTimeMillis();
prepareSendRequest(msg, timeout);
final String requestUniqId = msg.getProperty(MessageConst.PROPERTY_REQUEST_UNIQ_ID);
try {
final RequestResponseFuture requestResponseFuture = new RequestResponseFuture(requestUniqId, timeout, requestCallback);
RequestFutureTable.getRequestFutureTable().put(requestUniqId, requestResponseFuture);
final RequestResponseFuture requestResponseFuture = new RequestResponseFuture(requestUniqId, timeout, requestCallback);
RequestFutureTable.getRequestFutureTable().put(requestUniqId, requestResponseFuture);
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendReqeustOk(true);
}
long cost = System.currentTimeMillis() - beginTimestamp;
this.sendKernelImpl(msg, mq, CommunicationMode.ASYNC, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendReqeustOk(true);
}
@Override
public void onException(Throwable e) {
requestResponseFuture.setCause(e);
requestFail(requestUniqId);
}
}, null, timeout - cost);
} catch (Exception ex) {
log.warn("send request message to <{}> failed.", msg.getTopic(), ex);
throw new RemotingSendRequestException(msg.getTopic(), ex);
}
@Override
public void onException(Throwable e) {
requestResponseFuture.setCause(e);
requestFail(requestUniqId);
}
}, null, timeout - cost);
}
private void requestFail(final String requestUniqId) {
......
client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
浏览文件 @ ebc0ede8
......@@ -579,6 +579,7 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
* @throws RemotingException if there is any network-tier error.
* @throws MQBrokerException if there is any broker error.
* @throws InterruptedException if the thread is interrupted.
* @throws RequestTimeoutException if request timeout.
*/
@Override
public Message request(final Message msg, final long timeout) throws RequestTimeoutException, MQClientException,
......@@ -601,10 +602,11 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
* @throws MQClientException if there is any client error.
* @throws RemotingException if there is any network-tier error.
* @throws InterruptedException if the thread is interrupted.
* @throws MQBrokerException if there is any broker error.
*/
@Override
public void request(final Message msg, final RequestCallback requestCallback, final long timeout)
throws MQClientException, RemotingException, InterruptedException, MQBrokerException, RequestTimeoutException {
throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
msg.setTopic(withNamespace(msg.getTopic()));
this.defaultMQProducerImpl.request(msg, requestCallback, timeout);
}
......@@ -621,6 +623,7 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
* @throws RemotingException if there is any network-tier error.
* @throws MQBrokerException if there is any broker error.
* @throws InterruptedException if the thread is interrupted.
* @throws RequestTimeoutException if request timeout.
*/
@Override
public Message request(final Message msg, final MessageQueueSelector selector, final Object arg,
......@@ -641,11 +644,12 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
* @throws MQClientException if there is any client error.
* @throws RemotingException if there is any network-tier error.
* @throws InterruptedException if the thread is interrupted.
* @throws MQBrokerException if there is any broker error.
*/
@Override
public void request(final Message msg, final MessageQueueSelector selector, final Object arg,
final RequestCallback requestCallback, final long timeout) throws MQClientException, RemotingException,
InterruptedException {
InterruptedException, MQBrokerException {
msg.setTopic(withNamespace(msg.getTopic()));
this.defaultMQProducerImpl.request(msg, selector, arg, requestCallback, timeout);
}
......@@ -660,6 +664,7 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
* @throws RemotingException if there is any network-tier error.
* @throws MQBrokerException if there is any broker error.
* @throws InterruptedException if the thread is interrupted.
* @throws RequestTimeoutException if request timeout.
*/
@Override
public Message request(final Message msg, final MessageQueue mq, final long timeout)
......@@ -678,10 +683,11 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
* @throws MQClientException if there is any client error.
* @throws RemotingException if there is any network-tier error.
* @throws InterruptedException if the thread is interrupted.
* @throws MQBrokerException if there is any broker error.
*/
@Override
public void request(final Message msg, final MessageQueue mq, final RequestCallback requestCallback, long timeout)
throws MQClientException, RemotingException, InterruptedException {
throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
msg.setTopic(withNamespace(msg.getTopic()));
this.defaultMQProducerImpl.request(msg, mq, requestCallback, timeout);
}
......
client/src/main/java/org/apache/rocketmq/client/producer/MQProducer.java
浏览文件 @ ebc0ede8
......@@ -105,7 +105,7 @@ public interface MQProducer extends MQAdmin {
RemotingException, MQBrokerException, InterruptedException;
void request(final Message msg, final RequestCallback requestCallback, final long timeout)
throws RequestTimeoutException, MQClientException, RemotingException, InterruptedException, MQBrokerException;
throws MQClientException, RemotingException, InterruptedException, MQBrokerException;
Message request(final Message msg, final MessageQueueSelector selector, final Object arg,
final long timeout) throws RequestTimeoutException, MQClientException, RemotingException, MQBrokerException,
......@@ -113,12 +113,12 @@ public interface MQProducer extends MQAdmin {
void request(final Message msg, final MessageQueueSelector selector, final Object arg,
final RequestCallback requestCallback,
final long timeout) throws RequestTimeoutException, MQClientException, RemotingException,
InterruptedException;
final long timeout) throws MQClientException, RemotingException,
InterruptedException, MQBrokerException;
Message request(final Message msg, final MessageQueue mq, final long timeout)
throws RequestTimeoutException, MQClientException, RemotingException, MQBrokerException, InterruptedException;
void request(final Message msg, final MessageQueue mq, final RequestCallback requestCallback, long timeout)
throws RequestTimeoutException, MQClientException, RemotingException, InterruptedException;
throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
}
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或