使用ODP.NET和MSGID对Oracle AQ进行排队

Question

我正在尝试使用ODP.NET by service 从oracle AQ中取消消息队列

我有两个条件

1. 要出队的队列表中记录的MessageID将由提供服务
2. 一旦服务将记录不可用于其他服务实例的消息退出队列，意味着它应该被锁定，直到事务提交或事务过期

问题：

1. 取消排队选项MsgID无法识别我的MsgID
2. 找不到如何从同一消息ID的出队列中锁定其他实例

我的队列就像

BEGIN DBMS_AQADM.CREATE_QUEUE_TABLE(
Queue_table        => '"DEVENV1_CUSTOM"."INTEGRATION_MSG_QUEUE"',
Queue_payload_type => '"DEVENV1_CUSTOM"."TRIGGER_AUDIT_TYPE"',
Sort_list          => 'ENQ_TIME',
Compatible         => '10.0.0',
primary_instance   => 3,
secondary_instance => 2);
END;

有人能建议我一种方法来传递要从队列中退出的记录的消息ID吗？以及如何在提交或超时之前阻止其他人访问它

public static void DequeueMessage(string MessageID)
{
// Create connection
string constr = "my constring";
Oracle.DataAccess.Client.OracleConnection con = new Oracle.DataAccess.Client.OracleConnection(constr);
// Create queue
OracleAQQueue queue = new OracleAQQueue("DEVENV1_CUSTOM.INTEGRATION_MSG_QUEUE_Q", con);
try
{
// Open connection
con.Open();     
// Begin txn for enqueue
Oracle.DataAccess.Client.OracleTransaction txn = con.BeginTransaction();
// Prepare to Dequeue
queue.DequeueOptions.Visibility = OracleAQVisibilityMode.OnCommit;
queue.DequeueOptions.Wait = 10;
queue.DequeueOptions.DequeueMode = OracleAQDequeueMode.Locked;
// Set message type for the queue
queue.MessageType = OracleAQMessageType.Udt;
queue.UdtTypeName = "TRIGGER_AUDIT_TYPE";        
queue.DequeueOptions.MessageId = System.Text.Encoding.ASCII.GetBytes ("A0CB9F328889420D8F0C8A62BF38B454");
// Dequeue message
OracleAQMessage deqMsg = queue.Dequeue();
TRIGGER_AUDIT_TYPE data = (TRIGGER_AUDIT_TYPE)deqMsg.Payload;
// Dequeue txn commit
txn.Commit();
}
catch (Exception e)
{
Console.WriteLine("Error: {0}", e.Message);

}
finally
{
// Close/Dispose objects
con.Close();
con.Dispose();
}
}

Answer 1

问题是Oracle返回的字节数组和.net创建的字节数组不同(如果您感兴趣，请阅读有关小Endian和大Endian的信息(

所以我通过直接从oracle中铸造数组来解决这个问题，并将其用于Dequeue

OracleCommand oracleCommand = new OracleCommand("Select MSGID  from INTEGRATION_MSG_QUEUE where msgid='B85439241C9241A4A6B1869F08F382E7'", con);
var K=  oracleCommand.ExecuteScalar();
Byte[] rawBytesFromOracle =K as byte[];


// Begin txn for enqueue
Oracle.DataAccess.Client.OracleTransaction txn = con.BeginTransaction();
// Prepare to Dequeue
queue.DequeueOptions.Visibility = OracleAQVisibilityMode.OnCommit;
queue.DequeueOptions.Wait = 10;
queue.DequeueOptions.DequeueMode = OracleAQDequeueMode.Locked;
queue.DequeueOptions.MessageId = rawBytesFromOracle;

// Set message type for the queue
queue.MessageType = OracleAQMessageType.Udt;
queue.UdtTypeName = "TRIGGER_AUDIT_TYPE";

Console.WriteLine("DeQueueing ............ ");
// Dequeue message
OracleAQMessage deqMsg = queue.Dequeue();
TRIGGER_AUDIT_TYPE data = (TRIGGER_AUDIT_TYPE)deqMsg.Payload;
Console.WriteLine("Dequeued message...." + data.CMESSAGEBODY );