您的位置: 首页  >  IT文章  >  在ABP core中使用RabbitMq

在ABP core中使用RabbitMq

分类: IT文章 • 2022-04-12 22:38:59

在ABP core中使用RabbitMq

距上一篇博客的更新一集很久了，主要是最近做的事情比较杂，中间也有一个难点，就是在ABP中加入APP扫码登录，本来想些的，但是觉得这个写出来会不会让我们的系统被破解-_-||，所以想了想，就没有写。

这篇文章主要记录一下在ABP core中使用RabbitMq，abp的版本4.30

首先自己基础太菜，为了爬了很久的坑，最后参照的是这篇文章https://www.cnblogs.com/personball/p/7762931.html，使用了Abplus.MqMessages.RebusRabbitMqConsumer插件。本来我老早就看这篇文章了，可是一翻开里面的Github，发现是两年前的代码，以为早就没有人维护，所以就没有用，然后各种兜兜转转，饶了很多圈，中间包括去学习事件总线，ABP的后台任务等等，可是绕了半天，依然没有实现，最终又绕回来，然而让我意外的是，这个插件居然是有人维护的，我最后弄不出了的时候，去issues里提问，作者都回复得很快。

为什么要写这些无关紧要的呢？就是希望看见这篇文章的人，不要再去尝试自己写啦（来自一个绕了一个多星期都没有写出demo的人的忠告），毕竟已经有大神为我们封装好了，何必自己重复造*呢，当然如果你时间比较多，那个另算。

好了 开始我们的代码吧，直接去这里，将作者的源码下下来：https://github.com/personball/abplus，大致浏览一下里面的代码，然后重点是作者的demo，里面写了如何使用发布端和订阅端，重点看以下4个文件

在ABP core中使用RabbitMq

好了，开始搬进你的逻辑吧，中间还出了点小问题，在这里贴出来一下：

如果你像我一样，发布端和消费端在同一个项目中，那么你可以直接使用消费端模块即可，无需使用发布端。在abplus对接rebus的两个模块中，发布模块使用的是单向连接方式，消费端使用的是双向连接方式，即消费端模块同时支持发布和消费

意思已经很清楚了，只需要引入消费模块就可以了

1、在应用层添加 Abplus.MqMessages.RebusRabbitMqConsumer程序集，

2、在module上添加RebusRabbitMqConsumerModule依赖

3、添加消费配置：

            #region RabbitMq消费端配置 
            Configuration.Modules.UseAbplusRebusRabbitMqConsumer()
                .UseLogging(c => c.Log4Net())
                .ConnectTo("amqp://admin:123456@127.0.0.1:5672/")//连接
                .UseQueue("PayQueue")//队列名称
                .Prefetch(100)//用于控制每次拉取的资源消耗(内存,带宽),消费速度还要看消费端自己的消息处理速度
                .SetMaxParallelism(1)//最大并行数
                .SetNumberOfWorkers(10)//设置最大工作线程数
                                       //配置其他选项
                .UseOptions(x => x.SimpleRetryStrategy(maxDeliveryAttempts: 50, secondLevelRetriesEnabled: true, errorTrackingMaxAgeMinutes: 20))
                .RegisterHandlerInAssemblys(Assembly.GetExecutingAssembly());//注册包含Hander的处理程序??//(Assembly.Load("MLCDZ.Application"));
            #endregion RabbitMq消费端配置  Assembly.GetAssembly(typeof(PayHandler))

　其他的就是你的发布端和消费端逻辑，说一个重点：

消费端的handler不能写在AppService里面，需要单独写一个类来实现　

应该说得很清楚了，不懂的请留言，最主要的是看看作者的demo。

补充这后面遇到的一个坑，上次写这篇博客的时候，看到消费端确认就兴高采烈的跑来写博客了，结果在测试的时候，发现消费端没有应答，所以队列里的消息始终没有被消费掉，直到后来得到QQ群友的帮助才知道，原来消费端的类和方法不能有任何授权拦截，比如[AbpAuthorize]等

如果你的报错信息如下，请检查你的逻辑是否带登录验证：

