linux 查看ibm mq队列消息,.NET IBM MQ侦听器未确认的消息并从队列的开头读取

I have a C# application that sets up numerous MQ listeners (multiple threads and potentially multiple servers each with their own listeners). There are some messages that will come off the queue that I will want to leave on the queue, move on to the next message on the MQ, but then under some circumstances I will want to go back to re-read those messages...

var connectionFactory = XMSFactoryFactory.GetInstance(XMSC.CT_WMQ).CreateConnectionFactory();

connectionFactory.SetStringProperty(XMSC.WMQ_HOST_NAME, origination.Server);

connectionFactory.SetIntProperty(XMSC.WMQ_PORT, int.Parse(origination.Port));

connectionFactory.SetStringProperty(XMSC.WMQ_QUEUE_MANAGER, origination.QueueManager);

connectionFactory.SetStringProperty(XMSC.WMQ_CHANNEL, origination.Channel);

var connection = connectionFactory.CreateConnection(null, null);

_connections.Add(connection);

var session = connection.CreateSession(false, AcknowledgeMode.ClientAcknowledge); //changed to use ClientAcknowledge so that we will leave the message on the MQ until we're sure we're processing it

_sessions.Add(session);

var destination = session.CreateQueue(origination.Queue);

_destinations.Add(destination);

var consumer = session.CreateConsumer(destination);

_consumers.Add(consumer);

Logging.LogDebugMessage(Constants.ListenerStart);

connection.Start();

ThreadPool.QueueUserWorkItem((o) => Receive(forOrigination, consumer));

Then I have...

if (OnMQMessageReceived != null)

{

var message = consumer.Receive();

var identifier = string.Empty;

if (message is ITextMessage)

{

//do stuff with the message here

//populates identifier from the message

}

else

{

//do stuff with the message here

//populates identifier from the message

}

if (!string.IsNullOrWhiteSpace(identifier)&& OnMQMessageReceived != null)

{

if( some check to see if we should process the message now)

{

//process message here

message.Acknowledge(); //this really pulls it off of the MQ

//here is where I want to trigger the next read to be from the beginning of the MQ

}

else

{

//We actually want to do nothing here. As in do not do Acknowledge

//This leaves the message on the MQ and we'll pick it up again later

//But we want to move on to the next message in the MQ

}

}

else

{

message.Acknowledge(); //this really pulls it off of the MQ...its useless to us anyways

}

}

else

{

Thread.Sleep(0);

}

ThreadPool.QueueUserWorkItem((o) => Receive(forOrigination, consumer));

So a couple of questions:

If I do not acknowledge the message it stays on the MQ, right?

If the message is not acknowledged then by default when I read from the MQ again with the same listener it reads the next one and does not go to the beginning, right?

How do I change the listener so that the next time I read I start at the beginning of the queue?

Solutions1

Leaving messages on a queue is an anti-pattern. If you don't want to or cannot process the message at a certain point of your logic, then you have a number of choices:

Get it off the queue and put to another queue/topic for a delayed/different processing.

Get it off the queue and dump to a database, flat file - whatever, if you want to process it outside of messaging flow, or don't want to process at all.

If it is feasible, you may want to change the message producer so it doesn't mix the messages with different processing requirements in the same queue/topic.

In any case, do not leave a message on the queue, and always move forward to the next message. This will make the application way more predictable and easier to reason about. You will also avoid all kinds of performance problems. If your application is or may ever become sensitive to the sequence of message delivery, then manual acknowledgement of selected messages will be at odds with it too.

To your questions:

The JMS spec is vague regarding the behavior of unacknowledged messages - they may be delivered out of order, and it is undefined when exactly when they will be delivered. Also, the acknowledge method call will acknowledge all previously received and unacknowledged messages - probably not what you had in mind.

If you leave messages behind, the listener may or may not go back immediately. If you restart it, it of course will start afresh, but while it is sitting there waiting for messages it is implementation dependent.

So if you try to make your design work, you may get it kind of work under certain circumstances, but it will not be predictable or reliable.