2

以下代码片段包含我的 IBM MQ 队列的连接和订阅逻辑。当出现连接失败时，我使用 IConnection.ExceptionListener 委托通过队列建立新连接并重新订阅消息。但问题是，我可以看到多个队列句柄。我如何确保关闭以前的连接句柄并在由于网络问题或 MQ 服务器重新启动而导致连接中断的情况下建立新连接？

private CancellationToken _cancellationToken;
private IConnection _connection;
private IConnectionFactory _connectionfactory;
private IMessageConsumer _consumer;
private IDestination _destination;
private MessageFormat _msgFormat;
private IMessageProducer _producer;
private ISession _session;

private void CreateWebsphereQueueConnection () {
    SetConnectionFactory ();

    //Connection
    _connection = _connectionfactory.CreateConnection (null, null);
    _connection.ExceptionListener = new ExceptionListener (OnConnectionException);

    //Session
    _session = _connection.CreateSession (false, AcknowledgeMode.AutoAcknowledge);

    //Destination
    _destination = _session.CreateQueue ("queue://My.Queue.Name");
    _destination.SetIntProperty (XMSC.DELIVERY_MODE, 2);
    _destination.SetIntProperty (XMSC.WMQ_TARGET_CLIENT, 0);

    //Consumer
    _consumer = _session.CreateConsumer (_destination);
}

private IConnectionFactory SetConnectionFactory () {
    XMSFactoryFactory factoryFactory = XMSFactoryFactory.GetInstance (XMSC.CT_WMQ);
    IConnectionFactory cf = factoryFactory.CreateConnectionFactory ();

    // Set the properties
    cf.SetStringProperty (XMSC.WMQ_CHANNEL, ConnectionSettings.Channel);
    cf.SetIntProperty (XMSC.WMQ_CONNECTION_MODE, XMSC.WMQ_CM_CLIENT);
    cf.SetIntProperty (XMSC.WMQ_FAIL_IF_QUIESCE, XMSC.WMQ_FIQ_YES);
    cf.SetStringProperty (XMSC.WMQ_QUEUE_MANAGER, ConnectionSettings.QueueManager);
    cf.SetStringProperty (XMSC.WMQ_CONNECTION_NAME_LIST, ConnectionSettings.ConnectionList);
    cf.SetIntProperty (XMSC.WMQ_CLIENT_RECONNECT_TIMEOUT, ConnectionSettings.ReconnectTimeout);
    cf.SetIntProperty (XMSC.WMQ_CLIENT_RECONNECT_OPTIONS, ConnectionSettings.ReconnectOptions);

    cf.SetStringProperty (XMSC.WMQ_PROVIDER_VERSION, XMSC.WMQ_PROVIDER_VERSION_DEFAULT);
    cf.SetBooleanProperty (XMSC.WMQ_SYNCPOINT_ALL_GETS, true);
    return cf;
}

public override void Subscribe<T> (Action<T> onMessageReceived) {
    try {

        _connection.ExceptionListener = delegate (Exception connectionException) {
            //Using any of these two statements is termination my code. Debugger doesn't move to CreateWebsphereQueueConnection() line of code at all
            //_conection.Stop()
            //_conection.Close()
            CreateWebsphereQueueConnection ();
            Subscribe (onMessageReceived);
        };

        MessageListener messageListener = new MessageListener ((msg) => {
            onMessageReceived (message);
        });
        _consumer.MessageListener = messageListener;

        // Start the connection
        _connection.Start ();
    } catch (Exception ex) {
        //Log exception details
    }
}
4

3 回答 3

2

您已经在连接工厂上设置了重新连接选项。XMS 库将在与队列管理器的连接中断时自动重新连接，除非队列管理器在没有 -r 或 -s 选项的情况下关闭。因此，您的应用程序不需要显式重新连接。拥有异常侦听器将有助于了解重新连接过程的情况。

于 2018-01-29T03:34:43.610 回答
0

IBM.XMS.dll 将处理 MQ 故障转移或使用 -r 开关完成的重新启动。但是，如果在要求连接的客户端重新连接的情况下重新启动，XMS 库将不会尝试重新连接，并且客户将不得不手动处理这种情况，正如@Shashi 和 @JoshMc 所指出的那样。

我必须处理这种情况并更改我的 Connection ExceptionListener 如下帮助我：

