如何不断从远程IBM MQ获取消息

Question

我创建了一个windows服务，该服务将连接到远程MQ并以MQSTR格式获取消息，但在获取消息后，我没有关闭与远程MQ的连接。我的windows服务将不断检查数据是否在远程MQ中可用，但在获得一条消息后，我需要重新启动服务以从远程MQ获得另一条消息。有人能告诉我需要做什么才能不断地从远程MQ获取消息吗。任何线索或任何环节都可以。请帮助

我的C#窗口服务代码如下：

程序.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.ServiceProcess;
using System.Text;
using System.Threading.Tasks;
namespace MQ_listner
{
static class Program
{
static void Main()
{
ServiceBase[] ServicesToRun;
ServicesToRun = new ServiceBase[]
{
new Service1()
};
ServiceBase.Run(ServicesToRun);

}
}
}

Service1.cs

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Diagnostics;
using System.Linq;
using System.ServiceProcess;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace MQ_listner
{
public partial class Service1 : ServiceBase
{
private MQReader MQReader;
private string _serviceName = "MQ_Listener";
private DateTime _TimeStart;
private bool _run = true; 
private Thread _thread;
int WaitWhenStop = 0;
private DateTime _TimeEnd;
private TimeSpan _TimeDifference;
private TimeSpan _TimeElasped = new TimeSpan(0);

public Service1()
{
InitializeComponent();
}
protected override void OnStart(string[] args)
{
try
{
EventLog.WriteEntry(_serviceName + "was started at" + _TimeStart.ToString());
_run = true;
_thread = new Thread(new ThreadStart(StartMQListenerService));
_thread.IsBackground = true;
_thread.Start();
}
catch (Exception ex)
{
EventLog.WriteEntry(_serviceName + "was not started . Error Message : " + ex.ToString());
}

}
protected override void OnStop()
{
_run = false;
_thread.Join(WaitWhenStop);
_TimeEnd = DateTime.Now;
_TimeDifference = _TimeEnd.Subtract(_TimeStart); 
_TimeElasped = _TimeElasped.Add(_TimeDifference);
EventLog.WriteEntry(_serviceName + "was stopped at " + _TimeEnd.ToString() + "rn ran for total time :" + _TimeElasped.ToString());
}

// MQ connection service 
public void StartMQListenerService()
{
try
{
if (_run)
{
if (MQReader == null)
{
MQReader = new MQReader();
MQReader.InitializeConnections();
EventLog.WriteEntry(_serviceName + "MQ connection is established");
}
}
}
catch (Exception ex)
{
System.Diagnostics.EventLog.WriteEntry(_serviceName, ex.ToString());
System.Diagnostics.ProcessStartInfo startinfo = new System.Diagnostics.ProcessStartInfo();
startinfo.WindowStyle = System.Diagnostics.ProcessWindowStyle.Hidden;
startinfo.FileName = "NET";
startinfo.Arguments = "stop" + this.ServiceName;
Process.Start(startinfo);
}
}
}
}

****MQReader.cs****
using System;
using IBM.WMQ;
using System.Diagnostics;
using System.IO;
using System.Xml;
using System.Linq;
using System.Xml.Linq;
using System.Configuration;
namespace MQ_listner
{
internal class MQReader
{
public MQReader()
{
}
public void InitializeConnections()
{
MQQueueManager queueManager;
MQMessage queueMessage;
MQGetMessageOptions queueGetMessageOptions;
MQQueue queue;

string QueueName;
string QueueManagerName;
string ChannelInfo;
string channelName;
string PortNumber;
string transportType;
string connectionName;
QueueManagerName = ConfigurationManager.AppSettings["QueueManager"]; 
QueueName = ConfigurationManager.AppSettings["Queuename"];
ChannelInfo = ConfigurationManager.AppSettings["ChannelInformation"];
PortNumber = ConfigurationManager.AppSettings["Port"];
char[] separator = { '/' };
string[] ChannelParams;
ChannelParams = ChannelInfo.Split(separator);
channelName = ConfigurationManager.AppSettings["Channel"];
transportType = ConfigurationManager.AppSettings["TransportType"];
connectionName = ConfigurationManager.AppSettings["ConnectionName"];
String strReturn = "";
try
{
queueManager = new MQQueueManager(QueueManagerName,
channelName, connectionName);
strReturn = "Connected Successfully";
queue = queueManager.AccessQueue(QueueName,
MQC.MQOO_INPUT_AS_Q_DEF + MQC.MQOO_FAIL_IF_QUIESCING);
queueMessage = new MQMessage();
queueMessage.Format = MQC.MQFMT_STRING;
queueGetMessageOptions = new MQGetMessageOptions();
queue.Get(queueMessage, queueGetMessageOptions);
strReturn = queueMessage.ReadString(queueMessage.MessageLength);
}
catch (MQException exp)
{
strReturn = "Exception: " + exp.Message;
}
string path1 = @"C:documentsExample.txt";
System.IO.File.WriteAllText(path1, strReturn);
}
}
}

