2

我正在 c# 中实现 Active MQ 发布者和订阅者。我正在使用 Apache.NMS.ActiveMQ .net 客户端库与代理进行通信。

  <package id="Apache.NMS" version="1.7.1" targetFramework="net461" />
  <package id="Apache.NMS.ActiveMQ" version="1.7.2" targetFramework="net461" />

ActiveMQ 在 4 个服务器上配置了故障转移设置（.225、.226、.346、.347 - IP 的最后一部分供参考）。经纪人网址看起来像

故障转移：//tcp://103.24.34.225:61616,tcp://103.24.34.226:61616,tcp://103.24.34.346:61616,tcp://103.24.34.347:61616

这是我发布的方式

    var brokerUrl = "failover://tcp://103.24.34.225:61616,tcp://103.24.34.226:61616,tcp://103.24.34.346:61616,tcp://103.24.34.347:61616";
    var connectionFactory = new ConnectionFactory(brokerUrl);
    using (var connection = connectionFactory.CreateConnection("conn1", "conn$34"))
    {
        connection.ClientId = "TESTPUBLISHER";
        connection.AcknowledgementMode = AcknowledgementMode.ClientAcknowledge;
        connection.Start();

        var session = connection.CreateSession();
        var topic = new ActiveMQTopic("ACCOUNT.UPDATE");
        var producer = session.CreateProducer(topic);


        var msg = "43342_test"; //DateTime.Now.ToString("yyyyMdHHmmss_fff") + "-TEST";
        var textMessage = producer.CreateTextMessage(msg);

        textMessage.Properties.SetString("Topic", "ACCOUNT.UPDATE");
        textMessage.Properties.SetString("Action", "UPDATE");
        textMessage.Properties.SetString("DataContractType", "Account");

        producer.Send(textMessage, MsgDeliveryMode.Persistent, MsgPriority.Normal, new TimeSpan(0, 0, 60, 0, 0));

     }

这是我订阅该主题的方式。此代码设置为，多个共享订阅者可以侦听传入消息。有人告诉我，我必须使用虚拟主题来实现这一点。因此，我将订阅者配置为使用虚拟主题，并且它托管在 Windows 服务项目中。我使用确认模式作为 ClientAcknowledge，因此除非消息被确认，否则它应该不断返回。下面的代码片段仅代表 Windows 服务的重要订阅者部分。

var brokerUrl = "failover://tcp://103.24.34.225:61616,tcp://103.24.34.226:61616,tcp://103.24.34.346:61616,tcp://103.24.34.347:61616"; 
IConnectionFactory factory = new ConnectionFactory(new Uri(brokerUrl));
IConnection connection = factory.CreateConnection("conn1", "conn$34"))

    connection.ClientId = "TESTSUBSCRIBER";
    connection.AcknowledgementMode = AcknowledgementMode.ClientAcknowledge;
    connection.ConnectionInterruptedListener += OnConnectionInturrupted;
    connection.ExceptionListener += OnConnectionExceptionListener;
    connection.ConnectionResumedListener += OnConnectionResumedListener;
    connection.Start();

    ISession session = connection.CreateSession(AcknowledgementMode.ClientAcknowledge);
    var queue = new ActiveMQQueue("VT.TESTSUBSCRIBER.ACCOUNT.UPDATE");
    ActiveMQTopic topic = new ActiveMQTopic();
    IMessageConsumer consumer = session.CreateConsumer(queue); 

    consumer.Listener += OnMessage;



private void OnMessage(IMessage message)
{
    var payload = ((ITextMessage)message).Text;
    Log.Info($"Received message for Client TESTSUBSCRIBER - [{payload}]");
    if(payload != "43342_test")
    {
        message.Acknowledge(); 
            Log.Info($"Message acknowledged for Client TESTSUBSCRIBER - [{payload}]");  
    }
}

private void OnConnectionResumedListener()
{
    Log.Info($"Subscriber connection resumed for Client TESTSUBSCRIBER");
}

private void OnConnectionExceptionListener(Exception exception)
{
    Log.Error(exception);
}

private void OnConnectionInturrupted()
{
    Log.Error($"Subscriber connection interrupted for Client TESTSUBSCRIBER");
}

我能够发布和订阅消息。我遇到了一个特定案例的问题。假设订户从故障转移服务器池中建立了到（.225 代理服务器）的连接。发布者发布了一条消息。订户收到它，它正在处理中。但由于一些服务器补丁维护，windows服务不得不关闭。结果，与代理的订阅者连接断开了。当 Windows 服务恢复时，这一次订阅者建立了与故障转移池中的不同代理服务器（.346 代理服务器）的连接。发生这种情况时，未确认的消息永远不会重新传递。但是，如果我重新启动 Windows 服务并且运气好的话，如果建立了与 .225 代理的连接（与订阅者最初连接的同一服务器），

我的假设是，当在故障转移设置中配置 ActiveMQ 时，无论订阅者能够与故障转移池中的哪个代理服务器建立连接，它都应该始终接收未确认的消息。

在某些情况下，故障转移设置似乎可以正常工作。假设订阅者从故障转移池连接到 .346 代理服务器。发布者从同一个池连接到不同的代理服务器（.225 代理）并发布消息，订阅者正在接收消息。这证明故障转移设置正在运行。

但是，一旦订阅者从代理服务器接收到消息，并且如果订阅者在确认消息之前断开连接，则它必须重新建立与同一代理服务器的连接才能接收未确认的消息。这对我来说听起来不对。

Active MQ 服务器设置是否需要任何其他配置才能使此用例正常工作？

4

1 回答 1

0

此问题的解决方案不是在客户端，而是使用 Active MQ 服务器配置。

对于 Producer 流控目标策略，添加 ConditionalNetworkBridgeFilter 并启用 replayWhenNoConsumers。

于 2017-02-28T20:34:54.037 回答