private CancellationToken _cancellationToken;
private IConnection _connection;
private IConnectionFactory _connectionfactory;
private IMessageConsumer _consumer;
private IDestination _destination;
private MessageFormat _msgFormat;
private IMessageProducer _producer;
private ISession _session;
private bool _reConnectOnConnectionBreak = false;
private bool _connected = false;
private void CreateWebsphereQueueConnection () {
    SetConnectionFactory ();

    while (!_connected || _reConnectOnConnectionBreak) {
        try {
            //Connection
            _connection = _connectionfactory.CreateConnection (null, null);
            _connection.ExceptionListener = new ExceptionListener (OnConnectionException);

            //Session
            _session = _connection.CreateSession (false, AcknowledgeMode.AutoAcknowledge);

            //Destination
            _destination = _session.CreateQueue ("queue://My.Queue.Name");
            _destination.SetIntProperty (XMSC.DELIVERY_MODE, 2);
            _destination.SetIntProperty (XMSC.WMQ_TARGET_CLIENT, 0);

            //Consumer
            _consumer = _session.CreateConsumer (_destination);
            _connected = true;
        } catch (Exception ex) {
            _connected = false;
        }

    }
}

private IConnectionFactory SetConnectionFactory () {
    XMSFactoryFactory factoryFactory = XMSFactoryFactory.GetInstance (XMSC.CT_WMQ);
    IConnectionFactory cf = factoryFactory.CreateConnectionFactory ();

    // Set the properties
    cf.SetStringProperty (XMSC.WMQ_CHANNEL, ConnectionSettings.Channel);
    cf.SetIntProperty (XMSC.WMQ_CONNECTION_MODE, XMSC.WMQ_CM_CLIENT);
    cf.SetIntProperty (XMSC.WMQ_FAIL_IF_QUIESCE, XMSC.WMQ_FIQ_YES);
    cf.SetStringProperty (XMSC.WMQ_QUEUE_MANAGER, ConnectionSettings.QueueManager);
    cf.SetStringProperty (XMSC.WMQ_CONNECTION_NAME_LIST, ConnectionSettings.ConnectionList);
    cf.SetIntProperty (XMSC.WMQ_CLIENT_RECONNECT_TIMEOUT, ConnectionSettings.ReconnectTimeout);
    cf.SetIntProperty (XMSC.WMQ_CLIENT_RECONNECT_OPTIONS, ConnectionSettings.ReconnectOptions);

    cf.SetStringProperty (XMSC.WMQ_PROVIDER_VERSION, XMSC.WMQ_PROVIDER_VERSION_DEFAULT);
    cf.SetBooleanProperty (XMSC.WMQ_SYNCPOINT_ALL_GETS, true);
    return cf;
}

public override void Subscribe<T> (Action<T> onMessageReceived) {
    try {

        _connection.ExceptionListener = delegate (Exception connectionException) {
            XMSException xmsError = (XMSException) connectionException;
            int reasonCode = ((IBM.WMQ.MQException) (xmsError).LinkedException).ReasonCode;
            if (reasonCode == MQC.MQRC_Q_MGR_QUIESCING || reasonCode == MQC.MQRC_CONNECTION_BROKEN) {
                _reConnectOnConnectionBreak = true;
                _connection.Close ();

                CreateWebsphereQueueConnection ();
                Subscribe (onMessageReceived);
                _reConnectOnConnectionBreak = false;
            }
        }

        MessageListener messageListener = new MessageListener ((msg) => {
            onMessageReceived (message);
        });
        _consumer.MessageListener = messageListener;

        // Start the connection
        _connection.Start ();
    } catch (Exception ex) {
        //Log exception details
    }
}

没有更好的方法来检查 IBM MQ 版本 8 中连接 IConnection 的状态。所以，我不得不使用原因代码。在 IBM MQ 版本 9 中，我们可以使用服务器公开的其余 API 来检查连接状态。

于 2018-02-05T14:04:06.520 回答
0

对于我的服务，我将其合并CreateWebsphereQueueConnection()为Subscribe<T>()一个Connect()方法。

和：

connectionFactory.SetIntProperty(XMSC.WMQ_CLIENT_RECONNECT_OPTIONS, XMSC.WMQ_CLIENT_RECONNECT);                           
connectionFactory.SetIntProperty(XMSC.WMQ_CLIENT_RECONNECT_TIMEOUT, 3600);                          
connectionFactory.SetIntProperty(XMSC.WMQ_CONNECTION_MODE, XMSC.WMQ_CM_CLIENT_UNMANAGED);

var queueConnection = connectionFactory.CreateConnection();
queueConnection.ExceptionListener = OnException;

然后我处理这样的异常：

private void OnException(Exception exception)
{
    Policy.Handle<Exception>()
        .WaitAndRetryForever(retryAttempt => TimeSpan.FromSeconds(5), (ex, timespan) =>
        {
            _logger.Warning($"Unable to connect: {ex.Message}.");
        })
        .Execute(CreateWebsphereQueueConnection);
}

重试很重要，因为您不知道重新连接需要多长时间。

于 2018-01-29T04:49:22.913 回答