c# - 如何获取 Websphere MQ 连接状态以及如何重置连接 :

Question

1.) 从 .net 客户端，我如何测试客户端是否连接到服务器(即可以发送和接收)是的，我可以在 try block 内发送消息并捕获随后的异常，但我'我希望有一个更优雅的解决方案。

2) 如何打开、关闭和重新打开连接？在尝试解决上面的问题 1 时，我发现如果我打开一个连接然后调用 connection.Close()，我将无法从连接工厂获得另一个连接(请参见下面的代码片段)。我收到错误消息 XMSCC0008

我使用的是非常标准的 vanilla MQ 配置。以下是我的客户端的连接方式:

ISession session = MQAccess.GetSession(MQAccess.Connection);
IDestination destination = session.CreateTopic(SubTopicName);
Consumer = MQAccess.GetConsumer(session, destination);
Consumer.MessageListener = new MessageListener(HandleMQSubEvent);
MQAccess.Connection.Start();

其中 MQAccess 是一个小型实用程序类。

编辑问题以添加 MQAccess 代码:

public static class MQAccess
{
    public static readonly MQConfigurationSectionHandler ConfigSettings;
    public static readonly IConnectionFactory ConnectionFactory;

    private static readonly IConnection connection;
    public static IConnection Connection
    {
        get { return connection; }
    }

    static MQAccess()
    {
        ConfigSettings = (MQConfigurationSectionHandler)
            ConfigurationManager.GetSection("mq-configuration");

        XMSFactoryFactory factory = XMSFactoryFactory.GetInstance(XMSC.CT_WMQ);
        ConnectionFactory = factory.CreateConnectionFactory();
        ConnectionFactory.SetStringProperty(XMSC.WMQ_HOST_NAME, ConfigSettings.Hostname);
        ConnectionFactory.SetIntProperty(XMSC.WMQ_PORT, ConfigSettings.Port);
        ConnectionFactory.SetStringProperty(XMSC.WMQ_CHANNEL, ConfigSettings.Channel);

        if (ConfigSettings.QueueManager == string.Empty)
        {
            ConnectionFactory.SetStringProperty(XMSC.WMQ_QUEUE_MANAGER, "");
        }
        else
        {
            ConnectionFactory.SetStringProperty(XMSC.WMQ_QUEUE_MANAGER, ConfigSettings.QueueManager);
        }

        connection = GetConnection();
    }

    public static IConnection GetConnection()
    {
        return ConnectionFactory.CreateConnection();
    }

    public static ISession GetSession(IConnection connection)
    {
        return connection.CreateSession(false, AcknowledgeMode.AutoAcknowledge);
    }

    public static IMessageProducer GetProducer(ISession session, IDestination destination)
    {
        return session.CreateProducer(destination);
    }

    public static IMessageConsumer GetConsumer(ISession session, IDestination destination)
    {
        return session.CreateConsumer(destination);
    }

    public static void MQPub(string TopicURI, string message)
    {
        using (var session = GetSession(Connection))
        {
            using (var destination = session.CreateTopic(TopicURI))
            {
                using (var producer = GetProducer(session, destination))
                {
                    producer.Send(session.CreateTextMessage(message));
                }
            }
        }
    }

    public static void MQPub(string TopicURI, IEnumerable<string> messages)
    {
        using (var session = GetSession(Connection))
        {
            using (var destination = session.CreateTopic(TopicURI))
            {
                using (var producer = GetProducer(session, destination))
                {
                    foreach (var message in messages)
                    {
                        producer.Send(session.CreateTextMessage(message));
                    }
                }
            }
        }
    }
}

编辑:将 MQAccess 类重命名为 MQClient。根据 T Rob 的建议将其设为实例类。断开连接方法仍然崩溃并显示上面列出的错误消息

public class MQClient : IDisposable
{
    public MQConfigurationSectionHandler ConfigSettings { get; private set; }
    public IConnectionFactory ConnectionFactory { get; private set; }

    public IConnection Connection { get; private set;  }

    public IMessageConsumer Consumer { get; private set; }
    public IMessageProducer Producer { get; private set; }
    // Save sessions as fields for disposing and future subscription functionality
    private ISession ProducerSession;
    private ISession ConsumerSession;
    public string SubTopicName { get; private set; }
    public string PubTopicName { get; private set; }
    public bool IsConnected { get; private set; }
    public event Action<Exception> ConnectionError;
    private Action<IMessage> IncomingMessageHandler;

    public MQClient(string subTopicName, string pubTopicName, Action<IMessage> incomingMessageHandler)
    {
        // Dont put connect logic in the constructor.  If we lose the connection we may need to connect again.
        SubTopicName = subTopicName;
        PubTopicName = pubTopicName;
        IncomingMessageHandler = incomingMessageHandler;
    }

