问题标签 [rebus]

我正在尝试使用 Topshelf 创建一个将作为服务运行的 Rebus 端点。这应该如何设置，有什么例子吗？

将 Rebus 从 0.71.4 升级到 0.75.2 后如何解决 RabbitMq 连接问题？

我们有一个分布式处理器设置，其中有 N 个用于监控活动的仪表板、一个用于控制工作流的管理器和 N 个用于执行工作的工作人员。使用 RabbitMq 管理每个组件之间的通信。[在 Rebus 0.71.4 上] 一切都运行良好，直到我们升级到 Rebus 0.75.2（其中包括将 RabbitMq.Client 升级到 3.4.0.0）。

管理器是最复杂的组件，因为它发送多种形式的通信，包括向工作人员发布的心跳、向仪表板发布的请求/响应审计消息，以及包括工作人员要执行的工作的消息。所有这些消息传递都是同时进行的。

我们使用 1 的预取，以防止工作被长时间运行的工作任务（工作范围在 1 秒到 5 分钟之间）拖住。

现在，我们在管理器中发布期间遇到了一个不稳定的异常。：

2015-01-07 14:59:58.6205 | Error | An error occurred while rolling back the transaction! RabbitMQ.Client.Exceptions.AlreadyClosedException Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=504, text="CHANNEL_ERROR - unexpected command while processing 'tx.commit'", classId=60, methodId=40, cause= at RabbitMQ.Client.Impl.SessionBase.Transmit(Command cmd) at RabbitMQ.Client.Impl.ModelBase.TransmitAndEnqueue(Command cmd, IRpcContinuation k) at RabbitMQ.Client.Impl.ModelBase.ModelRpc(MethodBase method, ContentHeaderBase header, Byte[] body) at RabbitMQ.Client.Framing.Impl.Model.TxRollback() at System.Action.Invoke() at Rebus.Bus.TxBomkarl.RaiseDoRollback() at Rebus.Bus.Worker.<TryProcessIncomingMessage>d__1d.MoveNext()

2015-01-07 14:59:58.6275 | Warn | User exception in Rebus 1 worker 1: Rebus.Bus.QueueCommitException: An exception occurred while attempting to commit the queue transaction ---> RabbitMQ.Client.Exceptions.OperationInterruptedException: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=504, text="CHANNEL_ERROR - unexpected command while processing 'tx.commit'", classId=60, methodId=40, cause= at RabbitMQ.Client.Impl.SimpleBlockingRpcContinuation.GetReply() at RabbitMQ.Client.Impl.ModelBase.ModelRpc(MethodBase method, ContentHeaderBase header, Byte[] body) at RabbitMQ.Client.Framing.Impl.Model.TxCommit() at System.Action.Invoke() at Rebus.Bus.TxBomkarl.RaiseDoCommit() at Rebus.Bus.Worker.<TryProcessIncomingMessage>d__1d.MoveNext() --- End of inner exception stack trace --- at Rebus.Bus.Worker.<TryProcessIncomingMessage>d__1d.MoveNext() --- End of stack trace from previous location where exception was thrown --- at System.Runtime.CompilerServices.AsyncMethodBuilderCore.<ThrowAsync>b__0(Object state) at Rebus.Bus.RebusSynchronizationContext.Run() at Rebus.Bus.Worker.MainLoop()

以下是 RabbitMq 日志中的一些相关条目：

=ERROR REPORT==== 7-Jan-2015::15:41:54 === Error on AMQP connection <0.680.0> ([::1]:59429 -> [::1]:5672, vhost: 'efs', user: 'efs.eod', state: running), channel 1: {amqp_error,channel_error,"unexpected command while processing 'tx.commit'", 'basic.publish'}

=WARNING REPORT==== 7-Jan-2015::15:42:55 === closing AMQP connection <0.553.0> ([::1]:59421 -> [::1]:5672): connection_closed_abruptly

我们已经针对 RabitMq Server 3.3.5 和 3.4.3 进行了测试。

这种行为是可重现的，但是是零星的。而且，当它正常运行时，通信速度比 0.71.4 慢得多。

到目前为止，我们一直无法孤立地重现该问题。任何诊断问题的建议将不胜感激。

我在使用 Rebus 和延迟消息时遇到了一些问题，如下所示：

至少有时我会记录这样的错误，我无法解释，我是否必须关注这个问题？

堆栈跟踪

更新：

我的设置是我有一个发布/订阅者，如果订阅者失败，我希望它在符合其目的的逻辑问题上重试多次。

