从 Micronaut 中的 Listener 方法抛出 RabbitListenerException

问题描述

我在 Micronaut 中有一个消费者异常处理程序，如下所示

@Singleton
@Primary
@Replaces(DefaultRabbitListenerExceptionHandler.class)
public class RabbitListenerCustomExceptionHandler implements RabbitListenerExceptionHandler {
    private static final Logger LOG = LoggerFactory.getLogger(RabbitListenerCustomExceptionHandler.class);
    @Override
    public void handle(RabbitListenerException exception) {
        if (LOG.isErrorEnabled()) {
            Optional<RabbitConsumerState> messageState = exception.getMessageState();
            if (messageState.isPresent()) {
                LOG.error("Error processing a message for RabbitMQ consumer [" + exception.getListener() + "]",exception);
            } else {
                LOG.error("RabbitMQ consumer [" + exception.getListener() + "] produced an error",exception);
            }
        }
    }
}

从监听器方法，我想抛出一个异常，然后在上面的方法中捕获。

@RabbitListener
public class SubCategoryListener {

@Queue(ConstantValues.ADD_SUB_CATEGORY)
    public void Create(SubCategoryViewModel model) {
            LOG.info(String.format("Listener --> Adding the sub category collection"));
            Category category = new Category();
            category.setSubCategory(List.of(new CategorySubCategory(model.name(),model.description())));
            SubCategorySearchCriteria criteria = new SubCategorySearchCriteria();
            Document document = UpdateQueryBuilder.QueryBuilder(model,Document.class);
            Bson queryId = QueryBuilder.QueryBuilder(criteria,Bson.class).get(0);
            Single.fromPublisher(
                    this.repository.getCollection(ConstantValues.PRODUCT_CATEGORY_COLLECTION_NAME,Category.class)
                            .updateOne(queryId,document))
                    .subscribe(item -> {},error ->{
                        throw new RabbitListenerException(error,null,null);
                    });
    }

}

来自 Micronaut 库的 RabbitListenerException 包含以下参数

//Micronaut rabbit library
     public RabbitListenerException(Throwable cause,Object listener,@Nullable RabbitConsumerState messageState) {
            super(cause.getMessage(),cause);
            this.listener = listener;
            this.messageState = messageState;
        }

我不确定如何从我的侦听器 Create 方法中传递 listener 和 messageState。

更新 1

如果我不抛出异常，则下面的语句会出现 RXJava 异常 (io.reactivex.exceptions)

Single.fromPublisher(
                  this.repository.getCollection(ConstantValues.PRODUCT_CATEGORY_COLLECTION_NAME,document))
                    .subscribe(item -> {});

例外

Caused by: com.mongodb.MongoWriteException: Modifiers operate on fields but we found type array instead. For example: {$mod: {<field>: ...}} not {$push: [ { subCategory.name: "This is name update",subCategory.description: "This is update" } ]}
    at com.mongodb.internal.async.client.AsyncMongoCollectionImpl.lambda$executeSingleWriteRequest$9(AsyncMongoCollectionImpl.java:1075)
    ... 48 more
