Before Creating the Enhancement Request

• I have confirmed that this should be classified as an enhancement rather than a bug/feature.

Summary

ConsumeMessageConcurrentlyService#submitConsumeRequest code should be more concise.

1. Because the consumeExecutor uses a LinkedBlockingQueue, so as with ConsumeMessageConcurrentlyService#submitConsumeRequestLater , exception catching is unnecessary.
2. The code for submitting messages to the consumer in batches should be more concise.
   

   rocketmq/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java

   Lines 187 to 223 in b37d283

   	public void submitConsumeRequest(
   	final List<MessageExt> msgs,
   	final ProcessQueue processQueue,
   	final MessageQueue messageQueue,
   	final boolean dispatchToConsume) {
   	final int consumeBatchSize = this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
   	if (msgs.size() <= consumeBatchSize) {
   	ConsumeRequest consumeRequest = new ConsumeRequest(msgs, processQueue, messageQueue);
   	try {
   	this.consumeExecutor.submit(consumeRequest);
   	} catch (RejectedExecutionException e) {
   	this.submitConsumeRequestLater(consumeRequest);
   	}
   	} else {
   	for (int total = 0; total < msgs.size(); ) {
   	List<MessageExt> msgThis = new ArrayList<>(consumeBatchSize);
   	for (int i = 0; i < consumeBatchSize; i++, total++) {
   	if (total < msgs.size()) {
   	msgThis.add(msgs.get(total));
   	} else {
   	break;
   	}
   	}
   	ConsumeRequest consumeRequest = new ConsumeRequest(msgThis, processQueue, messageQueue);
   	try {
   	this.consumeExecutor.submit(consumeRequest);
   	} catch (RejectedExecutionException e) {
   	for (; total < msgs.size(); total++) {
   	msgThis.add(msgs.get(total));
   	}
   	this.submitConsumeRequestLater(consumeRequest);
   	}
   	}
   	}
   	}

   
   

   rocketmq/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java

   Lines 346 to 356 in b37d283

   	private void submitConsumeRequestLater(final ConsumeRequest consumeRequest
   	) {
   	this.scheduledExecutorService.schedule(new Runnable() {
   	@Override
   	public void run() {
   	ConsumeMessageConcurrentlyService.this.consumeExecutor.submit(consumeRequest);
   	}
   	}, 5000, TimeUnit.MILLISECONDS);
   	}

Motivation

.

Describe the Solution You'd Like

I will optimise the submitConsumeRequest and remove unnecessary exception catching.

Describe Alternatives You've Considered

.

Additional Context

No response