所以在订阅者我这样设置，我猜这是这些异常的发起者：

所以我相信这意味着它正在使用内部超时管理器并将其存储在我的 SQL Server 中，这在大多数情况下都可以正常工作，但我仍然不时遇到上述异常。

我们在 Azure 服务总线上使用 Rebus。有没有办法以异步方式执行发送消息？我还没有设法在 Rebus 中找到像 SendAsync 这样的东西。

我正在尝试使用（外部）超时服务的 Rebus。

TimeoutService 正确地发回 TimeoutReply。然后订阅者记录一个无法调用 TimeoutReplyHandler 的错误。

我正在使用 Autofac 进行依赖注入并使用以下配置来注册处理程序：

抛出以下异常：

Autofac.Core.DependencyResolutionException：无法使用可用服务和参数调用在类型“Rebus.Bus.TimeoutReplyHandler”上具有“公共绑定标志”的构造函数：无法解析构造函数的参数“Rebus.Bus.IHandleDeferredMessage handleDeferredMessage” '无效.ctor（Rebus.Bus.IHandleDeferredMessage）'。在 Autofac.Core.Activators.Reflection.ReflectionActivator.ActivateInstance（IC omponentContext 上下文，IEnumerable 1 parameters) at Autofac.Core.Resolving.InstanceLookup.Activate(IEnumerable1 参数）在 Autofac.Core.Resolving.InstanceLookup.Execute()
在 Autofac.Core.Resolving.ResolveOperation.GetOrCreateInstance(ISharingLifetimeScope currentOperationScope, IComponentRegistration 注册, IEnumerable`1 个参数）

TimeoutReply 由内部类 TimeoutReplyHandler 处理，它需要 ctor 中的 IHandleDeferredMessage。内部类 DeferredMessageReDispatcher 实现了 IHandleDeferredMessage 并且需要 ctor 中的 IBus。IHandleDeferredMessage 也是内部作用域的。

我应该如何配置 autofac 容器来处理 TimeoutReply？

rebus 什么时候取消订阅？如果订阅者没有运行，是否有任何东西会在一段时间后自动取消订阅？

我遇到过一些情况，当订阅者服务停止一段时间后，发布者不会继续以他们的方式发送消息。我可能做错了什么？

无论如何将传输消息正文类型从字节 [] 更改为字符串。这样做将有助于用户查看队列消息的内容，也有助于将消息写入队列（仅适用于某些支持场景）。所有这些都是必需的，因为我们试图扩展传输以将 TIBCO EMS 队列用于我们的解决方案。请指教。提前致谢。

我正在尝试建立一个通用的 rebus 发布服务，它允许我们的任何 Web 应用程序为订阅者推送事件消息。该服务的唯一工作是维护订阅并重新发布它收到的任何消息。我想避免在服务中创建特定的 IHandleMessage 实现，因为它不需要知道传入消息的类型以外的任何信息。

我尝试了多种方法（例如 IHandleMessages），并且当它尝试反序列化它没有程序集的消息时，rebus 崩溃。

那么从 0.75.1 开始，Rebus 不再支持 ASP.NET 4 了？

我的问题是：

1. 0.75 是 .NET 4 项目的稳定版本吗？
2. 我们是否会继续为 .NET 4 目标项目修复错误（如果有）？

谢谢

我正在探索同时使用 Rebus 和 RabbitMQ 来涵盖几个不同的场景。

场景 A 我希望能够让中央服务器将通知推送到任意订阅者的列表，但这些消息不需要是持久的或持久的。如果订阅者已连接，他们应该会收到通知，但如果他们断开连接，则无需为任何客户端排队消息。

到目前为止，在我的测试中，我能够让生产者和消费者与之通信UseRabbitMqInOneWayMode()，ManageSubscriptions()但是，当没有订阅者或订阅者断开连接时，消息会在 RabbitMQ 中建立。我尝试将标头设置为 false RabbitMqMessageQueue.InternalHeaders.MessageDurability，但没​​有效果。我怀疑这是因为 Rebus 设置的默认队列是持久的。Rebus 中有没有办法控制这种行为？

场景 B 当客户端上线或断开连接时，我想在客户端之间建立一个请求/回复通道。例如：

1. 客户端 A 和客户端 B 连接
2. 客户端 A 将发送一条消息，请求只有客户端 B 拥有的数据。客户 B 收集信息，然后回复给 A。
3. 客户端 B 断开连接
4. 客户端 A 向客户端 B 请求数据，并且由于 B 不再可用而应收到错误消息。

这种情况下推荐的配置是什么？

谢谢。