使用带有 MSGID 的 ODP.NET 使 Oracle AQ 出队

Posted 2023-02-22

【中文标题】使用带有 MSGID 的 ODP.NET 使 Oracle AQ 出队

【英文标题】：Dequeue Oracle AQ with ODP.NET With MSGID

【发布时间】：2020-08-11 07:50:30

【问题描述】：

我正在尝试通过服务使用 ODP.NET 将来自 oracle AQ 的消息出列

我有两个条件

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

问题：

DequeueOptions MsgID 无法识别我的 MSGID 找不到如何锁定其他实例，使其无法将相同的 msg 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;

有人可以建议我一种方法来传递记录的 msgid 以从队列中出列，以及如何锁定它们以防止其他人访问它直到提交或超时

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();
            

        

【问题讨论】：

我认为你做得对。 If LOCKED is specified, then the message is dequeued with a write lock that lasts for the duration of the transaction 不是这样吗？

@RobertoHernandez 是的，它已锁定，但如果我调用另一个服务实例，它也会将同一记录出列

在提交后也不会从 QueueTable 中删除

【参考方案1】：

问题是Oracle返回的字节数组和.net创建的字节数组是不同的（如果你感兴趣，请阅读little Endian和Big 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 );