WARN 2019-12-20 11:45:22,511 [ker 1] us.Retry.ErrorTracking.InMemErrorTracker - Unhandled exception 2 while handling message with ID "4cb46c5b-c35f-466f-964a-ae6a5b50875e"
Abp.Authorization.AbpAuthorizationException: 当前用户没有登录到系统！
at Abp.Authorization.AuthorizationHelper.AuthorizeAsync(IEnumerable1 authorizeAttributes) in D:GithubaspnetboilerplatesrcAbpAuthorizationAuthorizationHelper.cs:line 42 at Abp.Authorization.AuthorizationHelper.CheckPermissions(MethodInfo methodInfo, Type type) in D:GithubaspnetboilerplatesrcAbpAuthorizationAuthorizationHelper.cs:line 107 at Abp.Authorization.AuthorizationHelper.AuthorizeAsync(MethodInfo methodInfo, Type type) in D:GithubaspnetboilerplatesrcAbpAuthorizationAuthorizationHelper.cs:line 56 at Nito.AsyncEx.Synchronous.TaskExtensions.WaitAndUnwrapException(Task task) at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state) --- End of stack trace from previous location where exception was thrown --- at System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot) --- End of stack trace from previous location where exception was thrown --- at Nito.AsyncEx.Synchronous.TaskExtensions.WaitAndUnwrapException(Task task) at Nito.AsyncEx.AsyncContext.Run(Func1 action)
at Abp.Authorization.AuthorizationInterceptor.Intercept(IInvocation invocation) in D:GithubaspnetboilerplatesrcAbpAuthorizationAuthorizationInterceptor.cs:line 20
at Castle.DynamicProxy.AbstractInvocation.Proceed()
at Abp.Domain.Uow.UnitOfWorkInterceptor.PerformAsyncUow(IInvocation invocation, UnitOfWorkOptions options) in D:GithubaspnetboilerplatesrcAbpDomainUowUnitOfWorkInterceptor.cs:line 78
at Castle.DynamicProxy.AbstractInvocation.Proceed()
at Abp.Auditing.AuditingInterceptor.PerformAsyncAuditing(IInvocation invocation, AuditInfo auditInfo) in D:GithubaspnetboilerplatesrcAbpAuditingAuditingInterceptor.cs:line 86
at Castle.DynamicProxy.AbstractInvocation.Proceed()
at Castle.DynamicProxy.AbstractInvocation.Proceed()
at Castle.Proxies.IHandleMessages1Proxy.Handle(TestMessage message) at Rebus.Pipeline.Receive.HandlerInvoker1.Invoke() in C:projects-rebusRebusRebusPipelineReceiveHandlerInvoker.cs:line 154
at Rebus.Pipeline.Receive.DispatchIncomingMessageStep.Process(IncomingStepContext context, Func1 next) in C:projects-rebusRebusRebusPipelineReceiveDispatchIncomingMessageStep.cs:line 67 at Rebus.Sagas.LoadSagaDataStep.Process(IncomingStepContext context, Func1 next) in C:projects-rebusRebusRebusSagasLoadSagaDataStep.cs:line 66
at Rebus.Pipeline.Receive.ActivateHandlersStep.Process(IncomingStepContext context, Func1 next) in C:projects-rebusRebusRebusPipelineReceiveActivateHandlersStep.cs:line 47 at Rebus.Pipeline.Receive.HandleRoutingSlipsStep.Process(IncomingStepContext context, Func1 next) in C:projects-rebusRebusRebusPipelineReceiveHandleRoutingSlipsStep.cs:line 42
at Rebus.Retry.Simple.FailedMessageWrapperStep.Process(IncomingStepContext context, Func1 next) in C:projects-rebusRebusRebusRetrySimpleFailedMessageWrapperStep.cs:line 42 at Rebus.Pipeline.Receive.DeserializeIncomingMessageStep.Process(IncomingStepContext context, Func1 next) in C:projects-rebusRebusRebusPipelineReceiveDeserializeIncomingMessageStep.cs:line 34
at Rebus.Pipeline.Receive.HandleDeferredMessagesStep.Process(IncomingStepContext context, Func1 next) in C:projects-rebusRebusRebusPipelineReceiveHandleDeferredMessagesStep.cs:line 121 at Rebus.Retry.FailFast.FailFastStep.Process(IncomingStepContext context, Func1 next) in C:projects-rebusRebusRebusRetryFailFastFailFastStep.cs:line 41
at Rebus.Retry.Simple.SimpleRetryStrategyStep.DispatchWithTrackerIdentifier(Func`1 next, String identifierToTrackMessageBy, ITransactionContext transactionContext, String messageId, String secondLevelMessageId) in C:projects-rebusRebusRebusRetrySimpleSimpleRetryStrategyStep.cs:line 120

相关推荐