有人能告诉我我的代码出了什么问题吗？我是否需要在此处添加任何内容来不断地从远程MQ获取消息。请帮忙。任何链接或线索都可以。

编辑

经过一段时间后，我需要重新启动服务以从远程mq获取数据。你能告诉我为什么windows服务需要重新启动才能获取数据吗。有线索吗？知道吗？

Answer 1

您的队列关闭和队列管理器断开在哪里？如果连接和/或打开某个东西，必须确保关闭并断开连接。我强烈建议您参加MQ编程课程。或者参加MQ技术会议，该会议有关于编程MQ.的会议

我发布了一个功能齐全的C#程序，它在MQQueueManager消息池中检索队列上的所有消息

这里是MQReader类的更新版本，应该会给您正确的想法。注意：我没有测试它。我把它留给你。：(

此外，您应该将连接信息放在哈希表中，并将该哈希表传递给MQQueueManager类。

using System;
using IBM.WMQ;
using System.Diagnostics;
using System.IO;
using System.Xml;
using System.Linq;
using System.Xml.Linq;
using System.Configuration;
namespace MQ_listner
{
internal class MQReader
{
private MQQueueManager qManager = null;
private MQMessage      inQ = null;
private bool           running = true;
public MQReader()
{
}
public bool InitQMgrAndQueue()
{
bool flag = true;
Hashtable qMgrProp = new Hashtable();
qMgrProp.Add(MQC.TRANSPORT_PROPERTY, MQC.TRANSPORT_MQSERIES_MANAGED);
qMgrProp.Add(MQC.HOST_NAME_PROPERTY, ConfigurationManager.AppSettings["ConnectionName"]);
qMgrProp.Add(MQC.CHANNEL_PROPERTY, ConfigurationManager.AppSettings["Channel"]);
try
{
if (ConfigurationManager.AppSettings["Port"] != null)
qMgrProp.Add(MQC.PORT_PROPERTY, System.Int32.Parse(ConfigurationManager.AppSettings["Port"]));
else
qMgrProp.Add(MQC.PORT_PROPERTY, 1414);
}
catch (System.FormatException e)
{
qMgrProp.Add(MQC.PORT_PROPERTY, 1414);
}
if (ConfigurationManager.AppSettings["UserID"] != null)
qMgrProp.Add(MQC.USER_ID_PROPERTY, ConfigurationManager.AppSettings["UserID"]);
if (ConfigurationManager.AppSettings["Password"] != null)
qMgrProp.Add(MQC.PASSWORD_PROPERTY, ConfigurationManager.AppSettings["Password"]);
try
{
qManager = new MQQueueManager(ConfigurationManager.AppSettings["QueueManager"],
qMgrProp);
System.Console.Out.WriteLine("Connected Successfully");
inQ = qManager.AccessQueue(ConfigurationManager.AppSettings["Queuename"],
MQC.MQOO_INPUT_AS_Q_DEF + MQC.MQOO_FAIL_IF_QUIESCING);
System.Console.Out.WriteLine("Open queue Successfully");
}
catch (MQException exp)
{
System.Console.Out.WriteLine("MQException CC=" + mqex.CompletionCode + " : RC=" + mqex.ReasonCode);
flag = false;
}
return flag;
}
public void LoopThruMessages()
{
MQGetMessageOptions gmo = new MQGetMessageOptions();
gmo.Options |= MQC.MQGMO_WAIT | MQC.MQGMO_FAIL_IF_QUIESCING;
gmo.WaitInterval = 2500;  // 2.5 seconds wait time or use MQC.MQEI_UNLIMITED to wait forever
MQMessage msg = null;
while (running)
{
try
{
msg = new MQMessage();
inQ.Get(msg, gmo);
System.Console.Out.WriteLine("Message Data: " + msg.ReadString(msg.MessageLength));
}
catch (MQException mqex)
{
if (mqex.Reason == MQC.MQRC_NO_MSG_AVAILABLE)
{
// no meesage - life is good - loop again
}
else
{
running = false;  // severe error - time to exit
System.Console.Out.WriteLine("MQException CC=" + mqex.CompletionCode + " : RC=" + mqex.ReasonCode);
}
}
catch (System.IO.IOException ioex)
{
System.Console.Out.WriteLine("ioex=" + ioex);
}
}
try
{
if (inQ != null)
{
inQ.Close();
System.Console.Out.WriteLine("Closed queue");
}
}
catch (MQException mqex)
{
System.Console.Out.WriteLine("MQException CC=" + mqex.CompletionCode + " : RC=" + mqex.ReasonCode);
}
try
{
if (qMgr != null)
{
qMgr.Disconnect();
System.Console.Out.WriteLine("disconnected from queue manager");
}
}
catch (MQException mqex)
{
System.Console.Out.WriteLine("MQException CC=" + mqex.CompletionCode + " : RC=" + mqex.ReasonCode);
}
}
public void StopIt()
{
running = false;
}
}
}

无论何时停止服务，请确保它调用MQReader中的StopIt方法。