Exception in thread "Thread-1" io.reactivex.exceptions.OnErrorNotImplementedException: The exception was not handled due to missing onError handler in the subscribe() method call. Further reading: https://github.com/ReactiveX/RxJava/wiki/Error-Handling | com.mongodb.MongoWriteException: Modifiers operate on fields but we found type array instead. For example: {$mod: {<field>: ...}} not {$push: [ { subCategory.name: "This is name update",subCategory.description: "This is update" } ]}
    at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
    at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
    at io.reactivex.internal.observers.ConsumerSingleObserver.onError(ConsumerSingleObserver.java:46)
    at io.reactivex.internal.operators.single.SingleFromPublisher$ToSingleObserver.onError(SingleFromPublisher.java:87)
    at com.mongodb.reactivestreams.client.internal.AbstractSubscription.onError(AbstractSubscription.java:142)
    at com.mongodb.reactivestreams.client.internal.SingleResultCallbackSubscription.lambda$requestInitialData$0(SingleResultCallbackSubscription.java:41)
    at com.mongodb.internal.async.client.AsyncMongoCollectionImpl.lambda$executeUpdate$4(AsyncMongoCollectionImpl.java:711)
    at com.mongodb.internal.async.client.AsyncMongoCollectionImpl.lambda$executeSingleWriteRequest$9(AsyncMongoCollectionImpl.java:1080)
    at com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:48)
    at com.mongodb.internal.async.client.OperationExecutorImpl$2$1$1.onResult(OperationExecutorImpl.java:135)
    at com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:48)
    at com.mongodb.internal.operation.OperationHelper$ConnectionReleasingWrappedCallback.onResult(OperationHelper.java:551)
    at com.mongodb.internal.operation.MixedBulkWriteOperation.addBatchResult(MixedBulkWriteOperation.java:524)
    at com.mongodb.internal.operation.MixedBulkWriteOperation.access$1700(MixedBulkWriteOperation.java:76)
    at com.mongodb.internal.operation.MixedBulkWriteOperation$6.onResult(MixedBulkWriteOperation.java:506)
    at com.mongodb.internal.operation.MixedBulkWriteOperation$6.onResult(MixedBulkWriteOperation.java:476)
    at com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:48)
    at com.mongodb.internal.connection.DefaultServer$DefaultServerProtocolExecutor$2.onResult(DefaultServer.java:288)
    at com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:48)
    at com.mongodb.internal.connection.CommandProtocolImpl$1.onResult(CommandProtocolImpl.java:84)
    at com.mongodb.internal.connection.DefaultConnectionPool$PooledConnection$2.onResult(DefaultConnectionPool.java:530)
    at com.mongodb.internal.connection.UsageTrackingInternalConnection$2.onResult(UsageTrackingInternalConnection.java:142)
    at com.mongodb.internal.async.ErrorHandlingResultCallback.onResult(ErrorHandlingResultCallback.java:48)
    at com.mongodb.internal.connection.InternalStreamConnection$2$1.onResult(InternalStreamConnection.java:463)
    at com.mongodb.internal.connection.InternalStreamConnection$2$1.onResult(InternalStreamConnection.java:440)
    at com.mongodb.internal.connection.InternalStreamConnection$MessageHeaderCallback$MessageCallback.onResult(InternalStreamConnection.java:745)
    at com.mongodb.internal.connection.InternalStreamConnection$MessageHeaderCallback$MessageCallback.onResult(InternalStreamConnection.java:712)
    at com.mongodb.internal.connection.InternalStreamConnection$5.completed(InternalStreamConnection.java:582)
    at com.mongodb.internal.connection.InternalStreamConnection$5.completed(InternalStreamConnection.java:579)
    at com.mongodb.internal.connection.AsynchronousChannelStream$BasicCompletionHandler.completed(AsynchronousChannelStream.java:250)
    at com.mongodb.internal.connection.AsynchronousChannelStream$BasicCompletionHandler.completed(AsynchronousChannelStream.java:233)
    at java.base/sun.nio.ch.Invoker.invokeUnchecked(Invoker.java:127)
    at java.base/sun.nio.ch.Invoker.invokeDirect(Invoker.java:158)
    at java.base/sun.nio.ch.UnixAsynchronousSocketChannelImpl.implRead(UnixAsynchronousSocketChannelImpl.java:568)
    at java.base/sun.nio.ch.AsynchronousSocketChannelImpl.read(AsynchronousSocketChannelImpl.java:276)
    at java.base/sun.nio.ch.AsynchronousSocketChannelImpl.read(AsynchronousSocketChannelImpl.java:297)
    at com.mongodb.internal.connection.AsynchronousSocketChannelStream$AsynchronousSocketChannelAdapter.read(AsynchronousSocketChannelStream.java:144)
    at com.mongodb.internal.connection.AsynchronousChannelStream.readAsync(AsynchronousChannelStream.java:118)
    at com.mongodb.internal.connection.AsynchronousChannelStream.readAsync(AsynchronousChannelStream.java:107)
    at com.mongodb.internal.connection.InternalStreamConnection.readAsync(InternalStreamConnection.java:579)
    at com.mongodb.internal.connection.InternalStreamConnection.access$1100(InternalStreamConnection.java:78)
    at com.mongodb.internal.connection.InternalStreamConnection$MessageHeaderCallback.onResult(InternalStreamConnection.java:702)
    at com.mongodb.internal.connection.InternalStreamConnection$MessageHeaderCallback.onResult(InternalStreamConnection.java:687)
    at com.mongodb.internal.connection.InternalStreamConnection$5.completed(InternalStreamConnection.java:582)
    at com.mongodb.internal.connection.InternalStreamConnection$5.completed(InternalStreamConnection.java:579)
    at com.mongodb.internal.connection.AsynchronousChannelStream$BasicCompletionHandler.completed(AsynchronousChannelStream.java:250)
    at com.mongodb.internal.connection.AsynchronousChannelStream$BasicCompletionHandler.completed(AsynchronousChannelStream.java:233)
    at java.base/sun.nio.ch.Invoker.invokeUnchecked(Invoker.java:127)
    at java.base/sun.nio.ch.UnixAsynchronousSocketChannelImpl.finishRead(UnixAsynchronousSocketChannelImpl.java:443)
    at java.base/sun.nio.ch.UnixAsynchronousSocketChannelImpl.finish(UnixAsynchronousSocketChannelImpl.java:193)
    at java.base/sun.nio.ch.UnixAsynchronousSocketChannelImpl.onEvent(UnixAsynchronousSocketChannelImpl.java:215)
    at java.base/sun.nio.ch.KQueuePort$EventHandlerTask.run(KQueuePort.java:312)
    at java.base/sun.nio.ch.AsynchronousChannelGroupImpl$1.run(AsynchronousChannelGroupImpl.java:112)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630)
    at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: com.mongodb.MongoWriteException: Modifiers operate on fields but we found type array instead. For example: {$mod: {<field>: ...}} not {$push: [ { subCategory.name: "This is name update",subCategory.description: "This is update" } ]}
    at com.mongodb.internal.async.client.AsyncMongoCollectionImpl.lambda$executeSingleWriteRequest$9(AsyncMongoCollectionImpl.java:1075)
    ... 48 more

上述异常在RabbitListenerCustomExceptionHandler

中没有被捕获

解决方法

你根本不应该抛出那个异常。您抛出的任何异常都将为您包装在 RabbitListenerException 中。