    public string Connect()
    {
        IsConnected = false;
        string errorMsg = string.Empty;

        ConfigSettings = (MQConfigurationSectionHandler)
                ConfigurationManager.GetSection("mq-configuration");

        XMSFactoryFactory factory = XMSFactoryFactory.GetInstance(XMSC.CT_WMQ);
        ConnectionFactory = factory.CreateConnectionFactory();
        ConnectionFactory.SetStringProperty(XMSC.WMQ_HOST_NAME, ConfigSettings.Hostname);
        ConnectionFactory.SetIntProperty(XMSC.WMQ_PORT, ConfigSettings.Port);
        ConnectionFactory.SetStringProperty(XMSC.WMQ_CHANNEL, ConfigSettings.Channel);

        if (ConfigSettings.QueueManager == string.Empty)
            ConnectionFactory.SetStringProperty(XMSC.WMQ_QUEUE_MANAGER, "");
        else
            ConnectionFactory.SetStringProperty(XMSC.WMQ_QUEUE_MANAGER, ConfigSettings.QueueManager);

        Connection = ConnectionFactory.CreateConnection();


        if (!string.IsNullOrEmpty(PubTopicName))
        {
            ProducerSession = Connection.CreateSession(false, AcknowledgeMode.AutoAcknowledge);
            Producer = ProducerSession.CreateProducer(ProducerSession.CreateTopic(PubTopicName));
        }

        if (!string.IsNullOrEmpty(SubTopicName) && IncomingMessageHandler != null)
        {
            ConsumerSession = Connection.CreateSession(false, AcknowledgeMode.AutoAcknowledge);
            Consumer = ConsumerSession.CreateConsumer(ConsumerSession.CreateTopic(SubTopicName));
            Consumer.MessageListener = new MessageListener(IncomingMessageHandler);
        }

        try
        {
            Connection.Start();
            Connection.ExceptionListener = new ExceptionListener(ConnectionExceptionHandler);
            IsConnected = true;
        }
        catch (TypeInitializationException ex)
        {
            errorMsg = "A TypeInitializationException error occured while attempting to connect to MQ.  Check the Queue configuration in App.config. The error message is: " + ex.Message; 
        }
        catch (IllegalStateException ex)
        {
            errorMsg = "An IllegalStateException error occured while attempting to connect to MQ.  Check the Queue configuration in App.config. The error message is: " + ex.Message; 
        }

        return errorMsg;
    }

    public void Disconnect()
    {
        if (Producer != null)
        {
            Producer.Close();
            Producer.Dispose();
            Producer = null;
        }

        if (ProducerSession != null)
        {
            // Call Unsubscribe here if subscription is durable

            ProducerSession.Close();
            ProducerSession.Dispose();
            ProducerSession = null;
        }

        if (Connection != null)
        {
            Connection.Stop();

            //if (Connection.ExceptionListener != null)
            //    Connection.ExceptionListener = null;

            // Per Shashi............
            //if (Consumer.MessageListener != null)
            //    Consumer.MessageListener = null;

            Connection.Close();
            Connection.Dispose();
            Connection = null;
        }

        if (Consumer != null)
        {

            if (Consumer.MessageListener != null)
                Consumer.MessageListener = null;

            Consumer.Close();
            Consumer.Dispose();
            Consumer = null;
        }


        if (ConsumerSession != null)
        {
            // Call Unsubscribe here if subscription is durable
            ConsumerSession.Close();
            ConsumerSession.Dispose();
            ConsumerSession = null;
        }

        IsConnected = false;
    }


    public void Publish(string message)
    {
        Producer.Send(ProducerSession.CreateTextMessage(message));
    }


    public void Publish(string[] messages)
    {
        foreach (string msg in messages)
            Publish(msg);
    }

    public void ConnectionExceptionHandler(Exception ex)
    {
        Disconnect(); // Clean up

        if (ConnectionError != null)
            ConnectionError(ex);
    }

    #region IDisposable Members
    private bool disposed;

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    protected virtual void Dispose(bool disposing)
    {
        if (!this.disposed)
        {
            if (disposing)
                Disconnect();

            disposed = true;
        }
    }
    #endregion

}

Answer 1

问题就在这里 --> MQAccess 是一个小的实用类。

问题的第一部分询问如何判断连接是否处于事件状态。 WebSphere MQ 的 XMS 类是非 Java 平台的 JMS 规范的实现。它们非常严格地遵循 JMS 规范，而 JMS 规范在连接或 session 上没有等效于 isConnected 的方法，因此 XMS 也没有。但是，所有 GET 和 PUT 事件都应在 try/catch block 中发生，以便捕获 JMS 异常。 (您总是从中打印linkedException，对吗？)当抛出 JMS 异常时，应用要么将其视为致命异常并终止，要么关闭所有 JMS 对象，除了连接工厂，等待几秒钟，然后重新驱动连接序列。

根据问题中的新信息进行更新:
感谢您发布 MQAccess 类。这提供了对正在发生的事情的相当深入的了解，尽管仍然没有任何代码显示根据问题的第 2 部分关闭和重新打开连接的位置。

但是，代码显示 MQAccess 类在构造类实例时创建了 ICONNECTION 连接 的私有(private)实例，然后公开公开为 MQAccess .GetConnection。当前发布的 MQAccess 类没有公共(public)或私有(private)类方法可以替换 connection 持有的连接句柄，因此如果 MQAccess.Connection.Close() 被调用后，MQAccess 类中的 IConnection 对象实例将永远持有无效的连接句柄。一旦连接关闭，该 MQAccess 实例实际上就死了。您必须删除并重新实例化 MQAccess 才能获得新连接。

MQAccess 类确实公开了连接工厂，因此理论上可以从类外部调用 MQAccess.GetConnection 并获得有效的新 IConnection 对象，即使在关闭原始对象之后。但是，该实例将存在于 MQAccess 类的范围之外，因此对 MQAccess 的任何后续调用都将引用其失效的实例变量 connection 而不是比在类外创建的新连接实例。

如果您需要关闭并重新创建连接，您可以考虑从 MQAccess 内部进行管理。一种低技术含量的方法可能是为连接编写一个 MQAccess.Close() 方法，该方法将关闭现有连接，然后立即调用 connection = GetConnection(); 以便私有(private) connection 变量始终持有有效的连接句柄。

如果这不能解决问题，请发布关闭并重新创建连接的代码。

顺便说一句，网络连接上的非事务性 session 为任何 JMS 提供程序(包括 WMQ)打开了丢失或复制消息的可能性。这是你想要的吗？我已经解释了为什么这是在其他 SO 帖子中 here .